diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 94806924..52451cba 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -268,11 +268,7 @@ class TTSProvider(TTSProviderBase): ) if message.content_file and os.path.exists(message.content_file): # 先处理文件音频数据 - file_audio = self._process_audio_file(message.content_file) - self.before_stop_play_files.append( - (file_audio, message.content_detail) - ) - + 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).info("开始结束TTS会话...") @@ -486,7 +482,7 @@ class TTSProvider(TTSProviderBase): finally: self._monitor_task = None - def to_tts(self, text: str) -> list: + def to_tts_stream(self, text: str, opus_handler=self.handle_opus) -> None: """非流式TTS处理，用于测试及保存音频文件的场景""" try: # 创建新的事件循环 @@ -591,7 +587,7 @@ class TTSProvider(TTSProviderBase): if isinstance(msg, (bytes, bytearray)): # 编码为Opus并收集 self.opus_encoder.encode_pcm_to_opus_stream( - msg, False, self.handle_opus + msg, False, opus_handler ) # audio_data.extend(opus_frames) elif isinstance(msg, str): @@ -621,7 +617,5 @@ class TTSProvider(TTSProviderBase): loop.run_until_complete(_generate_audio()) loop.close() - return audio_data except Exception as e: logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") - return [] \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 575e1325..d55f434e 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -88,7 +88,12 @@ class TTSProviderBase(ABC): (SentenceType.MIDDLE, opus_data, None) ) - def to_tts(self, text, opus_handler=handle_opus) -> None: + def handle_audio_file(self, file_audio: bytes, text): + self.before_stop_play_files.append( + (file_audio, text) + ) + + def to_tts_stream(self, text, opus_handler=handle_opus) -> None: text = MarkdownCleaner.clean_markdown(text) max_repeat_time = 5 if self.delete_audio_file: @@ -146,7 +151,7 @@ class TTSProviderBase(ABC): self.tts_audio_queue.put( (SentenceType.FIRST, None, text) ) - self._process_audio_file(tmp_file, callback=opus_handler) + self._process_audio_file_stream(tmp_file, callback=opus_handler) except Exception as e: logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}") return None @@ -228,14 +233,14 @@ class TTSProviderBase(ABC): self.tts_text_buff.append(message.content_detail) segment_text = self._get_segment_text() if segment_text: - self.to_tts(segment_text, opus_handler=self.handle_opus) + self.to_tts_stream(segment_text, opus_handler=self.handle_opus) elif ContentType.FILE == message.content_type: - self._process_remaining_text() + self._process_remaining_text_stream(opus_handler=self.handle_opus) tts_file = message.content_file if tts_file and os.path.exists(tts_file): - self._process_audio_file(tts_file, callback=self.handle_opus) + self._process_audio_file_stream(tts_file, callback=self.handle_opus) if message.sentence_type == SentenceType.LAST: - self._process_remaining_text() + self._process_remaining_text_stream(opus_handler=self.handle_opus) self.tts_audio_queue.put( (message.sentence_type, [], message.content_detail) ) @@ -419,15 +424,12 @@ class TTSProviderBase(ABC): else: return None - def _process_audio_file(self, tts_file, callback: Callable[[Any], Any]): + def _process_audio_file_stream(self, tts_file, callback: Callable[[Any], Any]) -> None: """处理音频文件并转换为指定格式 Args: tts_file: 音频文件路径 - content_detail: 内容详情 - - Returns: - tuple: (sentence_type, audio_datas, content_detail) + callback: 文件处理函数 """ if tts_file.endswith(".p3"): p3.decode_opus_from_file_stream(tts_file, callback=callback) @@ -450,7 +452,7 @@ class TTSProviderBase(ABC): self.before_stop_play_files.clear() self.tts_audio_queue.put((SentenceType.LAST, [], None)) - def _process_remaining_text(self): + def _process_remaining_text_stream(self, opus_handler=handle_opus): """处理剩余的文本并生成语音 Returns: @@ -461,7 +463,7 @@ class TTSProviderBase(ABC): if remaining_text: segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text) if segment_text: - self.to_tts(segment_text, opus_handler=self.handle_opus) + self.to_tts_stream(segment_text, opus_handler=opus_handler) self.processed_chars += len(full_text) return True return False diff --git a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py index ea7f18fd..f45b9a96 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -4,6 +4,8 @@ import json import queue import asyncio import traceback +from typing import Callable, Any + import websockets from core.utils.tts import MarkdownCleaner from config.logger import setup_logging @@ -260,11 +262,7 @@ class TTSProvider(TTSProviderBase): ) if message.content_file and os.path.exists(message.content_file): # 先处理文件音频数据 - file_audio = self._process_audio_file(message.content_file) - self.before_stop_play_files.append( - (file_audio, message.content_detail) - ) - + 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).info("开始结束TTS会话...") @@ -452,21 +450,7 @@ class TTSProvider(TTSProviderBase): and res.header.message_type == AUDIO_ONLY_RESPONSE ): logger.bind(tag=TAG).debug(f"推送数据到队列里面～～") - opus_datas = self.wav_to_opus_data_audio_raw(res.payload) - logger.bind(tag=TAG).debug( - f"推送数据到队列里面帧数～～{len(opus_datas)}" - ) - if is_first_sentence: - first_sentence_segment_count += 1 - if first_sentence_segment_count <= 6: - self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus_datas, None) - ) - else: - opus_datas_cache.extend(opus_datas) - else: - # 后续句子缓存 - opus_datas_cache.extend(opus_datas) + self.wav_to_opus_data_audio_raw_stream(res.payload, callback=self.handle_opus) elif res.optional.event == EVENT_TTSSentenceEnd: logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}") if not is_first_sentence or first_sentence_segment_count > 10: @@ -642,15 +626,15 @@ class TTSProvider(TTSProviderBase): ) ) - def wav_to_opus_data_audio_raw(self, raw_data_var, is_end=False): - opus_datas = self.opus_encoder.encode_pcm_to_opus(raw_data_var, is_end) - return opus_datas + 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) - def to_tts(self, text: str) -> list: + def to_tts_stream(self, text: str, opus_handler=self.handle_opus) -> None: """非流式生成音频数据，用于生成音频及测试场景 Args: text: 要转换的文本 + opus_handler: opus数据处理方法 Returns: list: 音频数据列表 @@ -663,9 +647,6 @@ class TTSProvider(TTSProviderBase): # 生成会话ID session_id = uuid.uuid4().__str__().replace("-", "") - # 存储音频数据 - audio_data = [] - async def _generate_audio(): # 创建新的WebSocket连接 ws_header = { @@ -728,8 +709,7 @@ class TTSProvider(TTSProviderBase): res.optional.event == EVENT_TTSResponse and res.header.message_type == AUDIO_ONLY_RESPONSE ): - opus_datas = self.wav_to_opus_data_audio_raw(res.payload) - audio_data.extend(opus_datas) + self.wav_to_opus_data_audio_raw_stream(res.payload, callback=opus_handler) elif res.optional.event == EVENT_SessionFinished: break @@ -744,8 +724,6 @@ class TTSProvider(TTSProviderBase): loop.run_until_complete(_generate_audio()) loop.close() - return audio_data - except Exception as e: logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") return [] diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index d1be586c..a9877d43 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/main/xiaozhi-server/core/providers/tts/linkerai.py @@ -65,14 +65,10 @@ class TTSProvider(TTSProviderBase): ) if message.content_file and os.path.exists(message.content_file): # 先处理文件音频数据 - file_audio = self._process_audio_file(message.content_file) - self.before_stop_play_files.append( - (file_audio, message.content_detail) - ) - + 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(True) + self._process_remaining_text_stream(True) except queue.Empty: continue @@ -81,7 +77,7 @@ class TTSProvider(TTSProviderBase): f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" ) - def _process_remaining_text(self, is_last=False): + def _process_remaining_text_stream(self, is_last=False): """处理剩余的文本并生成语音 Returns: @@ -237,7 +233,7 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).error(f"TTS请求异常: {e}") self.tts_audio_queue.put((SentenceType.LAST, [], None)) - def to_tts(self, text: str) -> list: + def to_tts_stream(self, text: str) -> list: """非流式TTS处理，用于测试及保存音频文件的场景 Args: