Spaces:
Runtime error
Runtime error
| # -*- coding: utf-8 -*- | |
| """ | |
| AB 双语语音陪练 —— 单机 Gradio 应用 | |
| 交互设计（对应我们前面确认的方案）： | |
| A 说话(A语言语音) | |
| -> livetranslate: A原文(A语言文字) / A译文(B语言文字) / A译音(B语言语音，自动播放) | |
| -> 机器人吃 A译文 -> 生成 B原文(B语言文字) | |
| -> B原文 TTS 合成 -> 自动播放（外语，主线，可 🔁 重复播放，只缓存最近一句） | |
| -> B原文 文本翻译 -> B译文(A语言文字)，屏幕显示，不自动发声 | |
| -> 🌐 按钮：首次点击才合成 B译文 语音并播放，同一轮内再点直接用缓存 | |
| 屏幕四行：A原文 / A译文 / B原文 / B译文 | |
| 扬声器两路：A译音（自动播） / B原文语音（自动播，可重播）/ B译文语音（按需，可重播） | |
| 运行前： | |
| 1. pip install -r requirements.txt | |
| 2. export DASHSCOPE_API_KEY="sk-xxxx" | |
| 3. python app.py | |
| """ | |
| import tempfile | |
| import os | |
| import gradio as gr | |
| from config import LANGUAGE_NAME | |
| from bailian_client import ( | |
| LiveTranslateClient, | |
| chat_reply, | |
| translate_text, | |
| synthesize_speech, | |
| ) | |
| def _bytes_to_tempfile(audio_bytes: bytes, suffix: str = ".mp3") -> str: | |
| """ | |
| 把音频二进制写到一个新的临时文件里，返回路径。 | |
| 每次都用新文件名，是为了让 Gradio 的 Audio 组件在"内容没变但要求重播"时 | |
| 也能触发 autoplay（同一个文件路径浏览器可能不会重新播放）。 | |
| """ | |
| fd, path = tempfile.mkstemp(suffix=suffix) | |
| with os.fdopen(fd, "wb") as f: | |
| f.write(audio_bytes) | |
| return path | |
| def on_send(audio_filepath, a_lang, b_lang, history, b_audio_cache, b_translated_audio_cache): | |
| """ | |
| 主流程：A 发一句话 -> 走完整个 pipeline -> 更新四行文字 + 两路自动播放音频。 | |
| history: [{"role": "user"/"assistant", "content": "..."}] 都是 B 语言文本 | |
| b_audio_cache: 最近一句 B原文 的语音 bytes（供 🔁 重播用） | |
| b_translated_audio_cache: 最近一句 B译文 的语音 bytes（供 🌐 重播用，可能为 None 表示还没点过） | |
| """ | |
| if not audio_filepath: | |
| raise gr.Error("先录一句话再发送") | |
| # 1. A -> B 方向：实时翻译 | |
| lt_client = LiveTranslateClient(source_lang=a_lang, target_lang=b_lang) | |
| lt_result = lt_client.translate_utterance(audio_filepath) | |
| a_original = lt_result["source_text"] # A原文 | |
| a_translated = lt_result["target_text"] # A译文（B语言） | |
| a_audio_bytes = lt_result["target_audio"] # A译音（B语言语音，wav bytes） | |
| if not a_translated: | |
| raise gr.Error("没有识别到有效的语音内容，请重新录音") | |
| # 2. 聊天机器人：A译文 -> B原文 | |
| b_original = chat_reply(history, a_translated, target_lang=b_lang) | |
| new_history = history + [ | |
| {"role": "user", "content": a_translated}, | |
| {"role": "assistant", "content": b_original}, | |
| ] | |
| # 3. B原文 -> TTS（外语，自动播放，同时作为 🔁 重播的缓存） | |
| b_audio_bytes = synthesize_speech(b_original, lang=b_lang) | |
| # 4. B原文 -> 文本翻译 -> B译文（母语文字，只显示，不自动发声） | |
| b_translated = translate_text(b_original, source_lang=b_lang, target_lang=a_lang) | |
| # 新一轮开始，B译文语音缓存清空，等用户点 🌐 再现合成 | |
| new_b_translated_audio_cache = None | |
| a_audio_path = _bytes_to_tempfile(a_audio_bytes, suffix=".wav") if a_audio_bytes else None | |
| b_audio_path = _bytes_to_tempfile(b_audio_bytes, suffix=".mp3") | |
| return ( | |
| a_original, | |
| a_translated, | |
| b_original, | |
| b_translated, | |
| a_audio_path, # A译音，自动播放 | |
| b_audio_path, # B原文语音，自动播放 | |
| new_history, | |
| b_audio_bytes, # 存进 state，供 🔁 使用 | |
| new_b_translated_audio_cache, # 存进 state，供 🌐 使用（None=还没生成过） | |
| None, # 清空 🌐 播放框，避免残留上一轮的音频 | |
| ) | |
| def on_replay_foreign(b_audio_cache): | |
| """🔁 重播 B原文（外语）——直接用缓存，不重新调用 TTS。""" | |
| if not b_audio_cache: | |
| raise gr.Error("还没有可重播的内容，先发一句话") | |
| return _bytes_to_tempfile(b_audio_cache, suffix=".mp3") | |
| def on_replay_native(b_original_text, a_lang, b_lang, b_translated_audio_cache): | |
| """ | |
| 🌐 播放/重播 B译文（母语）。 | |
| 第一次点击：调用 TTS 合成并缓存；之后再点：直接用缓存，不重复计费。 | |
| """ | |
| if not b_original_text: | |
| raise gr.Error("还没有可播放的内容，先发一句话") | |
| if b_translated_audio_cache is None: | |
| b_translated_text = translate_text(b_original_text, source_lang=b_lang, target_lang=a_lang) | |
| audio_bytes = synthesize_speech(b_translated_text, lang=a_lang) | |
| else: | |
| audio_bytes = b_translated_audio_cache | |
| path = _bytes_to_tempfile(audio_bytes, suffix=".mp3") | |
| return path, audio_bytes # 第二个值写回 state 缓存 | |
| def build_app(): | |
| lang_choices = list(LANGUAGE_NAME.items()) # [(code, name), ...] | |
| lang_labels = [f"{name} ({code})" for code, name in lang_choices] | |
| code_by_label = {f"{name} ({code})": code for code, name in lang_choices} | |
| with gr.Blocks(title="AB 双语语音陪练") as demo: | |
| gr.Markdown("## AB 双语语音陪练\n对着麦克风说母语，机器人用你正在学的语言陪你聊天。") | |
| with gr.Row(): | |
| a_lang_dd = gr.Dropdown( | |
| lang_labels, value=lang_labels[0], label="A：你的母语" | |
| ) | |
| b_lang_dd = gr.Dropdown( | |
| lang_labels, value=lang_labels[1], label="B：机器人 / 练习目标语言" | |
| ) | |
| history_state = gr.State([]) | |
| b_audio_cache_state = gr.State(None) # B原文语音缓存（🔁 用） | |
| b_translated_audio_cache_state = gr.State(None) # B译文语音缓存（🌐 用） | |
| mic = gr.Audio(sources=["microphone"], type="filepath", label="按住说话（A）") | |
| send_btn = gr.Button("发送 🎤", variant="primary") | |
| with gr.Row(): | |
| with gr.Column(): | |
| a_original_box = gr.Textbox(label="A原文", interactive=False) | |
| a_translated_box = gr.Textbox(label="A译文", interactive=False) | |
| with gr.Column(): | |
| b_original_box = gr.Textbox(label="B原文", interactive=False) | |
| b_translated_box = gr.Textbox(label="B译文", interactive=False) | |
| with gr.Row(): | |
| replay_foreign_btn = gr.Button("🔁 重播机器人外语发音") | |
| replay_native_btn = gr.Button("🌐 听母语翻译") | |
| a_audio_player = gr.Audio(label="A译音（自动播放）", autoplay=True, visible=True) | |
| b_audio_player = gr.Audio(label="B原文语音 / 🔁重播", autoplay=True, visible=True) | |
| b_translated_audio_player = gr.Audio(label="B译文语音 / 🌐", autoplay=True, visible=True) | |
| def _send_wrapper(audio_filepath, a_label, b_label, history, b_cache, b_tr_cache): | |
| a_lang = code_by_label[a_label] | |
| b_lang = code_by_label[b_label] | |
| return on_send(audio_filepath, a_lang, b_lang, history, b_cache, b_tr_cache) | |
| send_btn.click( | |
| fn=_send_wrapper, | |
| inputs=[mic, a_lang_dd, b_lang_dd, history_state, b_audio_cache_state, b_translated_audio_cache_state], | |
| outputs=[ | |
| a_original_box, | |
| a_translated_box, | |
| b_original_box, | |
| b_translated_box, | |
| a_audio_player, | |
| b_audio_player, | |
| history_state, | |
| b_audio_cache_state, | |
| b_translated_audio_cache_state, | |
| b_translated_audio_player, | |
| ], | |
| ) | |
| replay_foreign_btn.click( | |
| fn=on_replay_foreign, | |
| inputs=[b_audio_cache_state], | |
| outputs=[b_audio_player], | |
| ) | |
| def _replay_native_wrapper(b_original_text, a_label, b_label, b_tr_cache): | |
| a_lang = code_by_label[a_label] | |
| b_lang = code_by_label[b_label] | |
| return on_replay_native(b_original_text, a_lang, b_lang, b_tr_cache) | |
| replay_native_btn.click( | |
| fn=_replay_native_wrapper, | |
| inputs=[b_original_box, a_lang_dd, b_lang_dd, b_translated_audio_cache_state], | |
| outputs=[b_translated_audio_player, b_translated_audio_cache_state], | |
| ) | |
| return demo | |
| if __name__ == "__main__": | |
| demo = build_app() | |
| demo.queue().launch() |