Update app.py

app.py

@@ -1,99 +1,278 @@
-import os
 import gradio as gr
+import os
+import tempfile
 import requests
+import json
+import time
+import numpy as np
+import re
+from pathlib import Path
 
-API_KEY = os.environ.get("SILICONFLOW_API_KEY", "")
+# ========== 配置 ==========
+API_KEY = os.environ.get("SILICONFLOW_API_KEY")
+if not API_KEY:
+    raise ValueError("请在 Space 的 Settings -> Secrets 中设置 SILICONFLOW_API_KEY")
 
 ASR_URL = "https://api.siliconflow.cn/v1/audio/transcriptions"
 CHAT_URL = "https://api.siliconflow.cn/v1/chat/completions"
 ASR_MODEL = "FunAudioLLM/SenseVoiceSmall"
 MT_MODEL = "tencent/Hunyuan-MT-7B"
 
-LANG_MAP = {
-    "英语 → 中文": {"src": "English", "tgt": "Chinese", "src_tag": "EN", "tgt_tag": "中"},
-    "汉语 → 英文": {"src": "Chinese", "tgt": "English", "src_tag": "中", "tgt_tag": "EN"},
-}
+# VAD 参数
+SILENCE_THRESHOLD = 0.015
+SILENCE_DURATION = 0.8
+MIN_SPEECH_DURATION = 0.3
 
+# ========== 全局状态 ==========
+class AppState:
+    def __init__(self):
+        self.mode = "en-zh"
+        self.history = []
+        self.is_speaking = False
+        self.audio_buffer = []
+        self.silence_start = None
+        self.speech_start = None
+        self.last_audio_time = 0
 
-def transcribe(audio_path):
+state = AppState()
+
+# ========== API 调用 ==========
+def transcribe_audio(audio_path):
     with open(audio_path, "rb") as f:
-        files = {"file": (os.path.basename(audio_path), f, "audio/wav")}
+        files = {"file": f}
         data = {"model": ASR_MODEL}
-        headers = {"Authorization": "Bearer " + API_KEY}
-        r = requests.post(ASR_URL, headers=headers, data=data, files=files, timeout=60)
-    r.raise_for_status()
-    text = r.json().get("text", "").strip()
-    text = text.replace("<|", "").replace("|>", "")
-    return text
-
+        headers = {"Authorization": f"Bearer {API_KEY}"}
+        try:
+            resp = requests.post(ASR_URL, files=files, data=data, headers=headers, timeout=30)
+            resp.raise_for_status()
+            result = resp.json()
+            text = result.get("text", "").strip()
+            text = re.sub(r"<\|[^|]*\|>", "", text).strip()
+            return text if text else None
+        except Exception as e:
+            print(f"ASR 错误: {e}")
+            return None
 
-def translate(text, src, tgt):
-    headers = {"Authorization": "Bearer " + API_KEY, "Content-Type": "application/json"}
+def translate_text(text, src_lang, tgt_lang):
+    system_prompt = f"Translate from {src_lang} to {tgt_lang}. Output ONLY translation, no extra text."
     payload = {
         "model": MT_MODEL,
         "messages": [
-            {
-                "role": "system",
-                "content": f"You are a professional translator. Translate the user text from {src} to {tgt}. "
-                           f"Output ONLY the translated text, no explanations, no quotes, no pinyin.",
-            },
-            {"role": "user", "content": text},
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": text}
         ],
         "temperature": 0.3,
-        "max_tokens": 512,
+        "max_tokens": 512
     }
-    r = requests.post(CHAT_URL, headers=headers, json=payload, timeout=60)
-    r.raise_for_status()
-    return r.json()["choices"][0]["message"]["content"].strip()
+    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
+    try:
+        resp = requests.post(CHAT_URL, json=payload, headers=headers, timeout=30)
+        resp.raise_for_status()
+        return resp.json()["choices"][0]["message"]["content"].strip()
+    except Exception as e:
+        print(f"翻译错误: {e}")
+        return None
 
+# ========== 音频处理 ==========
+def process_audio_chunk(audio_data, sr):
+    """处理音频块，返回 (是否触发翻译, 原文, 译文)"""
+    if len(audio_data.shape) > 1:
+        audio_data = audio_data.mean(axis=1)
+    
+    rms = np.sqrt(np.mean(audio_data**2))
+    is_speaking = rms > SILENCE_THRESHOLD
+    now = time.time()
 
-def process(audio_path, mode_label, history):
-    history = history or []
+    if is_speaking:
+        if not state.is_speaking:
+            state.is_speaking = True
+            state.speech_start = now
+            state.audio_buffer = [audio_data]
+            state.silence_start = None
+        else:
+            state.audio_buffer.append(audio_data)
+            state.silence_start = None
+        state.last_audio_time = now
+        return None, None, None
 
