mirror of
https://github.com/RVC-Boss/GPT-SoVITS.git
synced 2026-07-25 20:35:13 +08:00
6931a97594
1. emotion 双源兼容 (核心): proplus 训练WebUI 有两种 emotion 数据源, 互斥存在: - 旧: ASR标注 .list 第5列 (output/asr_opt/<exp>/*.list) - 新: logs/<exp>/audio_metadata.json (推理WebUI回写) _read_emotion_map_for_webui / _read_emotion_map 现合并两源, audio_metadata.json 优先(推理时回写的更新)。 实测: 光头TTS新(emotion在.list)->惊恐; 九九-许珺雯(emotion在json)->挑眉自信。 2. on_model_name_change 修复初始 emotion 不显示: 原返回3值[样本,GPT,SoVITS], 依赖 ref_sample_dropdown.change 级联触发。 但 app.load 设置 dropdown value 后 Gradio 不一定触发 .change, 导致页面加载时 emotion 框为空。 现返回6值[样本,GPT,SoVITS,参考音频,参考文本,情绪], 一次性联动全部。 3. 布局紧凑: - 参考音频去掉 Column 包裹+equal_height(导致Audio被拉伸留白), 自然高度 - 参考文本 4行->3行 - 多参考文件+v3参数从右列移出为独立满宽行(不再干扰主行对齐) 主行变两列: 参考音频(10) + 语种&文本&情绪(10), 紧凑对齐。
819 lines
31 KiB
Python
819 lines
31 KiB
Python
from __future__ import annotations
|
||
|
||
"""
|
||
# WebAPI文档
|
||
|
||
` python api_v2.py -a 127.0.0.1 -p 9880 -c GPT_SoVITS/configs/tts_infer.yaml `
|
||
|
||
## 执行参数:
|
||
`-a` - `绑定地址, 默认"127.0.0.1"`
|
||
`-p` - `绑定端口, 默认9880`
|
||
`-c` - `TTS配置文件路径, 默认"GPT_SoVITS/configs/tts_infer.yaml"`
|
||
|
||
## 调用:
|
||
|
||
### 推理
|
||
|
||
endpoint: `/tts`
|
||
GET:
|
||
```
|
||
http://127.0.0.1:9880/tts?text=先帝创业未半而中道崩殂，今天下三分，益州疲弊，此诚危急存亡之秋也。&text_lang=zh&ref_audio_path=archive_jingyuan_1.wav&prompt_lang=zh&prompt_text=我是「罗浮」云骑将军景元。不必拘谨，「将军」只是一时的身份，你称呼我景元便可&text_split_method=cut5&batch_size=1&media_type=wav&streaming_mode=true
|
||
```
|
||
|
||
POST:
|
||
```json
|
||
{
|
||
"text": "", # str.(required) text to be synthesized
|
||
"text_lang: "", # str.(required) language of the text to be synthesized
|
||
"ref_audio_path": "", # str.(required) reference audio path
|
||
"aux_ref_audio_paths": [], # list.(optional) auxiliary reference audio paths for multi-speaker tone fusion
|
||
"prompt_text": "", # str.(optional) prompt text for the reference audio
|
||
"prompt_lang": "", # str.(required) language of the prompt text for the reference audio
|
||
"top_k": 15, # int. top k sampling
|
||
"top_p": 1, # float. top p sampling
|
||
"temperature": 1, # float. temperature for sampling
|
||
"text_split_method": "cut5", # str. text split method, see text_segmentation_method.py for details.
|
||
"batch_size": 1, # int. batch size for inference
|
||
"batch_threshold": 0.75, # float. threshold for batch splitting.
|
||
"split_bucket": True, # bool. whether to split the batch into multiple buckets.
|
||
"speed_factor":1.0, # float. control the speed of the synthesized audio.
|
||
"fragment_interval":0.3, # float. to control the interval of the audio fragment.
|
||
"seed": -1, # int. random seed for reproducibility.
|
||
"parallel_infer": True, # bool. whether to use parallel inference.
|
||
"repetition_penalty": 1.35, # float. repetition penalty for T2S model.
|
||
"sample_steps": 32, # int. number of sampling steps for VITS model V3.
|
||
"super_sampling": False, # bool. whether to use super-sampling for audio when using VITS model V3.
|
||
"streaming_mode": False, # bool or int. return audio chunk by chunk.T he available options are: 0,1,2,3 or True/False (0/False: Disabled | 1/True: Best Quality, Slowest response speed (old version streaming_mode) | 2: Medium Quality, Slow response speed | 3: Lower Quality, Faster response speed )
|
||
"overlap_length": 2, # int. overlap length of semantic tokens for streaming mode.
|
||
"min_chunk_length": 16, # int. The minimum chunk length of semantic tokens for streaming mode. (affects audio chunk size)
|
||
}
|
||
```
|
||
|
||
RESP:
|
||
成功: 直接返回 wav 音频流， http code 200
|
||
失败: 返回包含错误信息的 json, http code 400
|
||
|
||
### 命令控制
|
||
|
||
endpoint: `/control`
|
||
|
||
command:
|
||
"restart": 重新运行
|
||
"exit": 结束运行
|
||
|
||
GET:
|
||
```
|
||
http://127.0.0.1:9880/control?command=restart
|
||
```
|
||
POST:
|
||
```json
|
||
{
|
||
"command": "restart"
|
||
}
|
||
```
|
||
|
||
RESP: 无
|
||
|
||
|
||
### 切换GPT模型
|
||
|
||
endpoint: `/set_gpt_weights`
|
||
|
||
GET:
|
||
```
|
||
http://127.0.0.1:9880/set_gpt_weights?weights_path=GPT_SoVITS/pretrained_models/s1bert25hz-2kh-longer-epoch=68e-step=50232.ckpt
|
||
```
|
||
RESP:
|
||
成功: 返回"success", http code 200
|
||
失败: 返回包含错误信息的 json, http code 400
|
||
|
||
|
||
### 切换Sovits模型
|
||
|
||
endpoint: `/set_sovits_weights`
|
||
|
||
GET:
|
||
```
|
||
http://127.0.0.1:9880/set_sovits_weights?weights_path=GPT_SoVITS/pretrained_models/s2G488k.pth
|
||
```
|
||
|
||
RESP:
|
||
成功: 返回"success", http code 200
|
||
失败: 返回包含错误信息的 json, http code 400
|
||
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
import sys
|
||
import traceback
|
||
from pathlib import Path
|
||
from typing import Generator, Union
|
||
|
||
now_dir = os.getcwd()
|
||
sys.path.append(now_dir)
|
||
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, UploadFile, File
|
||
from fastapi.responses import StreamingResponse, JSONResponse
|
||
import uvicorn
|
||
from io import BytesIO
|
||
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)
|
||
i18n = I18nAuto()
|
||
cut_method_names = get_cut_method_names()
|
||
|
||
parser = argparse.ArgumentParser(description="GPT-SoVITS api")
|
||
parser.add_argument("-c", "--tts_config", type=str, default="GPT_SoVITS/configs/tts_infer.yaml", help="tts_infer路径")
|
||
parser.add_argument("-a", "--bind_addr", type=str, default="127.0.0.1", help="default: 127.0.0.1")
|
||
parser.add_argument("-p", "--port", type=int, default="9880", help="default: 9880")
|
||
args = parser.parse_args()
|
||
config_path = args.tts_config
|
||
# device = args.device
|
||
port = args.port
|
||
host = args.bind_addr
|
||
argv = sys.argv
|
||
|
||
if config_path in [None, ""]:
|
||
config_path = "GPT-SoVITS/configs/tts_infer.yaml"
|
||
|
||
tts_config = TTS_Config(config_path)
|
||
print(tts_config)
|
||
tts_pipeline = TTS(tts_config)
|
||
|
||
APP = FastAPI()
|
||
|
||
|
||
class TTS_Request(BaseModel):
|
||
text: str = None
|
||
text_lang: str = None
|
||
ref_audio_path: str = None
|
||
aux_ref_audio_paths: list = None
|
||
prompt_lang: str = None
|
||
prompt_text: str = ""
|
||
top_k: int = 15
|
||
top_p: float = 1
|
||
temperature: float = 1
|
||
text_split_method: str = "cut5"
|
||
batch_size: int = 1
|
||
batch_threshold: float = 0.75
|
||
split_bucket: bool = True
|
||
speed_factor: float = 1.0
|
||
fragment_interval: float = 0.3
|
||
seed: int = -1
|
||
media_type: str = "wav"
|
||
streaming_mode: Union[bool, int] = False
|
||
parallel_infer: bool = True
|
||
repetition_penalty: float = 1.35
|
||
sample_steps: int = 32
|
||
super_sampling: bool = False
|
||
overlap_length: int = 2
|
||
min_chunk_length: int = 16
|
||
|
||
|
||
def pack_ogg(io_buffer: BytesIO, data: np.ndarray, rate: int):
|
||
# Author: AkagawaTsurunaki
|
||
# Issue:
|
||
# Stack overflow probabilistically occurs
|
||
# when the function `sf_writef_short` of `libsndfile_64bit.dll` is called
|
||
# using the Python library `soundfile`
|
||
# Note:
|
||
# This is an issue related to `libsndfile`, not this project itself.
|
||
# It happens when you generate a large audio tensor (about 499804 frames in my PC)
|
||
# and try to convert it to an ogg file.
|
||
# Related:
|
||
# https://github.com/RVC-Boss/GPT-SoVITS/issues/1199
|
||
# https://github.com/libsndfile/libsndfile/issues/1023
|
||
# https://github.com/bastibe/python-soundfile/issues/396
|
||
# Suggestion:
|
||
# Or split the whole audio data into smaller audio segment to avoid stack overflow?
|
||
|
||
def handle_pack_ogg():
|
||
with sf.SoundFile(io_buffer, mode="w", samplerate=rate, channels=1, format="ogg") as audio_file:
|
||
audio_file.write(data)
|
||
|
||
|
||
|
||
# See: https://docs.python.org/3/library/threading.html
|
||
# The stack size of this thread is at least 32768
|
||
# If stack overflow error still occurs, just modify the `stack_size`.
|
||
# stack_size = n * 4096, where n should be a positive integer.
|
||
# Here we chose n = 4096.
|
||
stack_size = 4096 * 4096
|
||
try:
|
||
threading.stack_size(stack_size)
|
||
pack_ogg_thread = threading.Thread(target=handle_pack_ogg)
|
||
pack_ogg_thread.start()
|
||
pack_ogg_thread.join()
|
||
except RuntimeError as e:
|
||
# If changing the thread stack size is unsupported, a RuntimeError is raised.
|
||
print("RuntimeError: {}".format(e))
|
||
print("Changing the thread stack size is unsupported.")
|
||
except ValueError as e:
|
||
# If the specified stack size is invalid, a ValueError is raised and the stack size is unmodified.
|
||
print("ValueError: {}".format(e))
|
||
print("The specified stack size is invalid.")
|
||
|
||
return io_buffer
|
||
|
||
|
||
def pack_raw(io_buffer: BytesIO, data: np.ndarray, rate: int):
|
||
io_buffer.write(data.tobytes())
|
||
return io_buffer
|
||
|
||
|
||
def pack_wav(io_buffer: BytesIO, data: np.ndarray, rate: int):
|
||
io_buffer = BytesIO()
|
||
sf.write(io_buffer, data, rate, format="wav")
|
||
return io_buffer
|
||
|
||
|
||
def pack_aac(io_buffer: BytesIO, data: np.ndarray, rate: int):
|
||
process = subprocess.Popen(
|
||
[
|
||
"ffmpeg",
|
||
"-f",
|
||
"s16le", # 输入16位有符号小端整数PCM
|
||
"-ar",
|
||
str(rate), # 设置采样率
|
||
"-ac",
|
||
"1", # 单声道
|
||
"-i",
|
||
"pipe:0", # 从管道读取输入
|
||
"-c:a",
|
||
"aac", # 音频编码器为AAC
|
||
"-b:a",
|
||
"192k", # 比特率
|
||
"-vn", # 不包含视频
|
||
"-f",
|
||
"adts", # 输出AAC数据流格式
|
||
"pipe:1", # 将输出写入管道
|
||
],
|
||
stdin=subprocess.PIPE,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
)
|
||
out, _ = process.communicate(input=data.tobytes())
|
||
io_buffer.write(out)
|
||
return io_buffer
|
||
|
||
|
||
def pack_audio(io_buffer: BytesIO, data: np.ndarray, rate: int, media_type: str):
|
||
if media_type == "ogg":
|
||
io_buffer = pack_ogg(io_buffer, data, rate)
|
||
elif media_type == "aac":
|
||
io_buffer = pack_aac(io_buffer, data, rate)
|
||
elif media_type == "wav":
|
||
io_buffer = pack_wav(io_buffer, data, rate)
|
||
else:
|
||
io_buffer = pack_raw(io_buffer, data, rate)
|
||
io_buffer.seek(0)
|
||
return io_buffer
|
||
|
||
|
||
# from https://huggingface.co/spaces/coqui/voice-chat-with-mistral/blob/main/app.py
|
||
def wave_header_chunk(frame_input=b"", channels=1, sample_width=2, sample_rate=32000):
|
||
# This will create a wave header then append the frame input
|
||
# It should be first on a streaming wav file
|
||
# Other frames better should not have it (else you will hear some artifacts each chunk start)
|
||
wav_buf = BytesIO()
|
||
with wave.open(wav_buf, "wb") as vfout:
|
||
vfout.setnchannels(channels)
|
||
vfout.setsampwidth(sample_width)
|
||
vfout.setframerate(sample_rate)
|
||
vfout.writeframes(frame_input)
|
||
|
||
wav_buf.seek(0)
|
||
return wav_buf.read()
|
||
|
||
|
||
def handle_control(command: str):
|
||
if command == "restart":
|
||
os.execl(sys.executable, sys.executable, *argv)
|
||
elif command == "exit":
|
||
os.kill(os.getpid(), signal.SIGTERM)
|
||
exit(0)
|
||
|
||
|
||
def check_params(req: dict):
|
||
text: str = req.get("text", "")
|
||
text_lang: str = req.get("text_lang", "")
|
||
ref_audio_path: str = req.get("ref_audio_path", "")
|
||
streaming_mode: bool = req.get("streaming_mode", False)
|
||
media_type: str = req.get("media_type", "wav")
|
||
prompt_lang: str = req.get("prompt_lang", "")
|
||
text_split_method: str = req.get("text_split_method", "cut5")
|
||
|
||
if ref_audio_path in [None, ""]:
|
||
return JSONResponse(status_code=400, content={"message": "ref_audio_path is required"})
|
||
if text in [None, ""]:
|
||
return JSONResponse(status_code=400, content={"message": "text is required"})
|
||
if text_lang in [None, ""]:
|
||
return JSONResponse(status_code=400, content={"message": "text_lang is required"})
|
||
elif text_lang.lower() not in tts_config.languages:
|
||
return JSONResponse(
|
||
status_code=400,
|
||
content={"message": f"text_lang: {text_lang} is not supported in version {tts_config.version}"},
|
||
)
|
||
if prompt_lang in [None, ""]:
|
||
return JSONResponse(status_code=400, content={"message": "prompt_lang is required"})
|
||
elif prompt_lang.lower() not in tts_config.languages:
|
||
return JSONResponse(
|
||
status_code=400,
|
||
content={"message": f"prompt_lang: {prompt_lang} is not supported in version {tts_config.version}"},
|
||
)
|
||
if media_type not in ["wav", "raw", "ogg", "aac"]:
|
||
return JSONResponse(status_code=400, content={"message": f"media_type: {media_type} is not supported"})
|
||
# elif media_type == "ogg" and not streaming_mode:
|
||
# return JSONResponse(status_code=400, content={"message": "ogg format is not supported in non-streaming mode"})
|
||
|
||
if text_split_method not in cut_method_names:
|
||
return JSONResponse(
|
||
status_code=400, content={"message": f"text_split_method:{text_split_method} is not supported"}
|
||
)
|
||
|
||
return None
|
||
|
||
|
||
async def tts_handle(req: dict):
|
||
"""
|
||
Text to speech handler.
|
||
|
||
Args:
|
||
req (dict):
|
||
{
|
||
"text": "", # str.(required) text to be synthesized
|
||
"text_lang: "", # str.(required) language of the text to be synthesized
|
||
"ref_audio_path": "", # str.(required) reference audio path
|
||
"aux_ref_audio_paths": [], # list.(optional) auxiliary reference audio paths for multi-speaker tone fusion
|
||
"prompt_text": "", # str.(optional) prompt text for the reference audio
|
||
"prompt_lang": "", # str.(required) language of the prompt text for the reference audio
|
||
"top_k": 15, # int. top k sampling
|
||
"top_p": 1, # float. top p sampling
|
||
"temperature": 1, # float. temperature for sampling
|
||
"text_split_method": "cut5", # str. text split method, see text_segmentation_method.py for details.
|
||
"batch_size": 1, # int. batch size for inference
|
||
"batch_threshold": 0.75, # float. threshold for batch splitting.
|
||
"split_bucket": True, # bool. whether to split the batch into multiple buckets.
|
||
"speed_factor":1.0, # float. control the speed of the synthesized audio.
|
||
"fragment_interval":0.3, # float. to control the interval of the audio fragment.
|
||
"seed": -1, # int. random seed for reproducibility.
|
||
"parallel_infer": True, # bool. whether to use parallel inference.
|
||
"repetition_penalty": 1.35, # float. repetition penalty for T2S model.
|
||
"sample_steps": 32, # int. number of sampling steps for VITS model V3.
|
||
"super_sampling": False, # bool. whether to use super-sampling for audio when using VITS model V3.
|
||
"streaming_mode": False, # bool or int. return audio chunk by chunk.T he available options are: 0,1,2,3 or True/False (0/False: Disabled | 1/True: Best Quality, Slowest response speed (old version streaming_mode) | 2: Medium Quality, Slow response speed | 3: Lower Quality, Faster response speed )
|
||
"overlap_length": 2, # int. overlap length of semantic tokens for streaming mode.
|
||
"min_chunk_length": 16, # int. The minimum chunk length of semantic tokens for streaming mode. (affects audio chunk size)
|
||
}
|
||
returns:
|
||
StreamingResponse: audio stream response.
|
||
"""
|
||
|
||
streaming_mode = req.get("streaming_mode", False)
|
||
return_fragment = req.get("return_fragment", False)
|
||
media_type = req.get("media_type", "wav")
|
||
|
||
check_res = check_params(req)
|
||
if check_res is not None:
|
||
return check_res
|
||
|
||
if streaming_mode == 0:
|
||
streaming_mode = False
|
||
return_fragment = False
|
||
fixed_length_chunk = False
|
||
elif streaming_mode == 1:
|
||
streaming_mode = False
|
||
return_fragment = True
|
||
fixed_length_chunk = False
|
||
elif streaming_mode == 2:
|
||
streaming_mode = True
|
||
return_fragment = False
|
||
fixed_length_chunk = False
|
||
elif streaming_mode == 3:
|
||
streaming_mode = True
|
||
return_fragment = False
|
||
fixed_length_chunk = True
|
||
|
||
else:
|
||
return JSONResponse(status_code=400, content={"message": f"the value of streaming_mode must be 0, 1, 2, 3(int) or true/false(bool)"})
|
||
|
||
req["streaming_mode"] = streaming_mode
|
||
req["return_fragment"] = return_fragment
|
||
req["fixed_length_chunk"] = fixed_length_chunk
|
||
|
||
print(f"{streaming_mode} {return_fragment} {fixed_length_chunk}")
|
||
|
||
streaming_mode = streaming_mode or return_fragment
|
||
|
||
|
||
try:
|
||
tts_generator = tts_pipeline.run(req)
|
||
|
||
if streaming_mode:
|
||
|
||
def streaming_generator(tts_generator: Generator, media_type: str):
|
||
if_frist_chunk = True
|
||
for sr, chunk in tts_generator:
|
||
if if_frist_chunk and media_type == "wav":
|
||
yield wave_header_chunk(sample_rate=sr)
|
||
media_type = "raw"
|
||
if_frist_chunk = False
|
||
yield pack_audio(BytesIO(), chunk, sr, media_type).getvalue()
|
||
|
||
# _media_type = f"audio/{media_type}" if not (streaming_mode and media_type in ["wav", "raw"]) else f"audio/x-{media_type}"
|
||
return StreamingResponse(
|
||
streaming_generator(
|
||
tts_generator,
|
||
media_type,
|
||
),
|
||
media_type=f"audio/{media_type}",
|
||
)
|
||
|
||
else:
|
||
sr, audio_data = next(tts_generator)
|
||
audio_data = pack_audio(BytesIO(), audio_data, sr, media_type).getvalue()
|
||
return Response(audio_data, media_type=f"audio/{media_type}")
|
||
except Exception as e:
|
||
return JSONResponse(status_code=400, content={"message": "tts failed", "Exception": str(e)})
|
||
|
||
|
||
@APP.get("/control")
|
||
async def control(command: str = None):
|
||
if command is None:
|
||
return JSONResponse(status_code=400, content={"message": "command is required"})
|
||
handle_control(command)
|
||
|
||
|
||
@APP.get("/tts")
|
||
async def tts_get_endpoint(
|
||
text: str = None,
|
||
text_lang: str = None,
|
||
ref_audio_path: str = None,
|
||
aux_ref_audio_paths: list = None,
|
||
prompt_lang: str = None,
|
||
prompt_text: str = "",
|
||
top_k: int = 15,
|
||
top_p: float = 1,
|
||
temperature: float = 1,
|
||
text_split_method: str = "cut5",
|
||
batch_size: int = 1,
|
||
batch_threshold: float = 0.75,
|
||
split_bucket: bool = True,
|
||
speed_factor: float = 1.0,
|
||
fragment_interval: float = 0.3,
|
||
seed: int = -1,
|
||
media_type: str = "wav",
|
||
parallel_infer: bool = True,
|
||
repetition_penalty: float = 1.35,
|
||
sample_steps: int = 32,
|
||
super_sampling: bool = False,
|
||
streaming_mode: Union[bool, int] = False,
|
||
overlap_length: int = 2,
|
||
min_chunk_length: int = 16,
|
||
):
|
||
req = {
|
||
"text": text,
|
||
"text_lang": text_lang.lower(),
|
||
"ref_audio_path": ref_audio_path,
|
||
"aux_ref_audio_paths": aux_ref_audio_paths,
|
||
"prompt_text": prompt_text,
|
||
"prompt_lang": prompt_lang.lower(),
|
||
"top_k": top_k,
|
||
"top_p": top_p,
|
||
"temperature": temperature,
|
||
"text_split_method": text_split_method,
|
||
"batch_size": int(batch_size),
|
||
"batch_threshold": float(batch_threshold),
|
||
"speed_factor": float(speed_factor),
|
||
"split_bucket": split_bucket,
|
||
"fragment_interval": fragment_interval,
|
||
"seed": seed,
|
||
"media_type": media_type,
|
||
"streaming_mode": streaming_mode,
|
||
"parallel_infer": parallel_infer,
|
||
"repetition_penalty": float(repetition_penalty),
|
||
"sample_steps": int(sample_steps),
|
||
"super_sampling": super_sampling,
|
||
"overlap_length": int(overlap_length),
|
||
"min_chunk_length": int(min_chunk_length),
|
||
}
|
||
return await tts_handle(req)
|
||
|
||
|
||
@APP.post("/tts")
|
||
async def tts_post_endpoint(request: TTS_Request):
|
||
req = request.dict()
|
||
return await tts_handle(req)
|
||
|
||
|
||
@APP.get("/set_refer_audio")
|
||
async def set_refer_aduio(refer_audio_path: str = None):
|
||
try:
|
||
tts_pipeline.set_ref_audio(refer_audio_path)
|
||
except Exception as e:
|
||
return JSONResponse(status_code=400, content={"message": "set refer audio failed", "Exception": str(e)})
|
||
return JSONResponse(status_code=200, content={"message": "success"})
|
||
|
||
|
||
# @APP.post("/set_refer_audio")
|
||
# async def set_refer_aduio_post(audio_file: UploadFile = File(...)):
|
||
# try:
|
||
# # 检查文件类型，确保是音频文件
|
||
# if not audio_file.content_type.startswith("audio/"):
|
||
# return JSONResponse(status_code=400, content={"message": "file type is not supported"})
|
||
|
||
# os.makedirs("uploaded_audio", exist_ok=True)
|
||
# save_path = os.path.join("uploaded_audio", audio_file.filename)
|
||
# # 保存音频文件到服务器上的一个目录
|
||
# with open(save_path , "wb") as buffer:
|
||
# buffer.write(await audio_file.read())
|
||
|
||
# tts_pipeline.set_ref_audio(save_path)
|
||
# except Exception as e:
|
||
# return JSONResponse(status_code=400, content={"message": f"set refer audio failed", "Exception": str(e)})
|
||
# return JSONResponse(status_code=200, content={"message": "success"})
|
||
|
||
|
||
@APP.get("/set_gpt_weights")
|
||
async def set_gpt_weights(weights_path: str = None):
|
||
try:
|
||
if weights_path in ["", None]:
|
||
return JSONResponse(status_code=400, content={"message": "gpt weight path is required"})
|
||
tts_pipeline.init_t2s_weights(weights_path)
|
||
except Exception as e:
|
||
return JSONResponse(status_code=400, content={"message": "change gpt weight failed", "Exception": str(e)})
|
||
|
||
return JSONResponse(status_code=200, content={"message": "success"})
|
||
|
||
|
||
@APP.get("/set_sovits_weights")
|
||
async def set_sovits_weights(weights_path: str = None):
|
||
try:
|
||
if weights_path in ["", None]:
|
||
return JSONResponse(status_code=400, content={"message": "sovits weight path is required"})
|
||
tts_pipeline.init_vits_weights(weights_path)
|
||
except Exception as e:
|
||
return JSONResponse(status_code=400, content={"message": "change sovits weight failed", "Exception": str(e)})
|
||
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 _read_emotion_map(model_name: str) -> dict[str, str]:
|
||
"""读取训练数据的 emotion，返回 {audio_name|stem: emotion}。
|
||
|
||
兼容新旧两种 emotion 数据源:
|
||
1. 旧: ASR 标注 .list 第5列 (output/asr_opt/<model_name>/*.list)
|
||
2. 新: logs/<model_name>/audio_metadata.json (推理WebUI回写的元数据)
|
||
合并两者, audio_metadata.json 优先(推理时回写的更新)。
|
||
"""
|
||
import json
|
||
from tools.list_metadata import parse_list_line
|
||
|
||
emotion_map: dict[str, str] = {}
|
||
|
||
def _add(name: str, emotion: str):
|
||
if name and emotion:
|
||
emotion_map[name] = emotion
|
||
emotion_map[Path(name).stem] = emotion
|
||
|
||
# 源1: .list (旧, ASR 标注)
|
||
list_dir = Path("output") / "asr_opt" / model_name
|
||
if list_dir.exists():
|
||
for list_path in sorted(list_dir.glob("*.list")):
|
||
try:
|
||
for line in list_path.read_text(encoding="utf-8", errors="ignore").splitlines():
|
||
item = parse_list_line(line)
|
||
if item is None or not item.emotion:
|
||
continue
|
||
_add(os.path.basename(item.wav_path.strip("\"'")), item.emotion)
|
||
except Exception:
|
||
continue
|
||
|
||
# 源2: audio_metadata.json (新, 推理WebUI回写; 优先级高, 覆盖源1)
|
||
meta_path = Path(exp_root) / model_name / "audio_metadata.json"
|
||
if meta_path.exists():
|
||
try:
|
||
meta = json.loads(meta_path.read_text(encoding="utf-8", errors="ignore"))
|
||
for name, info in meta.items():
|
||
if isinstance(info, dict) and info.get("emotion"):
|
||
_add(name, info["emotion"])
|
||
except Exception:
|
||
pass
|
||
|
||
return emotion_map
|
||
|
||
|
||
|
||
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_weights、GPT_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)
|
||
emotion_map = _read_emotion_map(model_name)
|
||
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)
|
||
emotion = emotion_map.get(f.name, "") or emotion_map.get(f.stem, "")
|
||
samples.append({
|
||
"audio_name": f.name,
|
||
"audio_path": rel.replace(os.sep, "/"),
|
||
"text": entry["text"],
|
||
"lang": entry["lang"],
|
||
"emotion": emotion,
|
||
})
|
||
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监听双栈
|
||
host = None
|
||
uvicorn.run(app=APP, host=host, port=port, workers=1)
|
||
except Exception:
|
||
traceback.print_exc()
|
||
os.kill(os.getpid(), signal.SIGTERM)
|
||
exit(0)
|