feat: add training model/sample browsing API + WebUI, relax ref audio duration

为 TTS More 工作台增加远程获取训练数据的能力,并解除参考音频时长的硬编码阻断。

API (api_v2.py, 端口 9880):
- GET  /models                       列出训练角色(exp_name)及匹配权重
- GET  /models/{name}/samples        列出训练样本(音频+参考文本), 含目录穿越校验
- GET  /status                       当前权重/版本/设备状态
- POST /upload_ref                   上传参考音频, 文件名白名单清洗

Gradio WebUI (inference_webui.py):
- 新增「训练角色选择」区域: 模型名下拉/刷新/自动匹配权重/样本下拉/预览/应用样本
- on_auto_select_weights 采用稳健写法: 仅返回 Dropdown 值, 由已有 .change
  绑定触发切换 (change_sovits_weights 是生成器, 手动消费会丢失组件更新)

时长硬编码解除 (3~10s 不再阻断):
- TTS_infer_pack/TTS.py:815       raise OSError -> print [Warning]
- inference_webui.py:853          raise OSError -> gr.Warning (建议性文案)
- inference_webui.py:1367         标签改为「推荐3~10秒」

依赖: requirements.txt 追加 python-multipart (/upload_ref 必需)

向后兼容: 现有 /tts /set_gpt_weights /set_sovits_weights 签名与返回不变,
TTS 推理 pipeline 本体逻辑未改。所有改动 py_compile 通过。
This commit is contained in:
XucroYuri 2026-07-07 11:32:14 +08:00
parent bf81cdb14a
commit 7be8a59e52
4 changed files with 397 additions and 7 deletions

View File

@ -813,8 +813,9 @@ class TTS:
)
with torch.no_grad():
wav16k, sr = librosa.load(ref_wav_path, sr=16000)
if wav16k.shape[0] > 160000 or wav16k.shape[0] < 48000:
raise OSError(i18n("参考音频在3~10秒范围外请更换"))
ref_duration = wav16k.shape[0] / 16000
if ref_duration < 3 or ref_duration > 10:
print(f"[Warning] Reference audio duration {ref_duration:.1f}s is outside the recommended 3~10s range, proceeding anyway")
wav16k = torch.from_numpy(wav16k)
zero_wav_torch = torch.from_numpy(zero_wav)
wav16k = wav16k.to(self.configs.device)

View File