-    if not API_KEY:
-        history.append({"role": "assistant", "content": "⚠️ 服务器未配置 SILICONFLOW_API_KEY"})
-        return history, None
+    else:
+        if state.is_speaking:
+            if state.silence_start is None:
+                state.silence_start = now
+            elif now - state.silence_start > SILENCE_DURATION:
+                duration = now - state.speech_start
+                if duration >= MIN_SPEECH_DURATION and len(state.audio_buffer) > 0:
+                    full_audio = np.concatenate(state.audio_buffer)
+                    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
+                        tmp_path = tmp.name
+                    import soundfile as sf
+                    sf.write(tmp_path, full_audio, sr)
+                    
+                    src_text = transcribe_audio(tmp_path)
+                    os.unlink(tmp_path)
+                    
+                    if src_text:
+                        if state.mode == "en-zh":
+                            src_lang, tgt_lang = "English", "Chinese"
+                        else:
+                            src_lang, tgt_lang = "Chinese", "English"
+                        translated = translate_text(src_text, src_lang, tgt_lang)
+                        if translated:
+                            state.history.append((src_text, translated))
+                            state.is_speaking = False
+                            state.audio_buffer = []
+                            state.silence_start = None
+                            return True, src_text, translated
+                
+                state.is_speaking = False
+                state.audio_buffer = []
+                state.silence_start = None
+    
+    return None, None, None
 
-    if audio_path is None:
-        return history, None
+# ========== Gradio 回调 ==========
+def on_audio_change(audio):
+    if audio is None:
+        return gr.update(), gr.update(), gr.update(), gr.update()
+    
+    sr, audio_data = audio
+    if audio_data.dtype != np.float32:
+        audio_data = audio_data.astype(np.float32) / 32768.0
+    
+    triggered, src, tgt = process_audio_chunk(audio_data, sr)
+    
+    # 构建历史记录 HTML
+    history_html = ""
+    if state.history:
+        for s, t in reversed(state.history):
+            tag_src = "EN" if state.mode == "en-zh" else "中"
+            tag_tgt = "中" if state.mode == "en-zh" else "EN"
+            history_html += f"""
+            <div style='background:#1C1915;border-left:3px solid #FFB000;border-radius:4px 10px 10px 4px;padding:12px 14px;margin-bottom:10px;'>
+                <div style='font-family:monospace;font-size:14px;color:#9A9488;margin-bottom:4px;'>
+                    <span style='color:#8A6112;margin-right:6px;'>{tag_src}</span>{s}
+                </div>
+                <div style='font-size:16px;font-weight:600;color:#5FA8AC;'>
+                    <span style='font-family:monospace;font-weight:400;font-size:12px;color:#2E5254;margin-right:6px;'>{tag_tgt}</span>{t}
+                </div>
+            </div>
+            """
+    else:
+        history_html = "<div style='color:#9A9488;text-align:center;padding:30px;'>还没有记录，开口试试看</div>"
+    
+    status = "🔴 正在说话..." if state.is_speaking else "⚪ 待命"
+    
+    if triggered and src and tgt:
+        return gr.update(value=f"✅ {src} → {tgt}"), gr.update(value=history_html), gr.update(value=status), gr.update()
+    else:
+        return gr.update(), gr.update(value=history_html), gr.update(value=status), gr.update()
 
-    cfg = LANG_MAP[mode_label]
+def change_mode(mode):
+    state.mode = mode
+    state.history = []  # 切换模式时清空历史
+    return gr.update(value="")
 
-    try:
-        text = transcribe(audio_path)
-        if not text:
-            return history, None
-        trans = translate(text, cfg["src"], cfg["tgt"])
-    except Exception as e:
-        history.append({"role": "assistant", "content": f"⚠️ 出错了：{e}"})
-        return history, None
+def get_history():
+    history_html = ""
+    if state.history:
+        for s, t in reversed(state.history):
+            tag_src = "EN" if state.mode == "en-zh" else "中"
+            tag_tgt = "中" if state.mode == "en-zh" else "EN"
+            history_html += f"""
+            <div style='background:#1C1915;border-left:3px solid #FFB000;border-radius:4px 10px 10px 4px;padding:12px 14px;margin-bottom:10px;'>
+                <div style='font-family:monospace;font-size:14px;color:#9A9488;margin-bottom:4px;'>
+                    <span style='color:#8A6112;margin-right:6px;'>{tag_src}</span>{s}
+                </div>
+                <div style='font-size:16px;font-weight:600;color:#5FA8AC;'>
+                    <span style='font-family:monospace;font-weight:400;font-size:12px;color:#2E5254;margin-right:6px;'>{tag_tgt}</span>{t}
+                </div>
+            </div>
+            """
+    else:
+        history_html = "<div style='color:#9A9488;text-align:center;padding:30px;'>还没有记录，开口试试看</div>"
+    return gr.update(value=history_html)
 
