Delete app.py

app.py

@@ -1,356 +0,0 @@
-import os
-import io
-import gc
-import tempfile
-import torch
-import soundfile as sf
-import numpy as np
-import gradio as gr
-from fastapi import FastAPI, UploadFile, File, Form, HTTPException
-from fastapi.responses import StreamingResponse
-from fastapi.middleware.cors import CORSMiddleware
-from typing import Optional
-import uvicorn
-
-# ── 设备 & 精度 ───────────────────────────────────────────────────────────────
-DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
-DTYPE  = torch.bfloat16
-
-print(f"[TTS] Device={DEVICE}, dtype={DTYPE}")
-from qwen_tts import Qwen3TTSModel
-
-# ── 启动只加载 Base ───────────────────────────────────────────────────────────
-print("[TTS] Loading Base model...")
-model_base = Qwen3TTSModel.from_pretrained(
-    "Qwen/Qwen3-TTS-12Hz-1.7B-Base",
-    device_map=DEVICE,
-    dtype=DTYPE,
-)
-print("[TTS] Base model ready.")
-LANGUAGES = model_base.get_supported_languages()
-
-# ── CustomVoice 按需加载 ──────────────────────────────────────────────────────
-model_custom = None
-
-def _load_custom():
-    global model_custom
-    if model_custom is not None:
-        return
-    print("[TTS] Loading CustomVoice...")
-    model_custom = Qwen3TTSModel.from_pretrained(
-        "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice",
-        device_map=DEVICE,
-        dtype=DTYPE,
-    )
-    print("[TTS] CustomVoice ready.")
-
-def _unload_custom():
-    global model_custom
-    if model_custom is None:
-        return
-    print("[TTS] Unloading CustomVoice...")
-    del model_custom
-    model_custom = None
-    gc.collect()
-    if DEVICE == "cuda":
-        torch.cuda.empty_cache()
-    print("[TTS] CustomVoice unloaded.")
-
-# ── Speaker 列表（启动时获取一次）────────────────────────────────────────────
-print("[TTS] Fetching speaker list...")
-_load_custom()
-SPEAKERS = model_custom.get_supported_speakers()
-_unload_custom()
-print(f"[TTS] Speakers: {SPEAKERS}")
-
-# ── 全局 anchor 存储  role -> {sr, audio, text} ───────────────────────────────
-_anchors: dict = {}
-
-# ─────────────────────────────────────────────────────────────────────────────
-# 工具
-# ─────────────────────────────────────────────────────────────────────────────
-
-def _wav_bytes(audio: np.ndarray, sr: int) -> bytes:
-    buf = io.BytesIO()
-    sf.write(buf, audio, sr, format="WAV")
-    buf.seek(0)
-    return buf.read()
-
-def _wav_response(audio: np.ndarray, sr: int, name: str = "out.wav") -> StreamingResponse:
-    return StreamingResponse(
-        io.BytesIO(_wav_bytes(audio, sr)),
-        media_type="audio/wav",
-        headers={"Content-Disposition": f"attachment; filename={name}"},
-    )
-
-def _tmp_wav(sr: int, audio: np.ndarray) -> str:
-    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
-        sf.write(f.name, audio, sr)
-        return f.name
-
-# ─────────────────────────────────────────────────────────────────────────────
-# FastAPI
-# ─────────────────────────────────────────────────────────────────────────────
-
-app_api = FastAPI(title="Qwen3-TTS", version="4.0.0")
-app_api.add_middleware(CORSMiddleware, allow_origins=["*"],
-                       allow_methods=["*"], allow_headers=["*"])
-
-@app_api.get("/health")
-def health():
-    return {
-        "status": "ok", "device": DEVICE,
-        "custom_loaded": model_custom is not None,
-        "anchors": list(_anchors.keys()),
-        "languages": LANGUAGES,
-    }
-
-# ── /batch_anchor ─────────────────────────────────────────────────────────────
-# 一次性建立所有角色 anchor，CustomVoice 只加载一次
-@app_api.post("/batch_anchor")
-async def batch_anchor(
-    roles:     str = Form(...),   # JSON list
-    speakers:  str = Form(...),   # JSON list
-    texts:     str = Form(...),   # JSON list
-    languages: str = Form(...),   # JSON list
-    instructs: str = Form(...),   # JSON list
-):
-    import json, base64
-    roles_     = json.loads(roles)
-    speakers_  = json.loads(speakers)
-    texts_     = json.loads(texts)
-    languages_ = json.loads(languages)
-    instructs_ = json.loads(instructs)
-
-    if not (len(roles_) == len(speakers_) == len(texts_) == len(languages_) == len(instructs_)):
-        raise HTTPException(400, "all lists must be same length")
-
-    results = {}
-    try:
-        _load_custom()
-        for role, spk, text, lang, inst in zip(
-                roles_, speakers_, texts_, languages_, instructs_):
-            inst = inst if inst and inst.strip() else None
-            wavs, sr = model_custom.generate_custom_voice(
-                text=text, language=lang, speaker=spk, instruct=inst)
-            audio = wavs[0]
-            _anchors[role] = {"sr": sr, "audio": audio, "text": text}
-            wav_b64 = base64.b64encode(_wav_bytes(audio, sr)).decode()
-            results[role] = {"sr": sr, "wav_b64": wav_b64}
-            print(f"[batch_anchor] ✓ {role} ({spk})")
-    finally:
-        _unload_custom()
-
-    return results
-
-# ── /tts (兼容旧版，单角色) ───────────────────────────────────────────────────
-@app_api.post("/tts")
-async def tts_single(
-    text:       str  = Form(...),
-    speaker:    str  = Form("Vivian"),
-    language:   str  = Form("chinese"),
-    instruct:   str  = Form(""),
-    role:       str  = Form("_narrator"),
-    set_anchor: bool = Form(False),   # FIX: 仅当显式要求时才(重新)写入/覆盖 _anchors[role]
-):
-    if not text.strip():
-        raise HTTPException(400, "text is empty")
-    inst = instruct if instruct.strip() else None
-    try:
-        _load_custom()
-        wavs, sr = model_custom.generate_custom_voice(
-            text=text, language=language, speaker=speaker, instruct=inst)
-        audio = wavs[0]
-        # FIX: 之前这里无条件覆盖 _anchors[role]，导致任何一次带 instruct 的
-        # 情绪台词（speak_emotion / dialogue 里非 narrator 行）都会把已经建立好的
-        # anchor 顶替掉，后续 /clone 就会用这句"带情绪"的音频当参考，造成音色跑偏。
-        # 现在：只有显式传 set_anchor=true（建立/重建锚点时）才写入。
-        if set_anchor or role not in _anchors:
-            _anchors[role] = {"sr": sr, "audio": audio, "text": text}
-        return _wav_response(audio, sr, "speech.wav")
-    except Exception as e:
-        raise HTTPException(500, str(e))
-    finally:
-        _unload_custom()
-
-# ── /clone ────────────────────────────────────────────────────────────────────
-# ⚠️ 修改：支持两种方式 - 使用服务端 anchor 或直接上传参考音频
-@app_api.post("/clone")
-async def clone(
-    text:            str         = Form(...),
-    language:        str         = Form("chinese"),
-    role:            str         = Form("_narrator"),
-    reference_text:  str         = Form(""),
-    reference_audio: UploadFile  = File(None),
-):
-    if not text.strip():
-        raise HTTPException(400, "text is empty")
-
-    tmp_path = None
-    try:
-        # FIX: 之前这里"服务端 anchor"永远优先，导致只要 role 在 _anchors 里
-        # （比如用 create 建立过），/clone 就完全无视客户端传来的 reference_audio/
-        # reference_text —— workflow.sh 里的"滚动参考"机制形同虚设，而且一旦
-        # _anchors[role] 被别的调用（如情绪台词）覆盖，全程都会跟着跑偏。
-        # 现在改为：客户端显式传了 reference_audio + reference_text 就优先用它
-        # （滚动参考 / 强制回锚都通过这个参数控制），客户端没传时才退回服务端 anchor。
-        if reference_audio is not None and reference_text.strip():
-            suffix = os.path.splitext(reference_audio.filename)[-1] or ".wav"
-            with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as f:
-                f.write(await reference_audio.read())
-                tmp_path = f.name
-            ref_text = reference_text
-            print(f"[clone] 使用客户端参考音频: {reference_audio.filename}")
-
-        elif role in _anchors:
-            anchor   = _anchors[role]
-            tmp_path = _tmp_wav(anchor["sr"], anchor["audio"])
-            ref_text = anchor["text"]
-            print(f"[clone] 使用服务端 anchor: {role}")
-
-        else:
-            raise HTTPException(
-                400, f"No anchor for role='{role}'. Call /batch_anchor or /tts first, or provide reference_audio and reference_text.")
-
-        # 生成克隆
-        wavs, sr = model_base.generate_voice_clone(
-            text=text, ref_audio=tmp_path,
-            ref_text=ref_text, language=language)
-        return _wav_response(wavs[0], sr, "clone.wav")
-        
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(500, str(e))
-    finally:
-        if tmp_path and os.path.exists(tmp_path):
-            os.unlink(tmp_path)
-
-# ── /clear_anchors ────────────────────────────────────────────────────────────
-@app_api.post("/clear_anchors")
-def clear_anchors():
-    _anchors.clear()
-    return {"status": "cleared"}
-
-# ─────────────────────────────────────────────────────────────────────────────
-# Gradio UI
-# ─────────────────────────────────────────────────────────────────────────────
-
-def ui_first(text, language, speaker, instruct):
-    if not text.strip():
-        return None, "❌ 请输入文本"
-    try:
-        _load_custom()
-        wavs, sr = model_custom.generate_custom_voice(
-            text=text, language=language, speaker=speaker,
-            instruct=instruct or None)
-        audio = wavs[0]
-        _anchors["_ui"] = {"sr": sr, "audio": audio, "text": text}
-        return (sr, audio), f"✅ 生成完毕，声音已锁定（{speaker}）"
-    except Exception as e:
-        return None, f"❌ {e}"
-    finally:
-        _unload_custom()
-
-def ui_clone(text, language):
-    if not text.strip():
-        return None, "❌ 请输入文本"
-    if "_ui" not in _anchors:
-        return None, "❌ 请先在「第一段」Tab 生成并锁定声音"
-    anchor = _anchors["_ui"]
-    tmp = _tmp_wav(anchor["sr"], anchor["audio"])
-    try:
-        wavs, sr = model_base.generate_voice_clone(
-            text=text, ref_audio=tmp,
-            ref_text=anchor["text"], language=language)
-        return (sr, wavs[0]), "✅ 生成成功"
-    except Exception as e:
-        return None, f"❌ {e}"
-    finally:
-        os.unlink(tmp)
-
-def ui_manual_clone(text, language, ref_audio, ref_text):
-    if not text.strip() or ref_audio is None:
-        return None, "❌ 请填写文本并上传参考音频"
-    if not ref_text.strip():
-        return None, "❌ 参考文本为必填项"
-    sr_r, audio_np = ref_audio
-    tmp = _tmp_wav(sr_r, audio_np)
-    try:
-        wavs, sr = model_base.generate_voice_clone(
-            text=text, ref_audio=tmp,
-            ref_text=ref_text, language=language)
-        return (sr, wavs[0]), "✅ 克隆成功"
-    except Exception as e:
-        return None, f"❌ {e}"
-    finally:
-        os.unlink(tmp)
-
-def ui_unlock():
-    _anchors.pop("_ui", None)
-    return "🔓 已解锁"
-
-with gr.Blocks(title="Qwen3-TTS", theme=gr.themes.Soft()) as gradio_ui:
-    gr.Markdown(
-        "# 🎙️ Qwen3-TTS 1.7B\n"
-        "**工作流**：第一段建立 anchor → 后续全部 clone 保持声音一致\n\n"
-        "> CustomVoice 只在生成第一段时临时加载，用完立刻释放。"
-    )
-    with gr.Tabs():
-        with gr.Tab("① 第一段"):
-            with gr.Row():
-                with gr.Column():
-                    f_text    = gr.Textbox(label="第一段文本", lines=4)
-                    f_lang    = gr.Dropdown(LANGUAGES, value="chinese", label="语言")
-                    f_speaker = gr.Dropdown(SPEAKERS, value=SPEAKERS[0], label="Speaker")
-                    f_inst    = gr.Textbox(label="风格指令（可选）",
-                                           placeholder="例：用沉稳的播音腔")
-                    with gr.Row():
-                        f_btn    = gr.Button("🎵 生成并锁定", variant="primary")
-                        f_unlock = gr.Button("🔓 解锁", variant="secondary")
-                with gr.Column():
-                    f_audio  = gr.Audio(label="输出", type="numpy")
-                    f_status = gr.Textbox(label="状态", interactive=False)
-            f_btn.click(ui_first, [f_text, f_lang, f_speaker, f_inst], [f_audio, f_status])
-            f_unlock.click(ui_unlock, [], f_status)
-
-        with gr.Tab("② 后续段落"):
-            gr.Markdown("使用锁定声音，每次只需输入文本，声音完全一致。")
-            with gr.Row():
-                with gr.Column():
-                    b_text = gr.Textbox(label="目标文本", lines=5)
-                    b_lang = gr.Dropdown(LANGUAGES, value="chinese", label="语言")
-                    b_btn  = gr.Button("🎵 生成", variant="primary")
-                with gr.Column():
-                    b_audio  = gr.Audio(label="输出", type="numpy")
-                    b_status = gr.Textbox(label="状态", interactive=False)
-            b_btn.click(ui_clone, [b_text, b_lang], [b_audio, b_status])
-
-        with gr.Tab("③ 手动克隆"):
-            with gr.Row():
-                with gr.Column():
-                    c_ref   = gr.Audio(label="参考音频（3~10 秒）", type="numpy")
-                    c_rtext = gr.Textbox(label="参考音频对应文本（必填）", lines=2)
-                    c_text  = gr.Textbox(label="目标文本", lines=4)
-                    c_lang  = gr.Dropdown(LANGUAGES, value="chinese", label="语言")
-                    c_btn   = gr.Button("🎵 生成", variant="primary")
-                with gr.Column():
-                    c_audio  = gr.Audio(label="输出", type="numpy")
-                    c_status = gr.Textbox(label="状态", interactive=False)
-            c_btn.click(ui_manual_clone,
-                        [c_text, c_lang, c_ref, c_rtext],
-                        [c_audio, c_status])
-
-    gr.Markdown(
-        "---\n### API\n"
-        "- `POST /batch_anchor` — 批量建立所有角色 anchor（CustomVoice 只加载一次）\n"
-        "- `POST /tts` — 单角色建立 anchor（兼容旧版）\n"
-        "- `POST /clone` — 用 anchor 或参考音频克隆生成\n"
-        "- `POST /clear_anchors` — 清除所有 anchor\n"
-        "- `GET /health` — 状态\n"
-    )
-
-final_app = gr.mount_gradio_app(app_api, gradio_ui, path="/")
-
-if __name__ == "__main__":
-    uvicorn.run(final_app, host="0.0.0.0", port=7860)