torin
b8d3cb4644
Some checks are pending
Build and Push Docker Images / changes (push) Waiting to run
Build and Push Docker Images / build-backend (push) Blocked by required conditions
Build and Push Docker Images / build-frontend (push) Blocked by required conditions
Build and Push Docker Images / build-admin-frontend (push) Blocked by required conditions
67 lines
1.9 KiB
Python
67 lines
1.9 KiB
Python
"""EdgeTTS 免费语音生成适配器。"""
|
||
|
||
import time
|
||
|
||
import edge_tts
|
||
|
||
from app.core.logging import get_logger
|
||
from app.services.adapters.base import BaseAdapter
|
||
from app.services.adapters.registry import AdapterRegistry
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
# 默认中文女声 (晓晓)
|
||
DEFAULT_VOICE = "zh-CN-XiaoxiaoNeural"
|
||
|
||
|
||
@AdapterRegistry.register("tts", "edge_tts")
|
||
class EdgeTTSAdapter(BaseAdapter[bytes]):
|
||
"""EdgeTTS 语音生成适配器 (Free)。
|
||
|
||
不需要 API Key。
|
||
"""
|
||
|
||
adapter_type = "tts"
|
||
adapter_name = "edge_tts"
|
||
|
||
async def execute(self, text: str, **kwargs) -> bytes:
|
||
"""生成语音。"""
|
||
# 支持动态指定音色
|
||
voice = kwargs.get("voice") or self.config.model or DEFAULT_VOICE
|
||
|
||
start_time = time.time()
|
||
logger.info("edge_tts_generate_start", text_length=len(text), voice=voice)
|
||
|
||
# EdgeTTS 只能输出到文件，我们需要用临时文件周转一下
|
||
# 或者直接 capture stream (communicate) 但 edge-tts 库主要面向文件
|
||
|
||
# 优化: 使用 communicate 直接获取 bytes，无需磁盘IO
|
||
communicate = edge_tts.Communicate(text, voice)
|
||
|
||
audio_data = b""
|
||
async for chunk in communicate.stream():
|
||
if chunk["type"] == "audio":
|
||
audio_data += chunk["data"]
|
||
|
||
elapsed = time.time() - start_time
|
||
logger.info(
|
||
"edge_tts_generate_success",
|
||
elapsed_seconds=round(elapsed, 2),
|
||
audio_size_bytes=len(audio_data),
|
||
)
|
||
|
||
return audio_data
|
||
|
||
async def health_check(self) -> bool:
|
||
"""检查 EdgeTTS 是否可用 (网络连通性)。"""
|
||
try:
|
||
# 简单生成一个词
|
||
await self.execute("Hi")
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
@property
|
||
def estimated_cost(self) -> float:
|
||
return 0.0 # Free!
|