""" Trading Studio — 自动化交易复盘视频配音系统 Gradio Web 中控:音色锁定 → Whisper 识别 → Gemma4 润色 → ChatTTS 合成 """ from __future__ import annotations import logging import shutil import sys import uuid from pathlib import Path import gradio as gr from config import ( GIT_REPO_URL, HOST, MODEL_NAME, OLLAMA_URL, PORT, SPEAKER_EMB_PATH, UPLOAD_DIR, ) from llm_service import check_ollama_health, polish_text from tts_service import generate_voice, save_fixed_speaker, speaker_is_ready from whisper_service import transcribe_audio # --------------------------------------------------------------------------- # 日志 # --------------------------------------------------------------------------- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[ logging.StreamHandler(sys.stdout), logging.FileHandler("trading_studio.log", encoding="utf-8"), ], ) logger = logging.getLogger("trading_studio") # --------------------------------------------------------------------------- # 全局 UI 状态(Gradio State) # --------------------------------------------------------------------------- # raw_transcript / polished_script 在流水线中传递 def _save_upload(upload_file) -> str | None: """将 Gradio 上传文件复制到本地 uploads 目录,返回持久化路径。""" if upload_file is None: return None src = Path(upload_file) if not src.exists(): return None dest = UPLOAD_DIR / f"{uuid.uuid4().hex}_{src.name}" shutil.copy2(src, dest) return str(dest) # --------------------------------------------------------------------------- # 模块 1:音色锁定 # --------------------------------------------------------------------------- def ui_lock_speaker(audio_file, sample_transcript: str) -> tuple[str, str]: """【音色锁定】从参考人声提取并保存 Speaker Embedding。""" path = _save_upload(audio_file) if not path: return "请上传 10-30 秒干净参考人声(wav/mp3 均可)。", ui_speaker_status_html() ok, msg = save_fixed_speaker(path, sample_transcript or "") result = msg if ok else f"❌ {msg}" return result, ui_speaker_status_html() def ui_speaker_status() -> str: """刷新音色状态(纯文本,供日志框使用)。""" ok, msg = speaker_is_ready() return f"✅ {msg}" if ok else f"⚠️ {msg}" # --------------------------------------------------------------------------- # 模块 2:音频极速识别 # --------------------------------------------------------------------------- def ui_transcribe(audio_file) -> tuple[str, str]: """【Whisper 识别】返回 (转写文本, 状态日志)。""" path = _save_upload(audio_file) if not path: return "", "请上传待识别的碎碎念录音。" ok, result = transcribe_audio(path) if ok: return result, f"✅ 识别完成,共 {len(result)} 字。" return "", f"❌ {result}" # --------------------------------------------------------------------------- # 模块 3:Gemma4 纪律审判 # --------------------------------------------------------------------------- def ui_polish(raw_text: str) -> tuple[str, str]: """【LLM 润色】对转写稿进行严厉自我反思式润色。""" if not raw_text or not raw_text.strip(): return "", "请先完成语音识别或手动粘贴转写文本。" ok, result = polish_text(raw_text) if ok: return result, f"✅ Gemma4 润色完成,共 {len(result)} 字。" return "", f"❌ {result}" def ui_check_ollama() -> str: """检测远程 Ollama 节点状态。""" ok, msg = check_ollama_health() return f"✅ {msg}" if ok else f"❌ {msg}" # --------------------------------------------------------------------------- # 模块 4:ChatTTS 音频合成 # --------------------------------------------------------------------------- def ui_synthesize(polished_text: str) -> tuple[str | None, str]: """【TTS 合成】生成最终 wav 配音文件。""" if not polished_text or not polished_text.strip(): return None, "请先完成 Gemma4 润色。" ok, msg, wav_path = generate_voice(polished_text) if ok and wav_path: return wav_path, f"✅ {msg}" return None, f"❌ {msg}" # --------------------------------------------------------------------------- # 一键流水线 # --------------------------------------------------------------------------- def ui_full_pipeline( audio_file, skip_polish: bool, manual_raw: str, ) -> tuple[str, str, str | None, str]: """ 串联执行:识别 → 润色(可跳过)→ 合成。 返回 (raw, polished, wav_path, log) """ logs: list[str] = [] # Step 1: 识别 if manual_raw and manual_raw.strip(): raw = manual_raw.strip() logs.append(f"使用手动输入转写稿({len(raw)} 字)。") else: path = _save_upload(audio_file) if not path: return "", "", None, "❌ 请上传录音或手动填写转写文本。" ok, result = transcribe_audio(path) if not ok: return "", "", None, f"❌ 识别失败: {result}" raw = result logs.append(f"✅ Whisper 识别完成({len(raw)} 字)。") # Step 2: 润色 if skip_polish: polished = raw logs.append("已跳过 LLM 润色,直接使用原文。") else: ok, result = polish_text(raw) if not ok: return raw, "", None, f"❌ 润色失败: {result}\n" + "\n".join(logs) polished = result logs.append(f"✅ Gemma4 润色完成({len(polished)} 字)。") # Step 3: 合成 ok, msg, wav_path = generate_voice(polished) if not ok: return raw, polished, None, f"❌ 合成失败: {msg}\n" + "\n".join(logs) logs.append(f"✅ {msg}") return raw, polished, wav_path, "\n".join(logs) # --------------------------------------------------------------------------- # Gradio 界面 # --------------------------------------------------------------------------- APP_ROOT = Path(__file__).resolve().parent PWA_DIR = APP_ROOT / "pwa" PWA_HEAD = """ """ INSTALL_APP_BUTTON_HTML = """ """ CUSTOM_CSS = """ /* ========== 居中布局 + 响应式 + 高对比度 ========== */ html, body { width: 100% !important; min-height: 100vh !important; margin: 0 !important; padding: 0 !important; background: #0f1419 !important; overflow-x: hidden !important; } /* 外层全宽背景,内容区水平居中 */ gradio-app, .gradio-app, #main, .app, .fillable, .contain { width: 100% !important; max-width: 100% !important; display: flex !important; flex-direction: column !important; align-items: center !important; justify-content: flex-start !important; background: #0f1419 !important; box-sizing: border-box !important; } /* 核心内容容器居中 */ .gradio-container { background: #0f1419 !important; color: #eef2f7 !important; font-size: 15px !important; width: 100% !important; max-width: min(1200px, 94vw) !important; margin-left: auto !important; margin-right: auto !important; padding: 16px clamp(12px, 3vw, 32px) !important; box-sizing: border-box !important; float: none !important; } /* 带鱼屏 / 超宽屏 */ @media (min-width: 1920px) { .gradio-container { max-width: min(1280px, 82vw) !important; } } @media (min-width: 2560px) { .gradio-container { max-width: min(1360px, 68vw) !important; } } @media (min-width: 3440px) { .gradio-container { max-width: 1440px !important; } } /* 平板 */ @media (max-width: 1024px) { .gradio-container { max-width: 100% !important; padding: 14px 18px !important; } } /* 手机 */ @media (max-width: 640px) { .gradio-container { max-width: 100% !important; padding: 10px 12px !important; font-size: 14px !important; } .gradio-container h1 { font-size: 1.35rem !important; } .gradio-container .tab-nav { flex-wrap: wrap !important; gap: 6px !important; } .gradio-container .tab-nav button { flex: 1 1 28% !important; min-width: 88px !important; padding: 8px 8px !important; font-size: 0.8rem !important; } .status-row, .status-row > div, .pipeline-steps, .pipeline-steps > div, .pipeline-output-row, .pipeline-output-row > div { flex-direction: column !important; width: 100% !important; } .status-row button, .pipeline-steps button, .pipeline-output-row button { width: 100% !important; } .dark-panel table { font-size: 0.85rem !important; display: block !important; overflow-x: auto !important; } .install-hint { font-size: 0.85rem !important; } } /* 触摸设备:按钮最小点击区域 44px */ @media (hover: none) and (pointer: coarse) { .gradio-container button { min-height: 44px !important; } } /* 全局文字 */ .gradio-container p, .gradio-container span, .gradio-container label, .gradio-container .prose, .gradio-container .markdown-text, .gradio-container .md { color: #eef2f7 !important; } .gradio-container h1 { color: #ffffff !important; font-size: 1.75rem !important; font-weight: 700 !important; } .gradio-container h2, .gradio-container h3 { color: #dbeafe !important; font-size: 1.15rem !important; font-weight: 600 !important; } .gradio-container .block-label, .gradio-container label, .gradio-container .label-wrap span { color: #93c5fd !important; font-weight: 600 !important; font-size: 0.95rem !important; } .gradio-container textarea, .gradio-container input[type="text"], .gradio-container .wrap textarea, .gradio-container .wrap input { color: #ffffff !important; background: #1a2332 !important; border: 1px solid #4b5563 !important; font-size: 0.95rem !important; line-height: 1.6 !important; } /* placeholder 高对比度 */ .gradio-container textarea::placeholder, .gradio-container input::placeholder, .gradio-container input[type="text"]::placeholder { color: #d1d5db !important; opacity: 1 !important; -webkit-text-fill-color: #d1d5db !important; } .gradio-container .bright-input textarea, .gradio-container .bright-input input { background: #1e293b !important; border: 1px solid #64748b !important; } /* 输入框下方 info 提示文字 */ .gradio-container .hint, .gradio-container .info, .gradio-container .form .secondary-wrap, .gradio-container span[data-testid="block-info"] { color: #94a3b8 !important; font-size: 0.88rem !important; } /* Markdown 内联代码 — 修复白底看不见 */ .gradio-container code, .gradio-container .prose code, .gradio-container .markdown-text code, .gradio-container pre code { background: #1e3a5f !important; color: #bfdbfe !important; border: 1px solid #3b82f6 !important; padding: 2px 10px !important; border-radius: 6px !important; font-size: 0.9em !important; } .gradio-container pre { background: #111827 !important; border: 1px solid #374151 !important; padding: 12px !important; border-radius: 8px !important; } /* 组件标签 — 修复白底蓝字 */ .gradio-container .block-label, .gradio-container .label-wrap, .gradio-container .label-wrap span, .gradio-container span.label { background: #1e293b !important; background-color: #1e293b !important; color: #93c5fd !important; font-weight: 600 !important; font-size: 0.95rem !important; border: 1px solid #475569 !important; border-radius: 6px !important; padding: 4px 10px !important; } /* 说明文字块 */ .hint-box { background: #1e293b !important; border: 1px solid #475569 !important; border-radius: 10px !important; padding: 14px 18px !important; color: #e2e8f0 !important; font-size: 0.95rem !important; line-height: 1.7 !important; margin-bottom: 12px !important; } .hint-box strong { color: #ffffff !important; } .file-tag { display: inline-block; background: #1e3a5f !important; color: #bfdbfe !important; border: 1px solid #3b82f6 !important; padding: 3px 12px !important; border-radius: 6px !important; font-family: "JetBrains Mono", Consolas, monospace !important; font-weight: 700 !important; font-size: 0.92em !important; } /* 页头 + 安装 App 按钮 */ .header-row { align-items: flex-start !important; width: 100% !important; } .header-row > div { flex: 1 1 auto !important; } .pwa-install-btn { display: none; align-items: center; gap: 10px; padding: 12px 20px; margin-top: 8px; background: linear-gradient(135deg, #2563eb 0%, #1d4ed8 100%); color: #ffffff !important; border: 2px solid #60a5fa; border-radius: 12px; cursor: pointer; font-weight: 700; font-size: 1rem; font-family: inherit; box-shadow: 0 4px 20px rgba(37, 99, 235, 0.45); transition: transform 0.15s, box-shadow 0.15s; white-space: nowrap; } .pwa-install-btn:hover { transform: translateY(-1px); box-shadow: 0 6px 24px rgba(37, 99, 235, 0.55); } .pwa-install-btn.pwa-ready { animation: pwa-pulse 2s infinite; } @keyframes pwa-pulse { 0%, 100% { box-shadow: 0 4px 20px rgba(37, 99, 235, 0.45); } 50% { box-shadow: 0 4px 28px rgba(96, 165, 250, 0.8); } } .pwa-install-icon { flex-shrink: 0; border-radius: 6px; } .pwa-install-text { color: #ffffff !important; } /* 安装引导弹窗 */ .pwa-modal-overlay { position: fixed; inset: 0; z-index: 99999; background: rgba(0, 0, 0, 0.75); display: flex; align-items: center; justify-content: center; padding: 20px; } .pwa-modal { position: relative; background: #1e293b; border: 2px solid #3b82f6; border-radius: 16px; padding: 28px 32px; max-width: 420px; width: 100%; color: #e2e8f0; box-shadow: 0 20px 60px rgba(0, 0, 0, 0.5); } .pwa-modal h3 { color: #ffffff !important; margin: 0 0 12px !important; font-size: 1.25rem !important; } .pwa-modal p { color: #cbd5e1 !important; line-height: 1.6 !important; } .pwa-modal ol { color: #e2e8f0 !important; padding-left: 20px !important; line-height: 1.8 !important; } .pwa-modal-tip { font-size: 0.85rem !important; color: #93c5fd !important; margin-top: 16px !important; } .pwa-modal-close { position: absolute; top: 12px; right: 14px; background: #374151; border: none; color: #fff; width: 32px; height: 32px; border-radius: 8px; cursor: pointer; font-size: 1rem; } @media (max-width: 640px) { .header-row { flex-direction: column !important; } .pwa-install-btn { width: 100% !important; justify-content: center !important; margin-top: 12px !important; } } .gradio-container .wrap .readonly textarea { background: #111827 !important; color: #e5e7eb !important; border-color: #374151 !important; } .gradio-container .tab-nav button { color: #9ca3af !important; font-weight: 600 !important; font-size: 0.95rem !important; padding: 10px 18px !important; } .gradio-container .tab-nav button.selected { color: #ffffff !important; background: #1e3a5f !important; border-bottom: 3px solid #3b82f6 !important; } .gradio-container button.primary, .gradio-container .primary { background: #2563eb !important; color: #ffffff !important; font-weight: 700 !important; font-size: 0.95rem !important; border: 1px solid #3b82f6 !important; } .gradio-container button.primary:hover { background: #1d4ed8 !important; } .gradio-container button.secondary, .gradio-container button:not(.primary) { color: #e5e7eb !important; background: #374151 !important; border: 1px solid #6b7280 !important; font-weight: 600 !important; } .dark-panel { border: 1px solid #3b82f6 !important; border-radius: 10px !important; padding: 18px 20px !important; background: #1a2332 !important; margin-bottom: 16px !important; width: 100% !important; box-sizing: border-box !important; } .dark-panel code { color: #93c5fd !important; background: #0f172a !important; padding: 2px 6px !important; border-radius: 4px !important; } .install-hint { margin-top: 12px !important; padding: 10px 14px !important; border-radius: 8px !important; background: #1e3a5f !important; border: 1px dashed #3b82f6 !important; color: #bfdbfe !important; font-size: 0.9rem !important; } .status-card { border-radius: 10px; padding: 14px 16px; margin: 4px 0; border-left: 5px solid #6b7280; background: #1f2937; min-height: 72px; width: 100%; box-sizing: border-box; } .status-card .status-title { font-size: 0.85rem; font-weight: 700; color: #93c5fd !important; margin-bottom: 8px; } .status-card .status-body { font-size: 0.92rem; line-height: 1.55; color: #f3f4f6 !important; word-break: break-word; } .status-ok { border-left-color: #22c55e !important; background: #14291a !important; } .status-ok .status-body { color: #bbf7d0 !important; } .status-warn { border-left-color: #f59e0b !important; background: #2a2010 !important; } .status-warn .status-body { color: #fde68a !important; } .status-err { border-left-color: #ef4444 !important; background: #2a1515 !important; } .status-err .status-body { color: #fecaca !important; } .gradio-container .audio-container, .gradio-container .upload-container { border: 2px dashed #4b5563 !important; background: #1a2332 !important; } footer { visibility: hidden; } """ def _status_html(title: str, message: str, level: str = "warn") -> str: """生成高对比度状态卡片 HTML。level: ok | warn | err""" icons = {"ok": "✅", "warn": "⚠️", "err": "❌"} icon = icons.get(level, "ℹ️") # 去掉 message 里重复的 emoji,避免双图标 clean = message.lstrip("✅❌⚠️ ").strip() return ( f'
' f'
{icon} {title}
' f'
{clean}
' f"
" ) def ui_check_ollama_html() -> str: ok, msg = check_ollama_health() return _status_html("Ollama 节点", msg, "ok" if ok else "err") def ui_speaker_status_html() -> str: ok, msg = speaker_is_ready() return _status_html("音色状态", msg, "ok" if ok else "warn") def build_theme() -> gr.themes.Base: """高对比度暗色主题(Gradio 6.0 需在 launch() 传入)。""" return gr.themes.Base( primary_hue="blue", secondary_hue="blue", neutral_hue="slate", font=[gr.themes.GoogleFont("Inter"), "system-ui", "sans-serif"], font_mono=[gr.themes.GoogleFont("JetBrains Mono"), "Consolas", "monospace"], ).set( body_background_fill="#0f1419", body_background_fill_dark="#0f1419", body_text_color="#eef2f7", body_text_color_dark="#eef2f7", block_background_fill="#1a2332", block_background_fill_dark="#1a2332", block_border_color="#4b5563", block_title_text_color="#ffffff", block_label_text_color="#93c5fd", input_background_fill="#1a2332", input_background_fill_dark="#1a2332", button_primary_background_fill="#2563eb", button_primary_background_fill_hover="#1d4ed8", button_primary_text_color="#ffffff", button_secondary_background_fill="#374151", button_secondary_text_color="#e5e7eb", border_color_primary="#3b82f6", ) def build_app() -> gr.Blocks: """构建 Gradio 主界面。""" with gr.Blocks( title="Trading Studio | 交易复盘配音中控", ) as demo: with gr.Row(elem_classes=["header-row"]): gr.Markdown( f""" # ⚡ Trading Studio **本地量化交易复盘 → B 站配音生产流水线** | 模块 | 说明 | |------|------| | Whisper | 本地 GPU 语音识别 | | Gemma4 | `{MODEL_NAME}` @ `{OLLAMA_URL.replace('/api/chat', '')}` | | ChatTTS | 本地 GPU 固定音色合成 | > 仓库: [{GIT_REPO_URL}]({GIT_REPO_URL}) """, elem_classes=["dark-panel"], ) gr.HTML(INSTALL_APP_BUTTON_HTML) with gr.Row(elem_classes=["status-row"]): ollama_status = gr.HTML(value=_status_html("Ollama 节点", "正在检测...", "warn")) speaker_status = gr.HTML(value=_status_html("音色状态", "正在检测...", "warn")) refresh_btn = gr.Button("🔄 刷新状态", variant="secondary", scale=0, min_width=120) refresh_btn.click( fn=lambda: (ui_check_ollama_html(), ui_speaker_status_html()), outputs=[ollama_status, speaker_status], ) with gr.Tabs(): # ---- Tab 1: 音色锁定 ---- with gr.Tab("🎙️ 音色锁定"): gr.HTML( f'
' f'上传 10-30 秒 干净人声样本,系统将提取 Speaker Embedding ' f'并保存至 {SPEAKER_EMB_PATH.name},' f'后续合成 100% 还原音色。' f"
" ) with gr.Row(): spk_audio = gr.Audio( label="参考人声(碎碎念盲录样本)", type="filepath", sources=["upload", "microphone"], ) spk_transcript = gr.Textbox( label="参考音频精确转写(可选,提升还原度)", placeholder="示例:今天开了三单,第一单手贱提前平了,第二单…", info="请尽量与参考音频内容完全一致,可提升音色还原度", lines=6, elem_classes=["bright-input"], ) lock_btn = gr.Button("🔒 锁定音色", variant="primary") lock_log = gr.Textbox(label="锁定结果", lines=4, interactive=False) lock_btn.click( ui_lock_speaker, [spk_audio, spk_transcript], [lock_log, speaker_status], ) # ---- Tab 2: 分步操作 ---- with gr.Tab("🔧 分步流水线"): with gr.Row(elem_classes=["pipeline-steps"]): with gr.Column(scale=1): gr.Markdown("### Step 1 · 音频极速识别") rec_audio = gr.Audio( label="交易复盘碎碎念录音", type="filepath", sources=["upload", "microphone"], ) transcribe_btn = gr.Button("⚡ Faster-Whisper 识别", variant="primary") transcribe_log = gr.Textbox(label="识别日志", lines=2, interactive=False) with gr.Column(scale=1): gr.Markdown("### Step 2 · Gemma4 纪律审判") raw_text = gr.Textbox( label="转写原文(可编辑)", lines=10, placeholder="识别结果将显示在此,也可手动粘贴...", ) polish_btn = gr.Button("⚖️ 远程 Gemma4 严厉润色", variant="primary") polish_log = gr.Textbox(label="润色日志", lines=2, interactive=False) with gr.Column(scale=1): gr.Markdown("### Step 3 · ChatTTS 配音合成") polished_text = gr.Textbox( label="润色配音稿(可编辑)", lines=10, placeholder="润色结果将显示在此...", ) synth_btn = gr.Button("🔊 合成配音 WAV", variant="primary") synth_log = gr.Textbox(label="合成日志", lines=2, interactive=False) output_audio = gr.Audio(label="成品配音", type="filepath") transcribe_btn.click(ui_transcribe, rec_audio, [raw_text, transcribe_log]) polish_btn.click(ui_polish, raw_text, [polished_text, polish_log]) synth_btn.click(ui_synthesize, polished_text, [output_audio, synth_log]) # ---- Tab 3: 一键生产 ---- with gr.Tab("🚀 一键生产"): gr.Markdown( "上传碎碎念录音,系统自动完成 **识别 → 润色 → 合成** 全流程。" ) with gr.Row(): pipe_audio = gr.Audio( label="复盘录音", type="filepath", sources=["upload", "microphone"], ) pipe_manual = gr.Textbox( label="或手动输入转写(跳过识别)", lines=4, placeholder="若已有转写文本,可直接粘贴,留空则走 Whisper 识别", ) skip_polish_cb = gr.Checkbox( label="跳过 Gemma4 润色(仅测试 TTS)", value=False, ) pipeline_btn = gr.Button("▶ 启动全流程", variant="primary", size="lg") pipeline_log = gr.Textbox(label="流水线日志", lines=6, interactive=False) with gr.Row(elem_classes=["pipeline-output-row"]): pipe_raw = gr.Textbox(label="转写原文", lines=6) pipe_polished = gr.Textbox(label="润色稿", lines=6) pipe_output = gr.Audio(label="成品配音", type="filepath") pipeline_btn.click( ui_full_pipeline, [pipe_audio, skip_polish_cb, pipe_manual], [pipe_raw, pipe_polished, pipe_output, pipeline_log], ) demo.load( fn=lambda: (ui_check_ollama_html(), ui_speaker_status_html()), outputs=[ollama_status, speaker_status], ) return demo def create_fastapi_app(): """创建 FastAPI 应用:Gradio 中控 + PWA 静态资源。""" from fastapi import FastAPI from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles fastapi_app = FastAPI(title="Trading Studio", docs_url=None, redoc_url=None) if PWA_DIR.is_dir(): fastapi_app.mount("/pwa", StaticFiles(directory=str(PWA_DIR)), name="pwa") @fastapi_app.get("/manifest.webmanifest") async def manifest(): return FileResponse( PWA_DIR / "manifest.webmanifest", media_type="application/manifest+json", ) @fastapi_app.get("/sw.js") async def service_worker(): return FileResponse( PWA_DIR / "sw.js", media_type="application/javascript", headers={"Service-Worker-Allowed": "/"}, ) blocks = build_app() gr.mount_gradio_app( fastapi_app, blocks, path="/", css=CUSTOM_CSS, theme=build_theme(), head=PWA_HEAD, allowed_paths=[str(APP_ROOT / "outputs")], ) return fastapi_app def main() -> None: """主入口:FastAPI + Gradio + PWA。""" import uvicorn logger.info("Trading Studio 启动中... HOST=%s PORT=%s", HOST, PORT) app = create_fastapi_app() uvicorn.run( app, host=HOST, port=PORT, log_level="info", access_log=True, ) if __name__ == "__main__": main()