import os import uuid import subprocess import gradio as gr import requests from faster_whisper import WhisperModel API_KEY = os.environ.get("SILICONFLOW_API_KEY", "") ASR_URL = "https://api.siliconflow.cn/v1/audio/transcriptions" CHAT_URL = "https://api.siliconflow.cn/v1/chat/completions" TTS_URL = "https://api.siliconflow.cn/v1/audio/speech" ASR_MODEL = "FunAudioLLM/SenseVoiceSmall" MT_MODEL = "tencent/Hunyuan-MT-7B" TTS_MODEL = "FunAudioLLM/CosyVoice2-0.5B" # 音频文件统一存到这个目录，用唯一文件名，实现"可重复回放历史" AUDIO_DIR = "/tmp/chuanshengtong_audio" os.makedirs(AUDIO_DIR, exist_ok=True) # 只有法语识别本地跑，其他都走硅基流动，兼顾速度和成本 FRENCH_ASR_MODEL_SIZE = "medium" french_asr_model = WhisperModel(FRENCH_ASR_MODEL_SIZE, device="cpu", compute_type="int8") # --------------------------------------------------------------------------- # Piper 本地法语 TTS：硅基流动的 CosyVoice2 不支持法语合成，所以法语这一路 # 用本地 Piper 兜底。HF Space 容器每次重建都是空的，所以在启动时自动下载模型， # 不需要手动登录服务器操作。 # --------------------------------------------------------------------------- PIPER_BIN = "piper" # pip install piper-tts 后，命令行工具会自动加入 PATH PIPER_MODEL_DIR = "piper_models" os.makedirs(PIPER_MODEL_DIR, exist_ok=True) PIPER_MODEL_FR = os.path.join(PIPER_MODEL_DIR, "fr_FR-siwis-medium.onnx") PIPER_MODEL_FR_JSON = os.path.join(PIPER_MODEL_DIR, "fr_FR-siwis-medium.onnx.json") PIPER_MODEL_URL = "https://huggingface.co/rhasspy/piper-voices/resolve/main/fr/fr_FR/siwis/medium/fr_FR-siwis-medium.onnx" PIPER_MODEL_JSON_URL = "https://huggingface.co/rhasspy/piper-voices/resolve/main/fr/fr_FR/siwis/medium/fr_FR-siwis-medium.onnx.json" def ensure_piper_model(): """启动时检查法语模型文件是否存在，不存在就自动下载。""" try: if not os.path.exists(PIPER_MODEL_FR): print("正在下载 Piper 法语模型（onnx）...") r = requests.get(PIPER_MODEL_URL, timeout=120) r.raise_for_status() with open(PIPER_MODEL_FR, "wb") as f: f.write(r.content) if not os.path.exists(PIPER_MODEL_FR_JSON): print("正在下载 Piper 法语模型配置（json）...") r = requests.get(PIPER_MODEL_JSON_URL, timeout=60) r.raise_for_status() with open(PIPER_MODEL_FR_JSON, "wb") as f: f.write(r.content) print("Piper 法语模型已就绪") except Exception as e: print(f"⚠️ Piper 法语模型下载失败：{e}（法语合成方向届时会报错，其他方向不受影响）") ensure_piper_model() # --------------------------------------------------------------------------- # 翻译方向配置 # asr : 识别引擎 "siliconflow" | "local_french" # tts : 合成引擎 "siliconflow" | "piper_fr" # voice: 硅基流动 CosyVoice2 的预置音色（走 piper_fr 时无用，填 None） # --------------------------------------------------------------------------- LANG_MAP = { "英语 → 中文": {"src": "English", "tgt": "Chinese", "src_tag": "EN", "tgt_tag": "中", "asr": "siliconflow", "tts": "siliconflow", "voice": "FunAudioLLM/CosyVoice2-0.5B:alex"}, "汉语 → 英文": {"src": "Chinese", "tgt": "English", "src_tag": "中", "tgt_tag": "EN", "asr": "siliconflow", "tts": "siliconflow", "voice": "FunAudioLLM/CosyVoice2-0.5B:david"}, "汉语 → 法文": {"src": "Chinese", "tgt": "French", "src_tag": "中", "tgt_tag": "FR", "asr": "siliconflow", "tts": "piper_fr", "voice": None}, "法语 → 中文": {"src": "French", "tgt": "Chinese", "src_tag": "FR", "tgt_tag": "中", "asr": "local_french", "tts": "siliconflow", "voice": "FunAudioLLM/CosyVoice2-0.5B:alex"}, "英语 → 法文": {"src": "English", "tgt": "French", "src_tag": "EN", "tgt_tag": "FR", "asr": "siliconflow", "tts": "piper_fr", "voice": None}, "法语 → 英文": {"src": "French", "tgt": "English", "src_tag": "FR", "tgt_tag": "EN", "asr": "local_french", "tts": "siliconflow", "voice": "FunAudioLLM/CosyVoice2-0.5B:david"}, } def transcribe_siliconflow(audio_path): with open(audio_path, "rb") as f: files = {"file": (os.path.basename(audio_path), f, "audio/wav")} 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 def transcribe_french_local(audio_path): segments, _info = french_asr_model.transcribe( audio_path, language="fr", task="transcribe", vad_filter=True, ) return "".join(seg.text for seg in segments).strip() def transcribe(audio_path, engine): if engine == "local_french": return transcribe_french_local(audio_path) return transcribe_siliconflow(audio_path) def translate(text, src, tgt): headers = {"Authorization": "Bearer " + API_KEY, "Content-Type": "application/json"} 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}, ], "temperature": 0.3, "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() def speak_siliconflow(text, voice, out_path): """硅基流动 CosyVoice2 合成，输出 mp3""" headers = {"Authorization": "Bearer " + API_KEY, "Content-Type": "application/json"} payload = { "model": TTS_MODEL, "input": text, "voice": voice, "response_format": "mp3", } r = requests.post(TTS_URL, headers=headers, json=payload, timeout=60) r.raise_for_status() with open(out_path, "wb") as f: f.write(r.content) return out_path def speak_piper_french(text, out_path): """本地 Piper 合成法语语音，输出 wav""" if not os.path.exists(PIPER_MODEL_FR): raise RuntimeError(f"未找到 Piper 法语模型文件：{PIPER_MODEL_FR}，请检查启动日志中的下载是否成功") cmd = [PIPER_BIN, "--model", PIPER_MODEL_FR, "--output_file", out_path] proc = subprocess.run(cmd, input=text.encode("utf-8"), capture_output=True) if proc.returncode != 0: err = proc.stderr.decode("utf-8", errors="ignore") raise RuntimeError(f"Piper 合成失败：{err}") return out_path def speak(text, cfg): """按翻译方向配置，选择硅基流动或本地 Piper 合成语音；文件名唯一，支持历史回放""" uid = uuid.uuid4().hex[:8] if cfg["tts"] == "piper_fr": out_path = os.path.join(AUDIO_DIR, f"{uid}.wav") return speak_piper_french(text, out_path) else: out_path = os.path.join(AUDIO_DIR, f"{uid}.mp3") return speak_siliconflow(text, cfg["voice"], out_path) def process(audio_path, mode_label, history): history = history or [] if not API_KEY: history.append({"role": "assistant", "content": "⚠️ 服务器未配置 SILICONFLOW_API_KEY，请检查 Space 的 Settings -> Repository secrets"}) return history, None, None if audio_path is None: return history, None, None cfg = LANG_MAP[mode_label] try: text = transcribe(audio_path, cfg["asr"]) if not text: return history, None, None trans = translate(text, cfg["src"], cfg["tgt"]) tts_path = speak(trans, cfg) except Exception as e: history.append({"role": "assistant", "content": f"⚠️ 出错了：{e}"}) return history, None, None # 文字记录 history.append({"role": "user", "content": f"[{cfg['src_tag']}] {text}"}) history.append({"role": "assistant", "content": f"[{cfg['tgt_tag']}] {trans}"}) # 把这句译文的音频也塞进聊天记录里，实现"点开历史消息随时重听" # 注意：新版 Gradio Chatbot 用 messages 格式，文件消息要写成 {"path": ...} 字典， # 而不是旧版 tuples 格式的 (path,) 元组，否则会报 Invalid message 错误 history.append({"role": "assistant", "content": {"path": tts_path}}) return history, None, tts_path with gr.Blocks(title="传声筒") as demo: gr.Markdown( "## 🎙️ 传声筒 · 实时语音传译\n" "点麦克风开始录音，说一句完整的话，再点一次停止，自动识别+翻译+朗读译文。\n" "识别与翻译主要使用硅基流动，法语识别、法语合成使用本地方案补充。\n" "每句译文的语音会自动播放一遍，也会留在下方记录里，随时可以点开重听。" ) mode = gr.Radio( choices=list(LANG_MAP.keys()), value="英语 → 中文", label="翻译方向", ) audio_in = gr.Audio( sources=["microphone"], type="filepath", label="点击录音，再点一次停止", ) chatbot = gr.Chatbot(label="传译记录（含可重听音频）", height=460) tts_out = gr.Audio( label="最新译文语音（当前这一句，可反复点播放）", autoplay=True, interactive=False, ) audio_in.stop_recording( fn=process, inputs=[audio_in, mode, chatbot], outputs=[chatbot, audio_in, tts_out], ) demo.launch()