import os import uuid import json import queue import asyncio import traceback import websockets from typing import Callable, Any from config.logger import setup_logging from core.utils.util import check_model_key from core.providers.tts.base import TTSProviderBase from core.utils.tts import MarkdownCleaner, convert_percentage_to_range from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType TAG = __name__ logger = setup_logging() PROTOCOL_VERSION = 0b0001 DEFAULT_HEADER_SIZE = 0b0001 # Message Type: FULL_CLIENT_REQUEST = 0b0001 AUDIO_ONLY_RESPONSE = 0b1011 FULL_SERVER_RESPONSE = 0b1001 ERROR_INFORMATION = 0b1111 # Message Type Specific Flags MsgTypeFlagNoSeq = 0b0000 # Non-terminal packet with no sequence MsgTypeFlagPositiveSeq = 0b1 # Non-terminal packet with sequence > 0 MsgTypeFlagLastNoSeq = 0b10 # last packet with no sequence MsgTypeFlagNegativeSeq = 0b11 # Payload contains event number (int32) MsgTypeFlagWithEvent = 0b100 # Message Serialization NO_SERIALIZATION = 0b0000 JSON = 0b0001 # Message Compression COMPRESSION_NO = 0b0000 COMPRESSION_GZIP = 0b0001 EVENT_NONE = 0 EVENT_Start_Connection = 1 EVENT_FinishConnection = 2 EVENT_ConnectionStarted = 50 # 成功建连 EVENT_ConnectionFailed = 51 # 建连失败（可能是无法通过权限认证） EVENT_ConnectionFinished = 52 # 连接结束 # 上行Session事件 EVENT_StartSession = 100 EVENT_CancelSession = 101 EVENT_FinishSession = 102 # 下行Session事件 EVENT_SessionStarted = 150 EVENT_SessionCanceled = 151 EVENT_SessionFinished = 152 EVENT_SessionFailed = 153 # 上行通用事件 EVENT_TaskRequest = 200 # 下行TTS事件 EVENT_TTSSentenceStart = 350 EVENT_TTSSentenceEnd = 351 EVENT_TTSResponse = 352 class Header: def __init__( self, protocol_version=PROTOCOL_VERSION, header_size=DEFAULT_HEADER_SIZE, message_type: int = 0, message_type_specific_flags: int = 0, serial_method: int = NO_SERIALIZATION, compression_type: int = COMPRESSION_NO, reserved_data=0, ): self.header_size = header_size self.protocol_version = protocol_version self.message_type = message_type self.message_type_specific_flags = message_type_specific_flags self.serial_method = serial_method self.compression_type = compression_type self.reserved_data = reserved_data def as_bytes(self) -> bytes: return bytes( [ (self.protocol_version << 4) | self.header_size, (self.message_type << 4) | self.message_type_specific_flags, (self.serial_method << 4) | self.compression_type, self.reserved_data, ] ) class Optional: def __init__( self, event: int = EVENT_NONE, sessionId: str = None, sequence: int = None ): self.event = event self.sessionId = sessionId self.errorCode: int = 0 self.connectionId: str | None = None self.response_meta_json: str | None = None self.sequence = sequence # 转成 byte 序列 def as_bytes(self) -> bytes: option_bytes = bytearray() if self.event != EVENT_NONE: option_bytes.extend(self.event.to_bytes(4, "big", signed=True)) if self.sessionId is not None: session_id_bytes = str.encode(self.sessionId) size = len(session_id_bytes).to_bytes(4, "big", signed=True) option_bytes.extend(size) option_bytes.extend(session_id_bytes) if self.sequence is not None: option_bytes.extend(self.sequence.to_bytes(4, "big", signed=True)) return option_bytes class Response: def __init__(self, header: Header, optional: Optional): self.optional = optional self.header = header self.payload: bytes | None = None def __str__(self): return super().__str__() class TTSProvider(TTSProviderBase): def __init__(self, config, delete_audio_file): super().__init__(config, delete_audio_file) self.ws = None self.interface_type = InterfaceType.DUAL_STREAM self._monitor_task = None # 监听任务引用 self.appId = config.get("appid") self.access_token = config.get("access_token") self.cluster = config.get("cluster") self.resource_id = config.get("resource_id") self.resource_type = True if self.resource_id == "seed-tts-2.0" else False self.report_on_last = self.resource_type self.activate_session = False if config.get("private_voice"): self.voice = config.get("private_voice") else: self.voice = config.get("speaker") # 默认 audio_params 配置 default_audio_params = { "speech_rate": 0, "loudness_rate": 0 } # 默认 additions 配置 default_additions = { "aigc_metadata": {}, "cache_config": {}, "post_process": { "pitch": 0 } } # 默认 mix_speaker 配置 default_mix_speaker = {} # 合并用户配置 self.audio_params = {**default_audio_params, **config.get("audio_params", {})} self.additions = {**default_additions, **config.get("additions", {})} self.mix_speaker = {**default_mix_speaker, **config.get("mix_speaker", {})} # 应用百分比调整（如果存在），否则使用公有化配置 if "ttsVolume" in config: self.audio_params["loudness_rate"] = int(convert_percentage_to_range( config["ttsVolume"], min_val=-50, max_val=100, base_val=0 )) if "ttsRate" in config: self.audio_params["speech_rate"] = int(convert_percentage_to_range( config["ttsRate"], min_val=-50, max_val=100, base_val=0 )) if "ttsPitch" in config: self.additions["post_process"]["pitch"] = int(convert_percentage_to_range( config["ttsPitch"], min_val=-12, max_val=12, base_val=0 )) self.ws_url = config.get("ws_url") self.authorization = config.get("authorization") self.header = {"Authorization": f"{self.authorization}{self.access_token}"} enable_ws_reuse_value = config.get("enable_ws_reuse", True) self.enable_ws_reuse = False if str(enable_ws_reuse_value).lower() == 'false' else True self.tts_text = "" model_key_msg = check_model_key("TTS", self.access_token) if model_key_msg: logger.bind(tag=TAG).error(model_key_msg) async def open_audio_channels(self, conn): try: await super().open_audio_channels(conn) # 更新 audio_params 中的采样率为实际的 conn.sample_rate self.audio_params["sample_rate"] = conn.sample_rate except Exception as e: logger.bind(tag=TAG).error(f"Failed to open audio channels: {str(e)}") self.ws = None raise async def _ensure_connection(self): """建立新的WebSocket连接，并启动监听任务（仅第一次）""" try: if self.ws: if self.enable_ws_reuse: logger.bind(tag=TAG).debug(f"使用已有链接...") return self.ws else: try: await self.finish_connection() except: pass logger.bind(tag=TAG).debug("开始建立新连接...") # 建立新连接前取消旧监听任务 await self._cancel_monitor_task() ws_header = { "X-Api-App-Key": self.appId, "X-Api-Access-Key": self.access_token, "X-Api-Resource-Id": self.resource_id, "X-Api-Connect-Id": uuid.uuid4(), } self.ws = await websockets.connect( self.ws_url, additional_headers=ws_header, max_size=1000000000 ) logger.bind(tag=TAG).debug("WebSocket连接建立成功") # 连接建立成功后，启动监听任务 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()) return self.ws except Exception as e: logger.bind(tag=TAG).error(f"建立连接失败: {str(e)}") self.ws = None raise async def finish_connection(self): """发送 FinishConnection 事件，等待服务端返回 EVENT_ConnectionFinished""" try: if self.ws: logger.bind(tag=TAG).debug("开始关闭连接...") header = Header( message_type=FULL_CLIENT_REQUEST, message_type_specific_flags=MsgTypeFlagWithEvent, serial_method=JSON, ).as_bytes() optional = Optional(event=EVENT_FinishConnection).as_bytes() payload = str.encode("{}") await self.send_event(self.ws, header, optional, payload) except: pass 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文本处理线程") if self.enable_ws_reuse: asyncio.run_coroutine_threadsafe( self.cancel_session(self.conn.sentence_id), loop=self.conn.loop, ) else: asyncio.run_coroutine_threadsafe( self.finish_connection(), 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(timeout=self.tts_timeout) 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(f"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: await self.send_text(self.voice, txt, self.conn.sentence_id) 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): logger.bind(tag=TAG).debug(f"开始会话～～{session_id}") try: # 上个会话处于激活状态时关闭上个连接新建链接 if self.activate_session: await self.close() # 设置会话激活标志 self.activate_session = True # 确保连接建立 await self._ensure_connection() header = Header( message_type=FULL_CLIENT_REQUEST, message_type_specific_flags=MsgTypeFlagWithEvent, serial_method=JSON, ).as_bytes() optional = Optional( event=EVENT_StartSession, sessionId=session_id ).as_bytes() payload = self.get_payload_bytes( event=EVENT_StartSession, speaker=self.voice ) await self.send_event(self.ws, header, optional, payload) 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): logger.bind(tag=TAG).debug(f"关闭会话～～{session_id}") try: if self.ws: header = Header( message_type=FULL_CLIENT_REQUEST, message_type_specific_flags=MsgTypeFlagWithEvent, serial_method=JSON, ).as_bytes() optional = Optional( event=EVENT_FinishSession, sessionId=session_id ).as_bytes() payload = str.encode("{}") await self.send_event(self.ws, header, optional, payload) logger.bind(tag=TAG).debug("会话结束请求已发送") except Exception as e: logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}") # 确保清理资源 await self.close() raise async def cancel_session(self,session_id): logger.bind(tag=TAG).debug(f"取消会话，释放服务端资源～～{session_id}") try: if self.ws: header = Header( message_type=FULL_CLIENT_REQUEST, message_type_specific_flags=MsgTypeFlagWithEvent, serial_method=JSON, ).as_bytes() optional = Optional( event=EVENT_CancelSession, sessionId=session_id ).as_bytes() payload = str.encode("{}") await self.send_event(self.ws, header, optional, payload) logger.bind(tag=TAG).debug("会话取消请求已发送") 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() if self.ws: try: await self.ws.close() except: pass self.ws = None async def _start_monitor_tts_response(self): """监听TTS响应 - 长期运行""" try: while not self.conn.stop_event.is_set(): try: # 确保 `recv()` 运行在同一个 event loop msg = await self.ws.recv() res = self.parser_response(msg) self.print_response(res, "send_text res:") # 优先处理连接级别事件 if res.optional.event == EVENT_ConnectionFinished: logger.bind(tag=TAG).debug(f"链接关闭成功～～") break # 只处理当前活跃会话的响应 if res.optional.sessionId and self.conn.sentence_id != res.optional.sessionId: # 如果是会话结束相关事件，即使会话ID不匹配也要重置状态 if res.optional.event in [EVENT_SessionCanceled, EVENT_SessionFailed, EVENT_SessionFinished]: logger.bind(tag=TAG).debug(f"收到残余下行结束响应重置会话状态～～") self.activate_session = False continue if res.optional.event == EVENT_SessionCanceled: logger.bind(tag=TAG).debug(f"释放服务端资源成功～～") self.activate_session = False elif not self.resource_type and res.optional.event == EVENT_TTSSentenceStart: json_data = json.loads(res.payload.decode("utf-8")) self.tts_text = json_data.get("text", "") logger.bind(tag=TAG).debug(f"句子语音生成开始: {self.tts_text}") # 将TTS服务返回的替换后文本还原为原始文本，用于字幕显示 display_text = self._restore_original_text(self.tts_text) self.tts_audio_queue.put((SentenceType.FIRST, [], display_text)) elif ( res.optional.event == EVENT_TTSResponse and res.header.message_type == AUDIO_ONLY_RESPONSE ): # 处理seed-tts-2.0文本字幕 if self.resource_type: 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) self.wav_to_opus_data_audio_raw_stream(res.payload, callback=self.handle_opus) elif not self.resource_type and res.optional.event == EVENT_TTSSentenceEnd: logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}") elif res.optional.event == EVENT_SessionFinished: logger.bind(tag=TAG).debug(f"会话结束～～") self.activate_session = False self._process_before_stop_play_files() # 非复用模式下，会话结束后发送 FinishConnection if not self.enable_ws_reuse: await self.finish_connection() except websockets.ConnectionClosed: logger.bind(tag=TAG).warning("WebSocket连接已关闭") break except Exception as e: logger.bind(tag=TAG).error( f"Error in _start_monitor_tts_response: {e}" ) traceback.print_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 async def send_event( self, ws: websockets.WebSocketClientProtocol, header: bytes, optional: bytes | None = None, payload: bytes = None, ): try: full_client_request = bytearray(header) if optional is not None: full_client_request.extend(optional) if payload is not None: payload_size = len(payload).to_bytes(4, "big", signed=True) full_client_request.extend(payload_size) full_client_request.extend(payload) await ws.send(full_client_request) except websockets.ConnectionClosed: logger.bind(tag=TAG).error(f"ConnectionClosed") raise async def send_text(self, speaker: str, text: str, session_id): header = Header( message_type=FULL_CLIENT_REQUEST, message_type_specific_flags=MsgTypeFlagWithEvent, serial_method=JSON, ).as_bytes() optional = Optional(event=EVENT_TaskRequest, sessionId=session_id).as_bytes() payload = self.get_payload_bytes( event=EVENT_TaskRequest, text=text, speaker=speaker ) return await self.send_event(self.ws, header, optional, payload) # 读取 res 数组某段 字符串内容 def read_res_content(self, res: bytes, offset: int): content_size = int.from_bytes(res[offset : offset + 4], "big", signed=True) offset += 4 content = res[offset : offset + content_size].decode('utf-8') offset += content_size return content, offset # 读取 payload def read_res_payload(self, res: bytes, offset: int): payload_size = int.from_bytes(res[offset : offset + 4], "big", signed=True) offset += 4 payload = res[offset : offset + payload_size] offset += payload_size return payload, offset def parser_response(self, res) -> Response: if isinstance(res, str): raise RuntimeError(res) response = Response(Header(), Optional()) # 解析结果 # header header = response.header num = 0b00001111 header.protocol_version = res[0] >> 4 & num header.header_size = res[0] & 0x0F header.message_type = (res[1] >> 4) & num header.message_type_specific_flags = res[1] & 0x0F header.serialization_method = res[2] >> num header.message_compression = res[2] & 0x0F header.reserved = res[3] # offset = 4 optional = response.optional if header.message_type == FULL_SERVER_RESPONSE or AUDIO_ONLY_RESPONSE: # read event if header.message_type_specific_flags == MsgTypeFlagWithEvent: optional.event = int.from_bytes(res[offset:8], "big", signed=True) offset += 4 if optional.event == EVENT_NONE: return response # read connectionId elif optional.event == EVENT_ConnectionStarted: optional.connectionId, offset = self.read_res_content(res, offset) elif optional.event == EVENT_ConnectionFailed: optional.response_meta_json, offset = self.read_res_content( res, offset ) elif ( optional.event == EVENT_SessionStarted or optional.event == EVENT_SessionFailed or optional.event == EVENT_SessionFinished ): optional.sessionId, offset = self.read_res_content(res, offset) optional.response_meta_json, offset = self.read_res_content( res, offset ) else: optional.sessionId, offset = self.read_res_content(res, offset) response.payload, offset = self.read_res_payload(res, offset) elif header.message_type == ERROR_INFORMATION: optional.errorCode = int.from_bytes( res[offset : offset + 4], "big", signed=True ) offset += 4 response.payload, offset = self.read_res_payload(res, offset) return response async def start_connection(self): header = Header( message_type=FULL_CLIENT_REQUEST, message_type_specific_flags=MsgTypeFlagWithEvent, ).as_bytes() optional = Optional(event=EVENT_Start_Connection).as_bytes() payload = str.encode("{}") return await self.send_event(self.ws, header, optional, payload) def print_response(self, res, tag_msg: str): logger.bind(tag=TAG).debug(f"===>{tag_msg} header:{res.header.__dict__}") logger.bind(tag=TAG).debug(f"===>{tag_msg} optional:{res.optional.__dict__}") def get_payload_bytes( self, uid="1234", event=EVENT_NONE, text="", speaker="", audio_format="pcm", ): # 构建 req_params req_params = { "text": text, "speaker": speaker, "audio_params": {**self.audio_params, "format": audio_format}, "additions": json.dumps(self.additions) } # 如果有 mix_speaker 配置，添加到 req_params if self.mix_speaker: req_params["mix_speaker"] = self.mix_speaker return str.encode( json.dumps( { "user": {"uid": uid}, "event": event, "namespace": "BidirectionalTTS", "req_params": req_params } ) ) 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 wav_to_opus_data_audio_raw_stream(self, raw_data_var, is_end=False, callback: Callable[[Any], Any]=None): return self.opus_encoder.encode_pcm_to_opus_stream(raw_data_var, is_end, callback=callback) 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 def to_tts(self, text: str) -> list: """非流式生成音频数据，用于生成音频及测试场景 Args: text: 要转换的文本 Returns: list: 音频数据列表 """ try: # 创建事件循环 loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) # 生成会话ID session_id = uuid.uuid4().__str__().replace("-", "") # 存储音频数据 audio_data = [] async def _generate_audio(): # 创建新的WebSocket连接 ws_header = { "X-Api-App-Key": self.appId, "X-Api-Access-Key": self.access_token, "X-Api-Resource-Id": self.resource_id, "X-Api-Connect-Id": uuid.uuid4(), } ws = await websockets.connect( self.ws_url, additional_headers=ws_header, max_size=1000000000 ) try: # 启动会话 header = Header( message_type=FULL_CLIENT_REQUEST, message_type_specific_flags=MsgTypeFlagWithEvent, serial_method=JSON, ).as_bytes() optional = Optional( event=EVENT_StartSession, sessionId=session_id ).as_bytes() payload = self.get_payload_bytes( event=EVENT_StartSession, speaker=self.voice ) await self.send_event(ws, header, optional, payload) # 发送文本 header = Header( message_type=FULL_CLIENT_REQUEST, message_type_specific_flags=MsgTypeFlagWithEvent, serial_method=JSON, ).as_bytes() optional = Optional( event=EVENT_TaskRequest, sessionId=session_id ).as_bytes() payload = self.get_payload_bytes( event=EVENT_TaskRequest, text=text, speaker=self.voice ) await self.send_event(ws, header, optional, payload) # 发送结束会话请求 header = Header( message_type=FULL_CLIENT_REQUEST, message_type_specific_flags=MsgTypeFlagWithEvent, serial_method=JSON, ).as_bytes() optional = Optional( event=EVENT_FinishSession, sessionId=session_id ).as_bytes() payload = str.encode("{}") await self.send_event(ws, header, optional, payload) # 接收音频数据 while True: msg = await ws.recv() res = self.parser_response(msg) if ( res.optional.event == EVENT_TTSResponse and res.header.message_type == AUDIO_ONLY_RESPONSE ): self.wav_to_opus_data_audio_raw_stream(res.payload, callback=lambda opus_frame: audio_data.append(opus_frame)) elif res.optional.event == EVENT_SessionFinished: break 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 []