import os import uuid import json import time import queue import asyncio import traceback import websockets from asyncio import Task from typing import Callable, Any from config.logger import setup_logging from core.utils.tts import MarkdownCleaner from core.providers.tts.base import TTSProviderBase from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType TAG = __name__ logger = setup_logging() class TTSProvider(TTSProviderBase): TTS_PARAM_CONFIG = [ ("ttsVolume", "volume", 0, 100, 50, int), ("ttsRate", "rate", 0.5, 2.0, 1.0, lambda v: round(v, 1)), ("ttsPitch", "pitch", 0.5, 2.0, 1.0, lambda v: round(v, 1)), ] def __init__(self, config, delete_audio_file): super().__init__(config, delete_audio_file) self.interface_type = InterfaceType.DUAL_STREAM # 基础配置 self.api_key = config.get("api_key") if not self.api_key: raise ValueError("api_key is required for CosyVoice TTS") self.report_on_last = True # WebSocket配置 self.ws_url = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/" self.ws = None self._monitor_task = None self.activate_session = False self.last_active_time = None # 模型和音色配置 self.model = config.get("model", "cosyvoice-v2") self.voice = config.get("voice", "longxiaochun_v2") # 默认音色 if config.get("private_voice"): self.voice = config.get("private_voice") # 音频参数配置 self.format = config.get("format", "pcm") volume = config.get("volume", "50") self.volume = int(volume) if volume else 50 rate = config.get("rate", "1.0") self.rate = float(rate) if rate else 1.0 pitch = config.get("pitch", "1.0") self.pitch = float(pitch) if pitch else 1.0 # 应用百分比调整（如果存在），否则使用公有化配置 self._apply_percentage_params(config) self.header = { "Authorization": f"Bearer {self.api_key}", # "user-agent": "your_platform_info", // 可选 # "X-DashScope-WorkSpace": workspace, // 可选，阿里云百炼业务空间ID "X-DashScope-DataInspection": "enable", } async def _ensure_connection(self): """确保WebSocket连接可用，支持60秒内连接复用""" try: current_time = time.time() if self.ws and current_time - self.last_active_time < 60: # 一分钟内才可以复用链接进行连续对话 logger.bind(tag=TAG).debug(f"使用已有链接...") return self.ws logger.bind(tag=TAG).debug("开始建立新连接...") # 建立新连接前取消旧监听任务 await self._cancel_monitor_task() self.ws = await websockets.connect( self.ws_url, additional_headers=self.header, ping_interval=30, ping_timeout=10, close_timeout=10, ) logger.bind(tag=TAG).debug("WebSocket连接建立成功") self.last_active_time = current_time return self.ws except Exception as e: logger.bind(tag=TAG).error(f"建立连接失败: {str(e)}") self.ws = None self.last_active_time = None raise def tts_text_priority_thread(self): """流式TTS文本处理线程""" while not self.conn.stop_event.is_set(): try: message = self.tts_text_queue.get(timeout=1) if self.conn.client_abort: try: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") asyncio.run_coroutine_threadsafe( self.finish_session(self.conn.sentence_id), loop=self.conn.loop, ) continue except Exception as e: logger.bind(tag=TAG).error(f"取消TTS会话失败: {str(e)}") continue # 过滤旧消息：检查sentence_id是否匹配 if message.sentence_id != self.conn.sentence_id: continue logger.bind(tag=TAG).debug( f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}" ) if message.sentence_type == SentenceType.FIRST: # 重置流式处理状态 self.reset_stream_state() # 初始化会话 try: if not getattr(self.conn, "sentence_id", None): self.conn.sentence_id = uuid.uuid4().hex logger.bind(tag=TAG).debug(f"自动生成新的 会话ID: {self.conn.sentence_id}") logger.bind(tag=TAG).debug("开始启动TTS会话...") future = asyncio.run_coroutine_threadsafe( self.start_session(self.conn.sentence_id), loop=self.conn.loop, ) future.result(timeout=self.tts_timeout) self.before_stop_play_files.clear() logger.bind(tag=TAG).debug("TTS会话启动成功") except Exception as e: logger.bind(tag=TAG).error(f"启动TTS会话失败: {str(e)}") continue elif ContentType.TEXT == message.content_type: if message.content_detail: try: logger.bind(tag=TAG).debug( f"开始发送TTS文本: {message.content_detail}" ) future = asyncio.run_coroutine_threadsafe( self.text_to_speak(message.content_detail, None), loop=self.conn.loop, ) future.result(timeout=self.tts_timeout) except Exception as e: logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}") continue elif ContentType.FILE == message.content_type: logger.bind(tag=TAG).info( f"添加音频文件到待播放列表: {message.content_file}" ) if message.content_file and os.path.exists(message.content_file): # 先处理文件音频数据 self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail)) if message.sentence_type == SentenceType.LAST: try: logger.bind(tag=TAG).debug("开始结束TTS会话...") future = asyncio.run_coroutine_threadsafe( self.finish_session(self.conn.sentence_id), loop=self.conn.loop, ) future.result() except Exception as e: logger.bind(tag=TAG).error(f"结束TTS会话失败: {str(e)}") continue except queue.Empty: continue except Exception as e: logger.bind(tag=TAG).error( f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" ) continue async def text_to_speak(self, text, _): """发送文本到TTS服务进行合成""" try: if self.ws is None: logger.bind(tag=TAG).warning("WebSocket连接不存在，终止发送文本") return # 过滤Markdown filtered_text = MarkdownCleaner.clean_markdown(text) if filtered_text: # 使用滑动窗口匹配处理跨分片的替换词 confirmed_texts, self._pending_prefix = self._match_stream_text(filtered_text) # 发送每个确定的文本片段 for txt in confirmed_texts: if txt and self.ws: continue_task_message = { "header": { "action": "continue-task", "task_id": self.conn.sentence_id, "streaming": "duplex", }, "payload": {"input": {"text": txt}}, } await self.ws.send(json.dumps(continue_task_message)) self.last_active_time = time.time() return except Exception as e: logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}") if self.ws: try: await self.ws.close() except: pass self.ws = None raise async def start_session(self, session_id): """启动TTS会话""" logger.bind(tag=TAG).debug(f"开始会话～～{session_id}") try: # 上个会话处于激活状态时关闭上个连接新建链接 if self.activate_session: await self.close() # 设置会话激活标志 self.activate_session = True # 确保连接可用 await self._ensure_connection() # 启动监听任务 if self._monitor_task is None or self._monitor_task.done(): logger.bind(tag=TAG).debug("启动监听任务...") self._monitor_task = asyncio.create_task(self._start_monitor_tts_response()) # 发送run-task消息启动会话 run_task_message = { "header": { "action": "run-task", "task_id": session_id, "streaming": "duplex", }, "payload": { "task_group": "audio", "task": "tts", "function": "SpeechSynthesizer", "model": self.model, "parameters": { "text_type": "PlainText", "voice": self.voice, "format": self.format, "sample_rate": self.conn.sample_rate, "volume": self.volume, "rate": self.rate, "pitch": self.pitch, }, "input": {} }, } await self.ws.send(json.dumps(run_task_message)) self.last_active_time = time.time() logger.bind(tag=TAG).debug("会话启动请求已发送") except Exception as e: logger.bind(tag=TAG).error(f"启动会话失败: {str(e)}") await self.close() raise async def finish_session(self, session_id): """结束TTS会话""" logger.bind(tag=TAG).debug(f"关闭会话～～{session_id}") try: if self.ws and session_id: # 发送finish-task消息 finish_task_message = { "header": { "action": "finish-task", "task_id": session_id, "streaming": "duplex", }, "payload": { "input": {} } } await self.ws.send(json.dumps(finish_task_message)) self.last_active_time = time.time() except Exception as e: logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}") await self.close() raise async def close(self): """清理资源""" await super().close() self.activate_session = False await self._cancel_monitor_task() # 关闭WebSocket连接 if self.ws: try: await self.ws.close() except: pass self.ws = None self.last_active_time = None async def _cancel_monitor_task(self): """取消监听任务""" if self._monitor_task and not self._monitor_task.done(): self._monitor_task.cancel() try: await self._monitor_task except asyncio.CancelledError: pass except Exception as e: logger.bind(tag=TAG).warning(f"取消监听任务错误: {e}") self._monitor_task = None async def _start_monitor_tts_response(self): """监听TTS响应 - 长期运行""" try: while not self.conn.stop_event.is_set(): try: msg = await self.ws.recv() self.last_active_time = time.time() if isinstance(msg, str): # JSON控制消息 try: data = json.loads(msg) header = data.get("header", {}) event = header.get("event") task_id = header.get("task_id") # 只处理当前活跃会话的响应 if task_id and self.conn.sentence_id != task_id: if event in ["task-finished", "task-failed"]: logger.bind(tag=TAG).debug(f"收到残余下行结束响应重置会话状态～～") self.activate_session = False continue if event == "task-started": logger.bind(tag=TAG).debug("TTS任务启动成功~") self.tts_audio_queue.put((SentenceType.FIRST, [], None)) elif event == "result-generated": # 发送缓存的数据 tts_text = self.get_tts_text(self.conn.sentence_id) if tts_text: logger.bind(tag=TAG).info( f"句子语音生成成功： {tts_text}" ) self.tts_audio_queue.put( (SentenceType.FIRST, [], tts_text) ) self.clear_tts_text(self.conn.sentence_id) elif event == "task-finished": logger.bind(tag=TAG).debug("TTS任务完成~") self.activate_session = False self._process_before_stop_play_files() elif event == "task-failed": error_code = header.get("error_code", "unknown") error_message = header.get("error_message", "未知错误") logger.bind(tag=TAG).error( f"TTS任务失败: {error_code} - {error_message}" ) break except json.JSONDecodeError: logger.bind(tag=TAG).warning("收到无效的JSON消息") elif isinstance(msg, (bytes, bytearray)): self.opus_encoder.encode_pcm_to_opus_stream( msg, False, callback=self.handle_opus ) except websockets.ConnectionClosed: logger.bind(tag=TAG).warning("WebSocket连接已关闭") break except Exception as e: logger.bind(tag=TAG).error( f"处理TTS响应时出错: {e}\n{traceback.format_exc()}" ) break # 连接异常时关闭WebSocket if self.ws: try: await self.ws.close() except: pass self.ws = None # 监听任务退出时清理引用 finally: self.activate_session = False self._monitor_task = None def audio_to_opus_data_stream( self, audio_file_path, callback: Callable[[Any], Any] = None ): """重写父类方法：使用独立的临时编码器处理音频文件，避免与TTS流式编码器并发冲突。 双流式TTS中，monitor任务在event loop线程接收TTS音频并使用self.opus_encoder编码， 同时tts_text_priority_thread处理音乐文件也使用self.opus_encoder， 共享的encoder.buffer非线程安全，并发访问会导致SILK resampler断言失败。 """ from core.utils.util import audio_to_data_stream return audio_to_data_stream( audio_file_path, is_opus=True, callback=callback, sample_rate=self.conn.sample_rate, opus_encoder=None, ) def to_tts(self, text: str) -> list: """非流式生成音频数据，用于生成音频及测试场景""" try: # 创建事件循环 loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) # 生成会话ID session_id = uuid.uuid4().hex # 存储音频数据 audio_data = [] async def _generate_audio(): ws = await websockets.connect( self.ws_url, additional_headers=self.header, ping_interval=30, ping_timeout=10, close_timeout=10, max_size=10 * 1024 * 1024, ) try: # 发送run-task消息启动会话 run_task_message = { "header": { "action": "run-task", "task_id": session_id, "streaming": "duplex", }, "payload": { "task_group": "audio", "task": "tts", "function": "SpeechSynthesizer", "model": self.model, "parameters": { "text_type": "PlainText", "voice": self.voice, "format": self.format, "sample_rate": self.conn.sample_rate, "volume": self.volume, "rate": self.rate, "pitch": self.pitch, }, "input": {} }, } await ws.send(json.dumps(run_task_message)) # 等待任务启动 task_started = False while not task_started: msg = await ws.recv() if isinstance(msg, str): data = json.loads(msg) header = data.get("header", {}) if header.get("event") == "task-started": task_started = True logger.bind(tag=TAG).debug("TTS任务已启动") elif header.get("event") == "task-failed": error_code = header.get("error_code", "unknown") error_message = header.get("error_message", "未知错误") raise Exception( f"启动任务失败: {error_code} - {error_message}" ) # 发送文本 filtered_text = MarkdownCleaner.clean_markdown(text) if self._correct_words_pattern: filtered_text = self._correct_words_pattern.sub(lambda m: self.correct_words[m.group(0)], filtered_text) # 发送continue-task消息 continue_task_message = { "header": { "action": "continue-task", "task_id": session_id, "streaming": "duplex", }, "payload": {"input": {"text": filtered_text}}, } await ws.send(json.dumps(continue_task_message)) # 发送finish-task消息 finish_task_message = { "header": { "action": "finish-task", "task_id": session_id, "streaming": "duplex", }, "payload": { "input": {} } } await ws.send(json.dumps(finish_task_message)) # 接收音频数据 task_finished = False while not task_finished: msg = await ws.recv() if isinstance(msg, (bytes, bytearray)): self.opus_encoder.encode_pcm_to_opus_stream( msg, end_of_stream=False, callback=lambda opus: audio_data.append(opus) ) elif isinstance(msg, str): data = json.loads(msg) header = data.get("header", {}) if header.get("event") == "task-finished": task_finished = True logger.bind(tag=TAG).debug("TTS任务完成") elif header.get("event") == "task-failed": error_code = header.get("error_code", "unknown") error_message = header.get("error_message", "未知错误") raise Exception( f"合成失败: {error_code} - {error_message}" ) finally: # 清理资源 try: await ws.close() except: pass # 运行异步任务 loop.run_until_complete(_generate_audio()) loop.close() return audio_data except Exception as e: logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") return []