Files
auto_agent/llm_server/tts_server.py
T

97 lines
3.2 KiB
Python

#!/usr/bin/env python3
"""
Genie TTS 极简服务 —— 单文本直接合成,无分块无内部线程池。
配合客户端分块 + 多 worker 实现高并发。
"""
import os
# 自动获取脚本所在目录作为基础路径
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
os.environ["GENIE_DATA_DIR"] = os.path.join(BASE_DIR, "models", "genie_tts", "GenieData")
import tempfile
from contextlib import asynccontextmanager
from fastapi import FastAPI, Query
from fastapi.responses import Response
import genie_tts as genie
# ======================== 配置区域 ========================
DEFAULT_CHARACTER = "feibi"
DEFAULT_ONNX_MODEL_DIR = os.path.join(BASE_DIR, "models", "genie_tts", "CharacterModels", "v2ProPlus", "feibi", "tts_models")
DEFAULT_REFERENCE_AUDIO = os.path.join(BASE_DIR, "models", "genie_tts", "CharacterModels", "v2ProPlus", "feibi", "prompt_wav", "zh_vo_Main_Linaxita_2_1_10_26.wav")
DEFAULT_REFERENCE_TEXT = "在此之前,请您务必继续享受旅居拉古那的时光。"
DEFAULT_LANGUAGE = "zh"
TMP_DIR = tempfile.gettempdir() # 临时文件存放目录
# =========================================================
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期:加载模型,卸载模型"""
print("🚀 正在启动 Genie TTS 极简服务...")
try:
genie.load_character(
character_name=DEFAULT_CHARACTER,
onnx_model_dir=DEFAULT_ONNX_MODEL_DIR,
language=DEFAULT_LANGUAGE
)
genie.set_reference_audio(
character_name=DEFAULT_CHARACTER,
audio_path=DEFAULT_REFERENCE_AUDIO,
audio_text=DEFAULT_REFERENCE_TEXT,
language=DEFAULT_LANGUAGE
)
print("✅ Genie TTS 模型加载成功!")
print("👉 调用方式:http://localhost:8902/tts?text=你要说的话")
except Exception as e:
print(f"❌ 服务启动失败: {e}")
raise e
yield
print("\n🛑 正在关闭服务...")
try:
genie.unload_character(DEFAULT_CHARACTER)
print("✅ 资源已释放")
except:
pass
app = FastAPI(lifespan=lifespan)
@app.get("/tts")
async def text_to_speech(text: str = Query(..., description="待合成的文本")):
"""接收任意长度文本,直接合成并返回 WAV 音频。"""
if not text.strip():
return Response(content="文本不能为空", status_code=400, media_type="text/plain")
# 生成临时文件,调用 genie 合成
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False, dir=TMP_DIR)
tmp_path = tmp.name
tmp.close()
try:
genie.tts(
character_name=DEFAULT_CHARACTER,
text=text,
play=False,
save_path=tmp_path
)
with open(tmp_path, "rb") as f:
audio_bytes = f.read()
return Response(content=audio_bytes, media_type="audio/wav")
except Exception as e:
import traceback
traceback.print_exc()
return Response(content=f"生成失败: {e}", status_code=500, media_type="text/plain")
finally:
if os.path.exists(tmp_path):
os.unlink(tmp_path)
@app.get("/health")
async def health():
return {"status": "ok"}