import os import uuid import json import hmac import hashlib import base64 import time import queue import asyncio import traceback import websockets from asyncio import Task from urllib import parse from datetime import datetime 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 AccessToken: @staticmethod def _encode_text(text): encoded_text = parse.quote_plus(text) return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~") @staticmethod def _encode_dict(dic): keys = dic.keys() dic_sorted = [(key, dic[key]) for key in sorted(keys)] encoded_text = parse.urlencode(dic_sorted) return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~") @staticmethod def create_token(access_key_id, access_key_secret, timeout): parameters = { "AccessKeyId": access_key_id, "Action": "CreateToken", "Format": "JSON", "RegionId": "cn-shanghai", # 使用上海地域进行Token获取 "SignatureMethod": "HMAC-SHA1", "SignatureNonce": str(uuid.uuid1()), "SignatureVersion": "1.0", "Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "Version": "2019-02-28", } query_string = AccessToken._encode_dict(parameters) string_to_sign = ( "GET" + "&" + AccessToken._encode_text("/") + "&" + AccessToken._encode_text(query_string) ) secreted_string = hmac.new( bytes(access_key_secret + "&", encoding="utf-8"), bytes(string_to_sign, encoding="utf-8"), hashlib.sha1, ).digest() signature = base64.b64encode(secreted_string) signature = AccessToken._encode_text(signature) full_url = "http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s" % ( signature, query_string, ) import requests response = requests.get(full_url, timeout=timeout) if response.ok: root_obj = response.json() key = "Token" if key in root_obj: token = root_obj[key]["Id"] expire_time = root_obj[key]["ExpireTime"] return token, expire_time return None, None class TTSProvider(TTSProviderBase): TTS_PARAM_CONFIG = [ ("ttsVolume", "volume", 0, 100, 50, int), ("ttsRate", "speech_rate", -500, 500, 0, int), ("ttsPitch", "pitch_rate", -500, 500, 0, int), ] def __init__(self, config, delete_audio_file): super().__init__(config, delete_audio_file) # 设置为流式接口类型 self.interface_type = InterfaceType.DUAL_STREAM # 基础配置 self.access_key_id = config.get("access_key_id") self.access_key_secret = config.get("access_key_secret") self.appkey = config.get("appkey") self.format = config.get("format", "pcm") self.report_on_last = True # 音色配置 - CosyVoice大模型音色 if config.get("private_voice"): self.voice = config.get("private_voice") else: self.voice = config.get("voice", "longxiaochun") # CosyVoice默认音色 # 音频参数配置 volume = config.get("volume", "50") self.volume = int(volume) if volume else 50 speech_rate = config.get("speech_rate", "0") self.speech_rate = int(speech_rate) if speech_rate else 0 pitch_rate = config.get("pitch_rate", "0") self.pitch_rate = int(pitch_rate) if pitch_rate else 0 # 应用百分比调整（如果存在），否则使用公有化配置 self._apply_percentage_params(config) # WebSocket配置 self.host = config.get("host", "nls-gateway-cn-beijing.aliyuncs.com") # 如果配置的是内网地址（包含-internal.aliyuncs.com），则使用ws协议，默认是wss协议 if "-internal." in self.host: self.ws_url = f"ws://{self.host}/ws/v1" else: # 默认使用wss协议 self.ws_url = f"wss://{self.host}/ws/v1" self.ws = None self._monitor_task = None self.activate_session = False self.last_active_time = None # 专属tts设置 self.task_id = uuid.uuid4().hex # Token管理 if self.access_key_id and self.access_key_secret: self._refresh_token() else: self.token = config.get("token") self.expire_time = None def _refresh_token(self): """刷新Token并记录过期时间""" if self.access_key_id and self.access_key_secret: self.token, expire_time_str = AccessToken.create_token( self.access_key_id, self.access_key_secret, self.tts_timeout ) if not expire_time_str: raise ValueError("无法获取有效的Token过期时间") expire_str = str(expire_time_str).strip() try: if expire_str.isdigit(): expire_time = datetime.fromtimestamp(int(expire_str)) else: expire_time = datetime.strptime(expire_str, "%Y-%m-%dT%H:%M:%SZ") self.expire_time = expire_time.timestamp() - 60 except Exception as e: raise ValueError(f"无效的过期时间格式: {expire_str}") from e else: self.expire_time = None if not self.token: raise ValueError("无法获取有效的访问Token") def _is_token_expired(self): """检查Token是否过期""" if not self.expire_time: return False return time.time() > self.expire_time async def _ensure_connection(self): """确保WebSocket连接可用""" try: if self._is_token_expired(): logger.bind(tag=TAG).warning("Token已过期，正在自动刷新...") self._refresh_token() current_time = time.time() if self.ws and current_time - self.last_active_time < 10: # 10秒内才可以复用链接进行连续对话 self.task_id = uuid.uuid4().hex logger.bind(tag=TAG).debug(f"使用已有链接..., task_id: {self.task_id}") return self.ws logger.bind(tag=TAG).debug("开始建立新连接...") # 建立新连接前取消旧监听任务 await self._cancel_monitor_task() self.ws = await websockets.connect( self.ws_url, additional_headers={"X-NLS-Token": self.token}, ping_interval=30, ping_timeout=10, close_timeout=10, ) self.task_id = uuid.uuid4().hex logger.bind(tag=TAG).debug(f"WebSocket连接建立成功, task_id: {self.task_id}") self.last_active_time = time.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): """流式文本处理线程""" while not self.conn.stop_event.is_set(): try: message = self.tts_text_queue.get(timeout=1) if self.conn.client_abort: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") asyncio.run_coroutine_threadsafe( self.finish_session(self.conn.sentence_id), loop=self.conn.loop, ) 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: logger.bind(tag=TAG).debug("开始启动TTS会话...") future = asyncio.run_coroutine_threadsafe( self.start_session(self.task_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.task_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()}" ) async def text_to_speak(self, text, _): try: if self.ws is None: logger.bind(tag=TAG).warning(f"WebSocket连接不存在，终止发送文本") return 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: run_request = { "header": { "message_id": uuid.uuid4().hex, "task_id": self.task_id, "namespace": "FlowingSpeechSynthesizer", "name": "RunSynthesis", "appkey": self.appkey, }, "payload": {"text": txt}, } await self.ws.send(json.dumps(run_request)) 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, task_id): logger.bind(tag=TAG).debug("开始会话～～") 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()) start_request = { "header": { "message_id": uuid.uuid4().hex, "task_id": self.task_id, "namespace": "FlowingSpeechSynthesizer", "name": "StartSynthesis", "appkey": self.appkey, }, "payload": { "voice": self.voice, "format": self.format, "sample_rate": self.conn.sample_rate, "volume": self.volume, "speech_rate": self.speech_rate, "pitch_rate": self.pitch_rate, "enable_subtitle": True, }, } await self.ws.send(json.dumps(start_request)) 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, task_id): logger.bind(tag=TAG).debug(f"关闭会话～～{task_id}") try: if self.ws: stop_request = { "header": { "message_id": uuid.uuid4().hex, "task_id": self.task_id, "namespace": "FlowingSpeechSynthesizer", "name": "StopSynthesis", "appkey": self.appkey, } } await self.ws.send(json.dumps(stop_request)) logger.bind(tag=TAG).debug("会话结束请求已发送") 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() if self.ws: try: await self.ws.close() except: pass self.ws = None self.last_active_time = 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): # 文本控制消息 try: data = json.loads(msg) header = data.get("header", {}) event_name = header.get("name") task_id = header.get("task_id") # 只处理当前活跃会话的响应 if task_id and self.task_id != task_id: if event_name in ["SynthesisCompleted", "TaskFailed"]: logger.bind(tag=TAG).debug(f"收到残余下行结束响应重置会话状态～～") self.activate_session = False continue if event_name == "SynthesisStarted": logger.bind(tag=TAG).debug("TTS合成已启动") self.tts_audio_queue.put( (SentenceType.FIRST, [], None) ) elif event_name == "SentenceEnd": # 发送缓存的数据 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_name == "SynthesisCompleted": logger.bind(tag=TAG).debug(f"会话结束～～") self.activate_session = False self._process_before_stop_play_files() except json.JSONDecodeError: logger.bind(tag=TAG).warning("收到无效的JSON消息") # 二进制消息（音频数据） elif isinstance(msg, (bytes, bytearray)): self.opus_encoder.encode_pcm_to_opus_stream(msg, False, 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, ) 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: """非流式TTS处理，用于测试及保存音频文件的场景""" try: # 创建新的事件循环 loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) # 存储音频数据 audio_data = [] async def _generate_audio(): # 刷新Token（如果需要） if self._is_token_expired(): self._refresh_token() # 建立WebSocket连接 ws = await websockets.connect( self.ws_url, additional_headers={"X-NLS-Token": self.token}, ping_interval=30, ping_timeout=10, close_timeout=10, ) try: # 发送StartSynthesis请求 start_request = { "header": { "message_id": uuid.uuid4().hex, "task_id": self.task_id, "namespace": "FlowingSpeechSynthesizer", "name": "StartSynthesis", "appkey": self.appkey, }, "payload": { "voice": self.voice, "format": self.format, "sample_rate": self.conn.sample_rate, "volume": self.volume, "speech_rate": self.speech_rate, "pitch_rate": self.pitch_rate, "enable_subtitle": True, }, } await ws.send(json.dumps(start_request)) # 等待SynthesisStarted响应 synthesis_started = False while not synthesis_started: msg = await ws.recv() if isinstance(msg, str): data = json.loads(msg) header = data.get("header", {}) if header.get("name") == "SynthesisStarted": synthesis_started = True logger.bind(tag=TAG).debug("TTS合成已启动") elif header.get("name") == "TaskFailed": error_info = data.get("payload", {}).get( "error_info", {} ) error_code = error_info.get("error_code") error_message = error_info.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) run_request = { "header": { "message_id": uuid.uuid4().hex, "task_id": self.task_id, "namespace": "FlowingSpeechSynthesizer", "name": "RunSynthesis", "appkey": self.appkey, }, "payload": {"text": filtered_text}, } await ws.send(json.dumps(run_request)) # 发送停止合成请求 stop_request = { "header": { "message_id": uuid.uuid4().hex, "task_id": self.task_id, "namespace": "FlowingSpeechSynthesizer", "name": "StopSynthesis", "appkey": self.appkey, } } await ws.send(json.dumps(stop_request)) # 接收音频数据 synthesis_completed = False while not synthesis_completed: 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", {}) event_name = header.get("name") if event_name == "SynthesisCompleted": synthesis_completed = True logger.bind(tag=TAG).debug("TTS合成完成") elif event_name == "TaskFailed": error_info = data.get("payload", {}).get( "error_info", {} ) error_code = error_info.get("error_code") error_message = error_info.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 []