Update app.py

app.py

@@ -2,6 +2,7 @@
 import sys
 import asyncio
 import re
+import uuid
 import torch
 import soundfile as sf
 import librosa
@@ -20,7 +21,7 @@
 print(f"⚙️ 正在初始化极速传译引擎 (设备: {DEVICE})")
 
 # ---------------------------------------------------------------------------
-# 2. 全局预加载模型 (支持 中/英/法/日/俄)
+# 2. 全局预加载模型
 # ---------------------------------------------------------------------------
 print("📦 [1/2] 正在加载 SenseVoiceSmall 多语言语音识别模型...")
 asr_model = AutoModel(
@@ -45,7 +46,7 @@
 print("🎉 所有模型加载完成！")
 
 # ---------------------------------------------------------------------------
-# 3. 5大语种映射配置 (包含 ASR 代码、LLM 名称及 TTS 高清发音人)
+# 3. 5大语种映射配置 (发音人采用兼容性极佳的微软 Neural 音色)
 # ---------------------------------------------------------------------------
 LANG_CONFIG = {
     "中文": {"asr": "zh", "name": "Chinese", "tag": "中", "voice": "zh-CN-XiaoxiaoNeural"},
@@ -56,30 +57,49 @@
 }
 
 # ---------------------------------------------------------------------------
-# 4. 核心处理逻辑
+# 4. 辅助函数：安全异步运行 TTS
 # ---------------------------------------------------------------------------
-async def generate_tts(text, voice, output_path="output.mp3"):
-    """使用 Edge-TTS 生成发音"""
+async def _async_tts(text, voice, filepath):
+    communicate = edge_tts.Communicate(text, voice)
+    await communicate.save(filepath)
+
+def generate_tts_sync(text, voice):
+    """生成带随机 UUID 的唯一音频文件，防止缓存和文件占用冲突"""
     try:
-        communicate = edge_tts.Communicate(text, voice)
-        await communicate.save(output_path)
-        return output_path
+        filename = f"tts_{uuid.uuid4().hex[:8]}.mp3"
+        # 安全处理事件循环
+        try:
+            loop = asyncio.get_event_loop()
+        except RuntimeError:
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            
+        if loop.is_running():
+            # 如果在现有的 async 循环中运行
+            import nest_asyncio
+            nest_asyncio.apply()
+            loop.run_until_complete(_async_tts(text, voice, filename))
+        else:
+            loop.run_until_complete(_async_tts(text, voice, filename))
+
+        if os.path.exists(filename) and os.path.getsize(filename) > 0:
+            return filename
+        return None
     except Exception as e:
-        print(f"TTS 生成失败: {e}")
+        print(f"❌ TTS 合成失败: {e}", file=sys.stderr)
         return None
 
+# ---------------------------------------------------------------------------
+# 5. 核心业务逻辑
+# ---------------------------------------------------------------------------
 def on_audio_change(audio_path, src_lang_choice):
-    """
-    步骤 3：停止录音后自动触发
-    作用：识别音频并立刻在界面输出【源语言文字】
-    """
+    """步骤 3：停止录音后识别源语言"""
     if audio_path is None or not os.path.exists(str(audio_path)) or os.path.getsize(str(audio_path)) == 0:
         return ""
 
     src_cfg = LANG_CONFIG[src_lang_choice]
 
     try:
-        # 极速语音识别
         res = asr_model.generate(
             input=audio_path,
             language=src_cfg["asr"],
@@ -87,18 +107,14 @@ def on_audio_change(audio_path, src_lang_choice):
             batch_size_s=60
         )
         text = res[0]["text"].strip()
-        # 清理 SenseVoice 富文本标签 (如 <|NEUTRAL|>)
         text = re.sub(r"<\|.*?\|>", "", text).strip()
         return text
     except Exception as e:
-        print(f"ASR 识别错误: {e}")
+        print(f"ASR 错误: {e}")
         return "⚠️ 识别失败，请重新录制。"
 
 def on_translate_click(src_text, src_lang_choice, tgt_lang_choice, history):
-    """
-    步骤 4：点击【翻译】按钮触发
-    作用：将源语言文本翻译成目的语，并生成声音播放器
-    """
+    """步骤 4：点击翻译，生成目的语文字和音频"""
     history = history or []
 
     if not src_text or src_text.startswith("⚠️"):
@@ -108,7 +124,7 @@ def on_translate_click(src_text, src_lang_choice, tgt_lang_choice, history):
     tgt_cfg = LANG_CONFIG[tgt_lang_choice]
 
     try:
-        # 如果源语言与目标语言相同，跳过翻译
+        # 1. 文本翻译
         if src_lang_choice == tgt_lang_choice:
             trans_text = src_text
         else:
@@ -138,27 +154,25 @@ def on_translate_click(src_text, src_lang_choice, tgt_lang_choice, history):
             input_len = inputs["input_ids"].shape[1]
             trans_text = mt_tokenizer.decode(outputs[0][input_len:], skip_special_tokens=True).strip()
 
-        # 生成目标语言音频
-        audio_output_path = "tmp_reply.mp3"
-        audio_res = asyncio.run(generate_tts(trans_text, tgt_cfg["voice"], audio_output_path))
+        # 2. 语音合成 (生成唯一的 MP3 文件)
+        audio_file = generate_tts_sync(trans_text, tgt_cfg["voice"])
 
-        # 更新历史记录
+        # 3. 更新历史对话记录
         history.append({"role": "user", "content": f"[{src_cfg['tag']}] {src_text}"})
         history.append({"role": "assistant", "content": f"[{tgt_cfg['tag']}] {trans_text}"})
 
-        return history, audio_res
+        return history, audio_file
 
     except Exception as e:
         import traceback
         traceback.print_exc()
-        history.append({"role": "assistant", "content": f"⚠️ 翻译出错: {str(e)}"})
+        history.append({"role": "assistant", "content": f"⚠️ 翻译失败: {str(e)}"})
         return history, None
 
 # ---------------------------------------------------------------------------
-# 5. Gradio 界面设计 (标准印刷字体 + 三步工作流)
+# 6. Gradio 界面设计 (标准印刷字体 + 交互修复)
 # ---------------------------------------------------------------------------
 PRINT_STYLE_CSS = """
-/* 强制使用标准印刷字体体系 (宋体/衬线印刷体) */
 body, input, button, textarea, span, div, p, label {
     font-family: "Georgia", "Times New Roman", "Songti SC", "SimSun", "STSong", serif !important;
 }
@@ -170,10 +184,10 @@ def on_translate_click(src_text, src_lang_choice, tgt_lang_choice, history):
 with gr.Blocks(title="极速多语种同传", css=PRINT_STYLE_CSS, theme=gr.themes.Soft()) as demo:
     gr.Markdown(
         "## 🎙️ 多语种极速传译系统\n"
-        "支持 **中·英·法·日·俄** 5 大语种自由互译，采用标准印刷体界面。"
+        "支持 **中·英·法·日·俄** 5 大语种自由互译。"
     )
 
-    # 1. 选方向（两行单选）
+    # 1. 选方向
     with gr.Group():
         src_radio = gr.Radio(
             choices=["中文", "英文", "法语", "日语", "俄语"],
@@ -186,26 +200,27 @@ def on_translate_click(src_text, src_lang_choice, tgt_lang_choice, history):
             label="2- 选择目的语 (译为什么)"
         )
 
-    # 2. 录音与识别区
+    # 2. 工作区
     with gr.Row():
         with gr.Column(scale=1):
             audio_in = gr.Audio(
                 sources=["microphone"],
                 type="filepath",
-                label="2 & 3- 点击录音 / 停止录音 (显示声波)",
+                label="2 & 3- 点击录音 / 停止录音",
                 waveform_options={"waveform_progress_color": "#FF5500"}
             )
             src_text_box = gr.Textbox(
-                label="源语言文字 (停止录音后自动显示)", 
+                label="源语言文字 (自动生成)", 
                 interactive=False, 
-                placeholder="录音结束后在此处自动生成识别文本..."
+                placeholder="录音结束后在此处显示识别文本..."
             )
             translate_btn = gr.Button("4- 🚀 翻译", variant="primary", size="lg")
 
         with gr.Column(scale=1):
             audio_out = gr.Audio(
-                label="目的语声音 (可重放声波)", 
-                autoplay=True
+                label="目的语声音 (可播放声波)", 
+                autoplay=True,
+                type="filepath"
             )
             chatbot = gr.Chatbot(
                 label="💬 传译历史记录", 
@@ -214,16 +229,16 @@ def on_translate_click(src_text, src_lang_choice, tgt_lang_choice, history):
             )
 
     # ---------------------------------------------------------------------------
-    # 事件绑定 (完美流转：录音结束 -> 出识别字 -> 点翻译 -> 出目的字和声波)
+    # 事件流转
     # ---------------------------------------------------------------------------
-    # 录音停止/改变时：自动识别源语言文字
-    audio_in.change(
+    # 录音结束自动识别
+    audio_in.stop_recording(
         fn=on_audio_change,
         inputs=[audio_in, src_radio],
         outputs=[src_text_box]
     )
 
-    # 点击翻译按钮：翻译文本并生成声音
+    # 点击翻译按钮，生成声音并播放
     translate_btn.click(
         fn=on_translate_click,
         inputs=[src_text_box, src_radio, tgt_radio, chatbot],