-    history.append({"role": "user", "content": f"[{cfg['src_tag']}] {text}"})
-    history.append({"role": "assistant", "content": f"[{cfg['tgt_tag']}] {trans}"})
-    return history, None
+# ========== 构建界面 ==========
+css = """
+.gradio-container { max-width: 640px; margin: 0 auto; }
+.header { font-family: 'Manrope', sans-serif; text-align: center; margin-bottom: 20px; }
+.header h1 { font-size: 32px; font-weight: 800; letter-spacing: -0.01em; margin: 0; }
+.header .sub { color: #9A9488; font-size: 14px; }
+.history-box { background: #14120F; border: 1px solid #33302A; border-radius: 14px; padding: 16px; min-height: 150px; max-height: 350px; overflow-y: auto; }
+"""
 
+with gr.Blocks(css=css, title="传声筒 · 实时语音传译") as demo:
+    gr.HTML("""
+    <div class="header">
+        <div style="font-family:'IBM Plex Mono',monospace;font-size:11px;letter-spacing:0.18em;color:#8A6112;text-transform:uppercase;">REAL-TIME VOICE RELAY · 硅基流动版</div>
+        <h1>传声筒</h1>
+        <div class="sub">开口即译 — 说一句，出一句</div>
+    </div>
+    """)
 
-with gr.Blocks(title="传声筒") as demo:
-    gr.Markdown("## 🎙️ 传声筒 · 实时语音传译\n点麦克风开始录音，说一句完整的话，再点一次停止，自动识别+翻译。")
+    with gr.Row():
+        mode_radio = gr.Radio(
+            choices=[("🇬🇧 英语 → 中文 🇨🇳", "en-zh"), ("🇨🇳 中文 → 英语 🇬🇧", "zh-en")],
+            value="en-zh",
+            label="翻译方向",
+            interactive=True
+        )
 
-    mode = gr.Radio(
-        choices=list(LANG_MAP.keys()),
-        value="英语 → 中文",
-        label="翻译方向",
+    with gr.Row():
+        status_display = gr.Textbox(value="⚪ 待命", label="状态", interactive=False)
+
+    with gr.Row():
+        audio_input = gr.Audio(
+            sources=["microphone"],
+            type="numpy",
+            label="🎙️ 点击麦克风开始说话",
+            streaming=True
+        )
+
+    result_display = gr.Textbox(label="实时识别结果", interactive=False)
+
+    with gr.Column():
+        gr.Markdown("### 📜 传译记录")
+        history_display = gr.HTML(value="<div style='color:#9A9488;text-align:center;padding:30px;'>还没有记录，开口试试看</div>", elem_classes="history-box")
+
+    with gr.Row():
+        refresh_btn = gr.Button("🔄 刷新历史")
+        clear_btn = gr.Button("🗑️ 清空历史")
+
+    # 事件绑定
+    audio_input.change(
+        fn=on_audio_change,
+        inputs=audio_input,
+        outputs=[result_display, history_display, status_display, audio_input]
     )
 
-    audio_in = gr.Audio(
-        sources=["microphone"],
-        type="filepath",
-        label="点击录音，再点一次停止",
+    mode_radio.change(
+        fn=change_mode,
+        inputs=mode_radio,
+        outputs=result_display
     )
 
-    chatbot = gr.Chatbot(label="传译记录", height=420, type="messages")
+    refresh_btn.click(
+        fn=get_history,
+        outputs=history_display
+    )
 
-    audio_in.stop_recording(
-        fn=process,
-        inputs=[audio_in, mode, chatbot],
-        outputs=[chatbot, audio_in],
+    def clear_history():
+        state.history = []
+        return gr.update(value="<div style='color:#9A9488;text-align:center;padding:30px;'>还没有记录，开口试试看</div>")
+    
+    clear_btn.click(
+        fn=clear_history,
+        outputs=history_display
     )
 
-demo.launch()
+    gr.Markdown("""
+    ---
+    **提示**：
+    - 点击麦克风开始录音，说完自动识别翻译
+    - 使用硅基流动 API（SenseVoiceSmall + Hunyuan-MT-7B）
+    - API Key 从环境变量 `SILICONFLOW_API_KEY` 读取
+    """)
+
+if __name__ == "__main__":
+    demo.launch(debug=True)