257 lines
8.4 KiB
Python
257 lines
8.4 KiB
Python
"""
|
||
Edge TTS 服务 v2.0 - 免费开源中文语音合成，支持背景音乐混音
|
||
"""
|
||
import asyncio
|
||
import tempfile
|
||
import os
|
||
from pathlib import Path
|
||
from fastapi import FastAPI, HTTPException
|
||
from fastapi.responses import FileResponse, StreamingResponse
|
||
from pydantic import BaseModel
|
||
import edge_tts
|
||
|
||
# pydub 混音支持（必须在 import pydub 之前将 ffmpeg 注入 PATH）
|
||
_FFMPEG_BIN = r"C:\Users\maxf\AppData\Local\Microsoft\WinGet\Packages\Gyan.FFmpeg_Microsoft.Winget.Source_8wekyb3d8bbwe\ffmpeg-8.1-full_build\bin"
|
||
if os.path.isdir(_FFMPEG_BIN) and _FFMPEG_BIN not in os.environ.get("PATH", ""):
|
||
os.environ["PATH"] = _FFMPEG_BIN + os.pathsep + os.environ.get("PATH", "")
|
||
|
||
try:
|
||
from pydub import AudioSegment
|
||
PYDUB_AVAILABLE = True
|
||
except ImportError:
|
||
PYDUB_AVAILABLE = False
|
||
|
||
app = FastAPI(title="Edge TTS Server", version="2.0")
|
||
|
||
# BGM 目录
|
||
BGM_DIR = Path(__file__).parent / "bgm"
|
||
BGM_DIR.mkdir(exist_ok=True)
|
||
|
||
# 中文常用语音
|
||
VOICES = {
|
||
"xiaoxiao": "zh-CN-XiaoxiaoNeural", # 晓晓 女声 温柔
|
||
"xiaoyi": "zh-CN-XiaoyiNeural", # 小艺 女声 活泼
|
||
"yunjian": "zh-CN-YunjianNeural", # 云健 男声 新闻
|
||
"yunxi": "zh-CN-YunxiNeural", # 云希 男声 年轻
|
||
"xiaochen": "zh-CN-XiaochenNeural", # 晓晨 女声 成熟
|
||
}
|
||
|
||
DEFAULT_VOICE = "zh-CN-XiaoxiaoNeural"
|
||
|
||
|
||
class TTSRequest(BaseModel):
|
||
text: str
|
||
voice: str = "xiaoxiao" # 语音名称
|
||
rate: str = "+0%" # 语速，如 "+10%" 或 "-10%"
|
||
volume: str = "+0%" # 音量
|
||
bgm: str = "" # 背景音乐文件名（bgm/ 目录下），留空则不混音
|
||
bgm_volume: float = 0.3 # 背景音乐音量比例（0.0~1.0，默认 0.3）
|
||
|
||
|
||
@app.get("/")
|
||
def root():
|
||
return {
|
||
"message": "Edge TTS Server v2.0",
|
||
"voices": list(VOICES.keys()),
|
||
"bgm_support": PYDUB_AVAILABLE,
|
||
"usage": "POST /tts with {text, voice?, rate?, volume?, bgm?, bgm_volume?}"
|
||
}
|
||
|
||
|
||
@app.get("/voices")
|
||
def list_voices():
|
||
"""列出可用语音"""
|
||
return {
|
||
"voices": {
|
||
k: {"id": v, "description": get_voice_desc(k)}
|
||
for k, v in VOICES.items()
|
||
}
|
||
}
|
||
|
||
|
||
@app.get("/bgm")
|
||
def list_bgm():
|
||
"""列出可用背景音乐"""
|
||
if not BGM_DIR.exists():
|
||
return {"bgm": []}
|
||
files = [f.name for f in BGM_DIR.iterdir()
|
||
if f.suffix.lower() in (".mp3", ".wav", ".ogg", ".flac")]
|
||
return {"bgm": files, "directory": str(BGM_DIR)}
|
||
|
||
|
||
def get_voice_desc(key: str) -> str:
|
||
desc = {
|
||
"xiaoxiao": "晓晓-女声-温柔自然",
|
||
"xiaoyi": "小艺-女声-活泼轻快",
|
||
"yunjian": "云健-男声-新闻播报",
|
||
"yunxi": "云希-男声-年轻活力",
|
||
"xiaochen": "晓晨-女声-成熟稳重",
|
||
}
|
||
return desc.get(key, "未知")
|
||
|
||
|
||
def mix_with_bgm(tts_path: str, bgm_name: str, bgm_volume: float) -> str:
|
||
"""将 TTS 音频与背景音乐混音，返回混音后的临时文件路径"""
|
||
bgm_path = BGM_DIR / bgm_name
|
||
if not bgm_path.exists():
|
||
raise FileNotFoundError(f"背景音乐文件不存在: {bgm_name}")
|
||
|
||
# 加载 TTS 音频
|
||
tts_audio = AudioSegment.from_mp3(tts_path)
|
||
tts_duration = len(tts_audio)
|
||
|
||
# 加载 BGM 并调整音量
|
||
bgm_audio = AudioSegment.from_file(str(bgm_path))
|
||
# bgm_volume: 0.0~1.0 → 转换为 dB 调整（0.3 ≈ -10dB）
|
||
volume_db = 20 * (bgm_volume ** 0.5 - 1) * 5 if bgm_volume > 0 else -60
|
||
bgm_audio = bgm_audio + volume_db
|
||
|
||
# 若 BGM 比 TTS 短则循环，截取与 TTS 等长
|
||
if len(bgm_audio) < tts_duration:
|
||
loops = (tts_duration // len(bgm_audio)) + 1
|
||
bgm_audio = bgm_audio * loops
|
||
bgm_audio = bgm_audio[:tts_duration]
|
||
|
||
# 淡入淡出（前后各 2 秒，使过渡自然）
|
||
fade_ms = min(2000, tts_duration // 4)
|
||
bgm_audio = bgm_audio.fade_in(fade_ms).fade_out(fade_ms)
|
||
|
||
# 混音叠加
|
||
mixed = tts_audio.overlay(bgm_audio)
|
||
|
||
# 输出到临时文件
|
||
with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as out:
|
||
out_path = out.name
|
||
mixed.export(out_path, format="mp3", bitrate="128k")
|
||
|
||
return out_path
|
||
|
||
|
||
async def generate_tts_with_retry(text: str, voice_id: str, rate: str, volume: str, max_retries: int = 3):
|
||
"""带重试机制的 TTS 生成函数"""
|
||
last_error = None
|
||
tmp_path = None
|
||
|
||
for attempt in range(max_retries):
|
||
try:
|
||
communicate = edge_tts.Communicate(
|
||
text=text,
|
||
voice=voice_id,
|
||
rate=rate,
|
||
volume=volume,
|
||
)
|
||
|
||
with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp:
|
||
tmp_path = tmp.name
|
||
|
||
await communicate.save(tmp_path)
|
||
return tmp_path
|
||
|
||
except Exception as e:
|
||
last_error = e
|
||
error_msg = str(e)
|
||
|
||
if "403" in error_msg or "Invalid response status" in error_msg:
|
||
if attempt < max_retries - 1:
|
||
wait_time = 2 ** attempt
|
||
print(f"第 {attempt + 1} 次尝试失败，{wait_time}秒后重试...")
|
||
await asyncio.sleep(wait_time)
|
||
continue
|
||
|
||
if tmp_path and os.path.exists(tmp_path):
|
||
os.unlink(tmp_path)
|
||
raise last_error
|
||
|
||
raise last_error
|
||
|
||
|
||
@app.post("/tts")
|
||
async def text_to_speech(req: TTSRequest):
|
||
"""文字转语音，返回 MP3 文件（可选背景音乐混音）"""
|
||
if not req.text or len(req.text) > 5000:
|
||
raise HTTPException(status_code=400, detail="文本不能为空或超过5000字")
|
||
|
||
voice_id = VOICES.get(req.voice, DEFAULT_VOICE)
|
||
tmp_path = None
|
||
mixed_path = None
|
||
|
||
try:
|
||
tmp_path = await generate_tts_with_retry(
|
||
text=req.text,
|
||
voice_id=voice_id,
|
||
rate=req.rate,
|
||
volume=req.volume,
|
||
max_retries=3
|
||
)
|
||
|
||
# 如果指定了 BGM 且 pydub 可用，进行混音
|
||
if req.bgm and PYDUB_AVAILABLE:
|
||
try:
|
||
mixed_path = mix_with_bgm(tmp_path, req.bgm, req.bgm_volume)
|
||
os.unlink(tmp_path) # 删除纯 TTS 临时文件
|
||
output_path = mixed_path
|
||
except FileNotFoundError as e:
|
||
raise HTTPException(status_code=404, detail=str(e))
|
||
except Exception as e:
|
||
# 混音失败时回退到纯 TTS
|
||
print(f"混音失败，回退到纯TTS: {e}")
|
||
output_path = tmp_path
|
||
elif req.bgm and not PYDUB_AVAILABLE:
|
||
raise HTTPException(status_code=503, detail="pydub 未安装，无法进行混音。请运行: pip install pydub")
|
||
else:
|
||
output_path = tmp_path
|
||
|
||
return FileResponse(
|
||
output_path,
|
||
media_type="audio/mpeg",
|
||
filename="tts_output.mp3",
|
||
background=None
|
||
)
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
if "403" in error_msg or "Invalid response status" in error_msg:
|
||
raise HTTPException(
|
||
status_code=503,
|
||
detail="TTS 服务暂时不可用（微软服务返回 403），请稍后重试。如问题持续，请更新：pip install --upgrade edge-tts"
|
||
)
|
||
raise HTTPException(status_code=500, detail=f"TTS失败: {error_msg}")
|
||
|
||
|
||
@app.post("/tts/stream")
|
||
async def text_to_speech_stream(req: TTSRequest):
|
||
"""流式返回音频（边生成边播放，不支持 BGM 混音）"""
|
||
if not req.text or len(req.text) > 5000:
|
||
raise HTTPException(status_code=400, detail="文本不能为空或超过5000字")
|
||
|
||
voice_id = VOICES.get(req.voice, DEFAULT_VOICE)
|
||
|
||
async def audio_stream():
|
||
communicate = edge_tts.Communicate(
|
||
text=req.text,
|
||
voice=voice_id,
|
||
rate=req.rate,
|
||
volume=req.volume,
|
||
)
|
||
async for chunk in communicate.stream():
|
||
if chunk["type"] == "audio":
|
||
yield chunk["data"]
|
||
|
||
return StreamingResponse(
|
||
audio_stream(),
|
||
media_type="audio/mpeg",
|
||
)
|
||
|
||
|
||
@app.get("/tts/get")
|
||
async def tts_get(text: str, voice: str = "xiaoxiao", rate: str = "+0%", volume: str = "+0%"):
|
||
"""GET 方式调用（方便测试）"""
|
||
req = TTSRequest(text=text, voice=voice, rate=rate, volume=volume)
|
||
return await text_to_speech(req)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
uvicorn.run(app, host="0.0.0.0", port=8000)
|