@ -1,3 +1,5 @@
from __future__ import annotations
"""
按中英混合识别
按日英混合识别
@ -27,6 +29,7 @@ import re
import sys
import traceback
import warnings
from pathlib import Path
import torch
import torchaudio
@ -851,9 +854,9 @@ def get_tts_wav(
if not ref_free:
with torch.no_grad():
wav16k, sr = librosa.load(ref_wav_path, sr=16000)
if wav16k.shape[0] > 160000 or wav16k.shape[0] < 48000:
gr.Warning(i18n("参考音频在3~10秒范围外请更换"))
raise OSError(i18n("参考音频在3~10秒范围外请更换"))
ref_duration = wav16k.shape[0] / 16000
if ref_duration < 3 or ref_duration > 10:
gr.Warning(i18n(f"参考音频时长 {ref_duration:.1f} 秒,超出推荐的 3~10 秒范围,可能影响合成质量"))
wav16k = torch.from_numpy(wav16k)
if is_half == True:
wav16k = wav16k.half().to(device)
@ -1202,6 +1205,136 @@ def html_left(text, label="p"):
</div>"""
# ===================== 训练角色选择 / 训练样本浏览 辅助函数 =====================
def scan_model_names() -> list[str]:
"""扫描 logs/ 目录获取所有训练角色名(含 2-name2text.txt 的子目录)。"""
from config import exp_root
import os
root = Path(exp_root)
if not root.exists():
return []
return sorted(
[
d.name
for d in root.iterdir()
if d.is_dir() and (d / "2-name2text.txt").exists()
]
)
def _read_name2text_for_webui(logs_dir: Path) -> dict[str, dict[str, str]]:
"""读取 2-name2text.txtWebUI 版,与 api_v2.py 的 _read_name2text 逻辑一致)。
返回 {wav_name: {"text": ..., "lang": ...}}同时以 stem 建索引
"""
path = logs_dir / "2-name2text.txt"
if not path.exists():
return {}
output: dict[str, dict[str, str]] = {}
for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
parts = line.split("\t")
if len(parts) < 4:
continue
wav_name = parts[0].strip()
text = parts[3].strip()
if wav_name and text:
entry = {"text": text, "lang": "zh"}
output[wav_name] = entry
output[Path(wav_name).stem] = entry
return output
def on_model_name_change(model_name: str):
"""模型名变化时,加载该模型的训练样本列表到下拉框与预览播放器。"""
if not model_name:
return gr.Dropdown(choices=[], value=""), gr.Audio(value=None)
from config import exp_root
logs_dir = Path(exp_root) / model_name
wav_dir = logs_dir / "5-wav32k"
samples: list[tuple[str, str]] = []
if wav_dir.exists():
name2text = _read_name2text_for_webui(logs_dir)
for f in sorted(wav_dir.iterdir()):
if f.is_file() and f.suffix.lower() in (".wav", ".mp3", ".flac"):
text = name2text.get(f.name, {}).get("text", "") or name2text.get(f.stem, {}).get("text", "")
label = f"{f.name}" + (f" | {text[:30]}" if text else "")
samples.append((label, str(f)))
choices = [s[0] for s in samples]
value = samples[0][0] if samples else ""
audio = samples[0][1] if samples else None
return gr.Dropdown(choices=choices, value=value), gr.Audio(value=audio)
def on_ref_sample_change(sample_label: str, model_name: str):
"""训练样本选择变化时,更新预览播放器。"""
if not sample_label or not model_name:
return gr.Audio(value=None)
from config import exp_root
wav_name = sample_label.split(" | ")[0]
audio_path = Path(exp_root) / model_name / "5-wav32k" / wav_name
return gr.Audio(value=str(audio_path) if audio_path.exists() else None)
def on_apply_sample(sample_label: str, model_name: str):
"""应用选中样本:写入参考音频路径和参考文本。"""
if not sample_label or not model_name:
return None, "", gr.Dropdown()
from config import exp_root
wav_name = sample_label.split(" | ")[0]
audio_path = Path(exp_root) / model_name / "5-wav32k" / wav_name
logs_dir = Path(exp_root) / model_name
name2text = _read_name2text_for_webui(logs_dir)
text = name2text.get(wav_name, {}).get("text", "") or name2text.get(Path(wav_name).stem, {}).get("text", "")
return (str(audio_path) if audio_path.exists() else None), text, gr.Dropdown()
def _scan_model_weights_for_webui() -> dict[str, dict[str, list[str]]]:
"""扫描权重目录WebUI 版),按 exp_name 分组。"""
from config import GPT_weight_root, SoVITS_weight_root
grouped: dict[str, dict[str, list[str]]] = {}
for root in GPT_weight_root:
root_path = Path(root)
if not root_path.exists():
continue
for f in root_path.iterdir():
if f.is_file() and f.suffix == ".ckpt":
name = re.split(r"-e\d+", f.stem, flags=re.IGNORECASE)[0]
grouped.setdefault(name, {"gpt": [], "sovits": []})
grouped[name]["gpt"].append(f"{root}/{f.name}")
for root in SoVITS_weight_root:
root_path = Path(root)
if not root_path.exists():
continue
for f in root_path.iterdir():
if f.is_file() and f.suffix == ".pth":
name = re.split(r"_e\d+_s\d+", f.stem, flags=re.IGNORECASE)[0]
grouped.setdefault(name, {"gpt": [], "sovits": []})
grouped[name]["sovits"].append(f"{root}/{f.name}")
return grouped
def on_auto_select_weights(model_name: str):
"""根据模型名自动匹配并选择最佳 GPT/SoVITS 权重。
仅返回 Dropdown value真正的权重切换由已绑定的
GPT_dropdown.change / SoVITS_dropdown.change value 变化时自动触发
change_sovits_weights 是生成器手动消费会丢失对其它组件的更新故不直接调用
"""
if not model_name:
return gr.Dropdown(), gr.Dropdown()
weights = _scan_model_weights_for_webui()
model_weights = weights.get(model_name, {"gpt": [], "sovits": []})
# 选择最高 epoch 的权重(按字符串排序后取末尾)
gpt_best = sorted(model_weights["gpt"])[-1] if model_weights["gpt"] else None
sovits_best = sorted(model_weights["sovits"])[-1] if model_weights["sovits"] else None
return gr.Dropdown(value=gpt_best), gr.Dropdown(value=sovits_best)
with gr.Blocks(title="GPT-SoVITS WebUI", analytics_enabled=False, js=js, css=css) as app:
gr.HTML(
top_html.format(
@ -1229,9 +1362,39 @@ with gr.Blocks(title="GPT-SoVITS WebUI", analytics_enabled=False, js=js, css=css
)
refresh_button = gr.Button(i18n("刷新模型路径"), variant="primary", scale=14)
refresh_button.click(fn=change_choices, inputs=[], outputs=[SoVITS_dropdown, GPT_dropdown])
# ===== 新增:训练角色选择区域 =====
with gr.Group():
gr.Markdown(html_center(i18n("训练角色选择(从训练数据中快速选择)"), "h3"))
with gr.Row():
model_name_dropdown = gr.Dropdown(
label=i18n("模型名(训练实验名)"),
choices=[],
value="",
interactive=True,
scale=14,
)
refresh_models_btn = gr.Button(i18n("刷新模型列表"), variant="primary", scale=7)
auto_select_weights_btn = gr.Button(i18n("自动匹配权重"), variant="secondary", scale=7)
with gr.Row():
ref_sample_dropdown = gr.Dropdown(
label=i18n("训练样本音频"),
choices=[],
value="",
interactive=True,
scale=14,
)
ref_sample_player = gr.Audio(
label=i18n("样本预览"),
type="filepath",
scale=14,
)
apply_sample_btn = gr.Button(
i18n("应用选中样本到参考音频和文本"),
variant="primary",
)
gr.Markdown(html_center(i18n("*请上传并填写参考信息"), "h3"))
with gr.Row():
inp_ref = gr.Audio(label=i18n("请上传3~10秒内参考音频超过会报错"), type="filepath", scale=13)
inp_ref = gr.Audio(label=i18n("请上传参考音频推荐3~10秒"), type="filepath", scale=13)
with gr.Column(scale=13):
ref_text_free = gr.Checkbox(
label=i18n("开启无参考文本模式。不填参考文本亦相当于开启。")
@ -1407,6 +1570,36 @@ with gr.Blocks(title="GPT-SoVITS WebUI", analytics_enabled=False, js=js, css=css
)
GPT_dropdown.change(change_gpt_weights, [GPT_dropdown], [])
# ===== 新增事件绑定:训练角色选择 / 样本浏览 =====
refresh_models_btn.click(
fn=scan_model_names,
inputs=[],
outputs=[model_name_dropdown],
)
model_name_dropdown.change(
fn=on_model_name_change,
inputs=[model_name_dropdown],
outputs=[ref_sample_dropdown, ref_sample_player],
)
ref_sample_dropdown.change(
fn=on_ref_sample_change,
inputs=[ref_sample_dropdown, model_name_dropdown],
outputs=[ref_sample_player],
)
apply_sample_btn.click(
fn=on_apply_sample,
inputs=[ref_sample_dropdown, model_name_dropdown],
outputs=[inp_ref, prompt_text, ref_sample_dropdown],
)
auto_select_weights_btn.click(
fn=on_auto_select_weights,
inputs=[model_name_dropdown],
outputs=[GPT_dropdown, SoVITS_dropdown],
)
# 页面加载时自动扫描模型列表
app.load(fn=scan_model_names, inputs=[], outputs=[model_name_dropdown])
# gr.Markdown(value=i18n("文本切分工具。太长的文本合成出来效果不一定好,所以太长建议先切。合成会根据文本的换行分开合成再拼起来。"))
# with gr.Row():
# text_inp = gr.Textbox(label=i18n("需要合成的切分前文本"), value="")

195
api_v2.py
View File

@ -1,3 +1,5 @@
from __future__ import annotations
"""
# WebAPI文档
@ -102,8 +104,10 @@ RESP:
"""
import os
import re
import sys
import traceback
from pathlib import Path
from typing import Generator, Union
now_dir = os.getcwd()
@ -112,11 +116,12 @@ sys.path.append("%s/GPT_SoVITS" % (now_dir))
import argparse
import subprocess
import uuid
import wave
import signal
import numpy as np
import soundfile as sf
from fastapi import FastAPI, Response
from fastapi import FastAPI, Response, UploadFile, File
from fastapi.responses import StreamingResponse, JSONResponse
import uvicorn
from io import BytesIO
@ -124,6 +129,7 @@ from tools.i18n.i18n import I18nAuto
from GPT_SoVITS.TTS_infer_pack.TTS import TTS, TTS_Config
from GPT_SoVITS.TTS_infer_pack.text_segmentation_method import get_method_names as get_cut_method_names
from pydantic import BaseModel
from config import GPT_weight_root, SoVITS_weight_root, exp_root
import threading
# print(sys.path)
@ -565,6 +571,193 @@ async def set_sovits_weights(weights_path: str = None):
return JSONResponse(status_code=200, content={"message": "success"})
# ===================== 训练角色 / 训练样本 / 状态 辅助函数 =====================
def _extract_exp_name_from_gpt_weight(filename: str) -> str:
"""从 GPT 权重文件名提取 exp_name。
: '光头TTS-华-e10.ckpt' -> '光头TTS-华'
"""
stem = Path(filename).stem
return re.split(r"-e\d+", stem, flags=re.IGNORECASE)[0]
def _extract_exp_name_from_sovits_weight(filename: str) -> str:
"""从 SoVITS 权重文件名提取 exp_name。
: '光头TTS-华_e4_s72.pth' -> '光头TTS-华'
"""
stem = Path(filename).stem
return re.split(r"_e\d+_s\d+", stem, flags=re.IGNORECASE)[0]
def _scan_model_weights() -> dict[str, dict[str, list[str]]]:
"""扫描所有权重目录,按 exp_name 分组。
返回: {"光头TTS-华": {"gpt": [...], "sovits": [...]}}
"""
grouped: dict[str, dict[str, list[str]]] = {}
for root in GPT_weight_root:
root_path = Path(root)
if not root_path.exists():
continue
for f in root_path.iterdir():
if f.is_file() and f.suffix == ".ckpt":
name = _extract_exp_name_from_gpt_weight(f.name)
grouped.setdefault(name, {"gpt": [], "sovits": []})
grouped[name]["gpt"].append(f"{root}/{f.name}")
for root in SoVITS_weight_root:
root_path = Path(root)
if not root_path.exists():
continue
for f in root_path.iterdir():
if f.is_file() and f.suffix == ".pth":
name = _extract_exp_name_from_sovits_weight(f.name)
grouped.setdefault(name, {"gpt": [], "sovits": []})
grouped[name]["sovits"].append(f"{root}/{f.name}")
return grouped
def _read_name2text(logs_dir: Path) -> dict[str, dict[str, str]]:
"""读取 2-name2text.txt返回 {wav_name: {"text": ..., "lang": ...}}。
每行 Tab 分隔 4 字段: wav_name\tphones\tword2ph\tnorm_text
同时以 wav_name 和其 stem去扩展名建立索引便于按文件名或 stem 查询
"""
path = logs_dir / "2-name2text.txt"
if not path.exists():
return {}
output: dict[str, dict[str, str]] = {}
for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
parts = line.split("\t")
if len(parts) < 4:
continue
wav_name = parts[0].strip()
text = parts[3].strip()
if wav_name and text:
entry = {"text": text, "lang": "zh"}
output[wav_name] = entry
output[Path(wav_name).stem] = entry
return output
def _safe_logs_subpath(model_name: str, *parts: str) -> Path | None:
"""解析 logs/<model_name>/<parts...> 路径并做目录穿越校验。
若解析后的绝对路径不在 exp_root 之内返回 None拒绝否则返回绝对路径
"""
root_abs = Path(exp_root).resolve()
target = (root_abs / model_name, *parts)
target_abs = Path(*[str(p) for p in target]).resolve()
try:
target_abs.relative_to(root_abs)
except ValueError:
return None
return target_abs
@APP.get("/models")
async def list_models():
"""列出所有训练角色logs 目录名)及其匹配的权重。
logs/ 目录扫描子目录名作为模型名再从 GPT/SoVITS 权重目录中
匹配同名权重文件支持 GPT_weightsGPT_weights_v2 6 个版本目录
"""
weights = _scan_model_weights()
logs_root = Path(exp_root)
models: list[dict] = []
if logs_root.exists():
for d in logs_root.iterdir():
if not d.is_dir():
continue
name = d.name
name2text = _read_name2text(d)
has_training_data = (d / "2-name2text.txt").exists()
# name2text 同时以 wav_name 和 stem 建索引2 个键),样本数按 wav_name 计:即原始行数
sample_count = len(name2text) // 2 if name2text else 0
w = weights.get(name, {"gpt": [], "sovits": []})
models.append({
"name": name,
"gpt_weights": sorted(w["gpt"]),
"sovits_weights": sorted(w["sovits"]),
"has_training_data": has_training_data,
"sample_count": sample_count,
})
# 补上 logs 中没有但有权重的模型
existing_names = {m["name"] for m in models}
for name, w in weights.items():
if name in existing_names:
continue
models.append({
"name": name,
"gpt_weights": sorted(w["gpt"]),
"sovits_weights": sorted(w["sovits"]),
"has_training_data": (Path(exp_root) / name).exists(),
"sample_count": 0,
})
models.sort(key=lambda m: m["name"])
return {"models": models}
@APP.get("/models/{model_name}/samples")
async def list_model_samples(model_name: str):
"""列出指定角色的训练样本(音频文件 + 参考文本)。
读取 logs/<model_name>/2-name2text.txt 解析标注
扫描 logs/<model_name>/5-wav32k/ 获取音频文件列表
"""
logs_abs = _safe_logs_subpath(model_name)
if logs_abs is None or not logs_abs.exists():
return JSONResponse(status_code=404, content={"message": f"model '{model_name}' not found"})
name2text = _read_name2text(logs_abs)
wav_dir = logs_abs / "5-wav32k"
samples: list[dict] = []
if wav_dir.exists():
for f in sorted(wav_dir.iterdir()):
if not f.is_file():
continue
if f.suffix.lower() not in (".wav", ".mp3", ".flac"):
continue
entry = name2text.get(f.name) or name2text.get(f.stem)
if entry is None:
continue # 与 name2text 取交集,只返回有标注的样本
rel = os.path.relpath(f, now_dir)
samples.append({
"audio_name": f.name,
"audio_path": rel.replace(os.sep, "/"),
"text": entry["text"],
"lang": entry["lang"],
})
return {"model_name": model_name, "samples": samples, "total": len(samples)}
@APP.get("/status")
async def service_status():
"""返回当前服务状态:加载的权重、版本、设备。"""
return {
"version": tts_config.version,
"device": str(tts_config.device),
"gpt_weights": tts_config.t2s_weights_path,
"sovits_weights": tts_config.vits_weights_path,
"languages": list(tts_config.languages),
}
@APP.post("/upload_ref")
async def upload_reference_audio(file: UploadFile = File(...)):
"""上传参考音频文件,返回服务端本地路径供 /tts 使用。
同机部署不需要此端点直接传本地路径即可
"""
upload_dir = Path("uploaded_audio")
upload_dir.mkdir(parents=True, exist_ok=True)
raw_name = os.path.basename(file.filename or "")
# 文件名白名单清洗:仅保留字母数字、中文、._- 与扩展名前的点
safe_name = re.sub(r"[^\w.\u4e00-\u9fff\-]", "_", raw_name) or "ref.wav"
save_name = f"{uuid.uuid4().hex[:16]}_{safe_name}"
save_path = upload_dir / save_name
content = await file.read()
save_path.write_bytes(content)
rel = os.path.relpath(save_path, now_dir)
return {"path": rel.replace(os.sep, "/")}
if __name__ == "__main__":
try:
if host == "None": # 在调用时使用 -a None 参数可以让api监听双栈

View File

@ -41,3 +41,6 @@ pydantic<=2.10.6
ctranslate2>=4.0,<5
av>=11
tqdm
# FastAPI file upload support for /upload_ref endpoint
python-multipart