Upload 3 files

Dockerfile.txt

@@ -0,0 +1,39 @@
+FROM python:3.11-slim
+
+# 安装系统依赖（含 sox、libsndfile）
+RUN apt-get update && apt-get install -y \
+    sox \
+    libsox-fmt-all \
+    libsndfile1 \
+    ffmpeg \
+    git \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+# 先装 CPU 版 torch（HF 免费 Space 无 GPU）
+RUN pip install --no-cache-dir \
+    torch==2.3.1+cpu \
+    torchaudio==2.3.1+cpu \
+    --index-url https://download.pytorch.org/whl/cpu
+
+# 再装其他依赖
+RUN pip install --no-cache-dir \
+    transformers==4.57.3 \
+    accelerate \
+    einops \
+    gradio \
+    fastapi \
+    uvicorn[standard] \
+    python-multipart \
+    librosa \
+    soundfile \
+    numpy \
+    onnxruntime \
+    qwen-tts
+
+COPY app.py .
+
+EXPOSE 7860
+
+CMD ["python", "app.py"]

README.md

@@ -1,10 +1,10 @@
 ---
-title: Subtitle Gen
-emoji: 🚀
+title: Qwen3 TTS
+emoji: 🎙️
 colorFrom: blue
-colorTo: red
+colorTo: purple
 sdk: docker
+app_port: 7860
 pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+license: apache-2.0
+---

app.py

@@ -0,0 +1,356 @@
+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)