app.py

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)