import os
|
import uuid
|
import edge_tts
|
from datetime import datetime
|
from core.providers.tts.base import TTSProviderBase
|
|
|
class TTSProvider(TTSProviderBase):
|
TTS_PARAM_CONFIG = [
|
("ttsVolume", "volume", 0, 100, 50, int),
|
("ttsRate", "speech_rate", -100, 100, 0, int),
|
("ttsPitch", "pitch_rate", -100, 100, 0, int),
|
]
|
|
def __init__(self, config, delete_audio_file):
|
super().__init__(config, delete_audio_file)
|
if config.get("private_voice"):
|
self.voice = config.get("private_voice")
|
else:
|
self.voice = config.get("voice")
|
self.audio_file_type = config.get("format", "mp3")
|
|
volume = config.get("volume", "50")
|
self.volume = int(volume) if volume else 50
|
|
speech_rate = config.get("rate", "0")
|
self.speech_rate = int(speech_rate) if speech_rate else 0
|
|
pitch_rate = config.get("pitch", "0")
|
self.pitch_rate = int(pitch_rate) if pitch_rate else 0
|
|
# 应用百分比调整
|
self._apply_percentage_params(config)
|
|
self.edge_rate = f"{self.speech_rate:+}%"
|
self.edge_volume = f"{self.volume:+}%"
|
self.edge_pitch = f"{self.pitch_rate:+}Hz"
|
|
def generate_filename(self, extension=".mp3"):
|
return os.path.join(
|
self.output_file,
|
f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}",
|
)
|
|
async def text_to_speak(self, text, output_file):
|
try:
|
communicate = edge_tts.Communicate(
|
text,
|
voice=self.voice,
|
rate=self.edge_rate,
|
volume=self.edge_volume,
|
pitch=self.edge_pitch,
|
)
|
if output_file:
|
# 确保目录存在并创建空文件
|
os.makedirs(os.path.dirname(output_file), exist_ok=True)
|
with open(output_file, "wb") as f:
|
pass
|
|
# 流式写入音频数据
|
with open(output_file, "ab") as f: # 改为追加模式避免覆盖
|
async for chunk in communicate.stream():
|
if chunk["type"] == "audio": # 只处理音频数据块
|
f.write(chunk["data"])
|
else:
|
# 返回音频二进制数据
|
audio_bytes = b""
|
async for chunk in communicate.stream():
|
if chunk["type"] == "audio":
|
audio_bytes += chunk["data"]
|
return audio_bytes
|
except Exception as e:
|
error_msg = f"Edge TTS请求失败: {e}"
|
raise Exception(error_msg) # 抛出异常，让调用方捕获
|
