File size: 2,117 Bytes
a8dead5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 | from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import edge_tts
import asyncio
app = FastAPI(title="Edge TTS API", description="基于Edge TTS的文本转语音API", version="1.0")
# 添加CORS支持
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 允许所有来源，生产环境中应限制为特定域名
allow_credentials=True,
allow_methods=["*"], # 允许所有HTTP方法
allow_headers=["*"], # 允许所有HTTP头
)
class TTSRequest(BaseModel):
text: str
voice: str = "zh-CN-YunxiNeural"
rate: str = "+0%"
volume: str = "+0%"
@app.post("/tts", response_class=StreamingResponse, summary="文本转语音")
async def text_to_speech(request: TTSRequest):
try:
communicate = edge_tts.Communicate(request.text, request.voice, rate=request.rate, volume=request.volume)
# 先将所有音频数据收集到内存中
audio_data = b""
async for chunk in communicate.stream():
if chunk["type"] == "audio":
audio_data += chunk["data"]
# 然后返回完整的音频数据
return StreamingResponse(iter([audio_data]), media_type="audio/mpeg")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/voices", summary="获取可用语音列表")
async def get_voices():
try:
voices = await edge_tts.list_voices()
return [{
"short_name": voice.get("ShortName", ""),
"friendly_name": voice.get("FriendlyName", voice.get("ShortName", "")),
"gender": voice.get("Gender", ""),
"locale": voice.get("Locale", "")
} for voice in voices]
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/", summary="首页")
def read_root():
return {"message": "欢迎使用Edge TTS API", "docs": "/docs"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)
|