import os import json import time import queue import asyncio import aiohttp import requests import traceback from core.utils import textUtils from config.logger import setup_logging from core.utils.util import parse_string_to_list from core.providers.tts.base import TTSProviderBase from core.providers.tts.dto.dto import SentenceType, ContentType from core.utils.tts import MarkdownCleaner, convert_percentage_to_range TAG = __name__ logger = setup_logging() class TTSProvider(TTSProviderBase): def __init__(self, config, delete_audio_file): super().__init__(config, delete_audio_file) self.group_id = config.get("group_id") self.api_key = config.get("api_key") self.model = config.get("model") if config.get("private_voice"): self.voice = config.get("private_voice") else: self.voice = config.get("voice_id") default_voice_setting = { "voice_id": "female-shaonv", "speed": 1, "vol": 1, "pitch": 0, "emotion": "happy", } default_pronunciation_dict = {"tone": ["处理/(chu3)(li3)", "危险/dangerous"]} defult_audio_setting = { "sample_rate": 24000, "bitrate": 128000, "format": "pcm", "channel": 1, } self.voice_setting = { **default_voice_setting, **config.get("voice_setting", {}), } self.pronunciation_dict = { **default_pronunciation_dict, **config.get("pronunciation_dict", {}), } self.audio_setting = {**defult_audio_setting, **config.get("audio_setting", {})} self.timber_weights = parse_string_to_list(config.get("timber_weights")) if self.voice: self.voice_setting["voice_id"] = self.voice # 应用百分比调整（如果存在），否则使用公有化配置 if "ttsVolume" in config: self.voice_setting["vol"] = round(convert_percentage_to_range( config["ttsVolume"], min_val=0.1, max_val=10, base_val=1.0 ), 1) if "ttsRate" in config: self.voice_setting["speed"] = round(convert_percentage_to_range( config["ttsRate"], min_val=0.5, max_val=2, base_val=1.0 ), 1) if "ttsPitch" in config: self.voice_setting["pitch"] = int(convert_percentage_to_range( config["ttsPitch"], min_val=-12, max_val=12, base_val=0 )) self.host = "api.minimaxi.com" # 备用地址：api-bj.minimaxi.com self.api_url = f"https://{self.host}/v1/t2a_v2?GroupId={self.group_id}" self.header = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}", } self.audio_file_type = defult_audio_setting.get("format", "pcm") # PCM缓冲区 self.pcm_buffer = bytearray() async def open_audio_channels(self, conn): """初始化音频通道,并根据conn.sample_rate更新配置""" # 调用父类方法 await super().open_audio_channels(conn) # 更新audio_setting中的采样率为实际的conn.sample_rate self.audio_setting["sample_rate"] = conn.sample_rate def tts_text_priority_thread(self): """流式文本处理线程""" while not self.conn.stop_event.is_set(): try: message = self.tts_text_queue.get(timeout=1) if message.sentence_type == SentenceType.FIRST: # 初始化参数 self.tts_stop_request = False self.processed_chars = 0 self.tts_text_buff = [] self.before_stop_play_files.clear() elif ContentType.TEXT == message.content_type: self.tts_text_buff.append(message.content_detail) segment_text = self._get_segment_text() if segment_text: self.to_tts_single_stream(segment_text) 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: # 处理剩余的文本 self._process_remaining_text_stream(True) except queue.Empty: continue except Exception as e: logger.bind(tag=TAG).error( f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" ) def _process_remaining_text_stream(self, is_last=False): """处理剩余的文本并生成语音 Returns: bool: 是否成功处理了文本 """ full_text = "".join(self.tts_text_buff) remaining_text = full_text[self.processed_chars :] if remaining_text: segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text) if segment_text: self.to_tts_single_stream(segment_text, is_last) self.processed_chars += len(full_text) else: self._process_before_stop_play_files() else: self._process_before_stop_play_files() def to_tts_single_stream(self, text, is_last=False): try: max_repeat_time = 5 original_text = text text = MarkdownCleaner.clean_markdown(text) if self._correct_words_pattern: text = self._correct_words_pattern.sub(lambda m: self.correct_words[m.group(0)], text) try: asyncio.run(self.text_to_speak(text, is_last)) except Exception as e: logger.bind(tag=TAG).warning( f"语音生成失败{5 - max_repeat_time + 1}次: {original_text}，错误: {e}" ) max_repeat_time -= 1 if max_repeat_time > 0: logger.bind(tag=TAG).info( f"语音生成成功: {original_text}，重试{5 - max_repeat_time}次" ) else: logger.bind(tag=TAG).error( f"语音生成失败: {original_text}，请检查网络或服务是否正常" ) except Exception as e: logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}") finally: return None async def text_to_speak(self, text, is_last): """流式处理TTS音频，每句只推送一次音频列表""" payload = { "model": self.model, "text": text, "stream": True, "voice_setting": self.voice_setting, "pronunciation_dict": self.pronunciation_dict, "audio_setting": self.audio_setting, } if type(self.timber_weights) is list and len(self.timber_weights) > 0: payload["timber_weights"] = self.timber_weights payload["voice_setting"]["voice_id"] = "" frame_bytes = int( self.opus_encoder.sample_rate * self.opus_encoder.channels # 1 * self.opus_encoder.frame_size_ms / 1000 * 2 ) # 16-bit = 2 bytes try: async with aiohttp.ClientSession() as session: async with session.post( self.api_url, headers=self.header, data=json.dumps(payload), timeout=self.tts_timeout, ) as resp: if resp.status != 200: logger.bind(tag=TAG).error( f"TTS请求失败: {resp.status}, {await resp.text()}" ) self.tts_audio_queue.put((SentenceType.LAST, [], None)) return self.pcm_buffer.clear() self.tts_audio_queue.put((SentenceType.FIRST, [], text)) # 处理音频流数据 buffer = b"" async for chunk in resp.content.iter_any(): if not chunk: continue buffer += chunk while True: # 查找数据块分隔符 header_pos = buffer.find(b"data: ") if header_pos == -1: break end_pos = buffer.find(b"\n\n", header_pos) if end_pos == -1: break # 提取单个完整JSON块 json_str = buffer[header_pos + 6 : end_pos].decode("utf-8") buffer = buffer[end_pos + 2 :] try: data = json.loads(json_str) # 检查业务层错误 base_resp = data.get("base_resp", {}) status_code = base_resp.get("status_code", 0) if status_code != 0: status_msg = base_resp.get("status_msg", "未知错误") logger.bind(tag=TAG).error( f"TTS请求失败, 错误码:{status_code}, 错误消息:{status_msg}" ) self.tts_audio_queue.put((SentenceType.LAST, [], None)) return status = data.get("data", {}).get("status", 1) audio_hex = data.get("data", {}).get("audio") # 仅处理status=1的有效音频块 忽略status=2的结束汇总块 if status == 1 and audio_hex: pcm_data = bytes.fromhex(audio_hex) self.pcm_buffer.extend(pcm_data) except json.JSONDecodeError as e: logger.bind(tag=TAG).error(f"JSON解析失败: {e}") continue while len(self.pcm_buffer) >= frame_bytes: frame = bytes(self.pcm_buffer[:frame_bytes]) del self.pcm_buffer[:frame_bytes] self.opus_encoder.encode_pcm_to_opus_stream( frame, end_of_stream=False, callback=self.handle_opus ) # flush 剩余不足一帧的数据 if self.pcm_buffer: self.opus_encoder.encode_pcm_to_opus_stream( bytes(self.pcm_buffer), end_of_stream=True, callback=self.handle_opus, ) self.pcm_buffer.clear() # 如果是最后一段，输出音频获取完毕 if is_last: self._process_before_stop_play_files() except Exception as e: logger.bind(tag=TAG).error(f"TTS请求异常: {e}") self.tts_audio_queue.put((SentenceType.LAST, [], None)) async def close(self): """资源清理""" await super().close() if hasattr(self, "opus_encoder"): self.opus_encoder.close() def to_tts(self, text: str) -> list: """非流式TTS处理，用于测试及保存音频文件的场景 Args: text: 要转换的文本 Returns: list: 返回opus编码后的音频数据列表 """ start_time = time.time() text = MarkdownCleaner.clean_markdown(text) if self._correct_words_pattern: text = self._correct_words_pattern.sub(lambda m: self.correct_words[m.group(0)], text) payload = { "model": self.model, "text": text, "stream": True, "voice_setting": self.voice_setting, "pronunciation_dict": self.pronunciation_dict, "audio_setting": self.audio_setting, } if type(self.timber_weights) is list and len(self.timber_weights) > 0: payload["timber_weights"] = self.timber_weights payload["voice_setting"]["voice_id"] = "" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}", } try: with requests.post( self.api_url, data=json.dumps(payload), headers=headers, timeout=self.tts_timeout, ) as response: if response.status_code != 200: logger.bind(tag=TAG).error( f"TTS请求失败: {response.status_code}, {response.text}" ) return [] logger.info(f"TTS请求成功: {text}, 耗时: {time.time() - start_time}秒") # 使用opus编码器处理PCM数据 opus_datas = [] full_content = response.content.decode('utf-8') pcm_data = bytearray() for data_block in full_content.split('\n\n'): if not data_block.startswith('data: '): continue try: json_str = data_block[6:] # 去除'data: '前缀 data = json.loads(json_str) if data.get('data', {}).get('status') == 1: audio_hex = data['data']['audio'] pcm_data.extend(bytes.fromhex(audio_hex)) except (json.JSONDecodeError, KeyError) as e: logger.bind(tag=TAG).warning(f"无效数据块: {e}") continue # 计算每帧的字节数 frame_bytes = int( self.opus_encoder.sample_rate * self.opus_encoder.channels * self.opus_encoder.frame_size_ms / 1000 * 2 ) # 分帧处理合并后的PCM数据 for i in range(0, len(pcm_data), frame_bytes): frame = bytes(pcm_data[i:i+frame_bytes]) if len(frame) < frame_bytes: frame += b"\x00" * (frame_bytes - len(frame)) self.opus_encoder.encode_pcm_to_opus_stream( frame, end_of_stream=(i + frame_bytes >= len(pcm_data)), callback=lambda opus: opus_datas.append(opus) ) return opus_datas except Exception as e: logger.bind(tag=TAG).error(f"TTS请求异常: {e}") return []