Merge pull request #3235 from xinnan-tech/py-fix-tts

fix: 超时情况不与activate_session同一判断条件导致的监听任务未清除问题

main/xiaozhi-server/core/providers/tts/alibl_stream.py

@@ -80,6 +80,9 @@ class TTSProvider(TTSProviderBase):
                 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,
@@ -306,16 +309,7 @@ class TTSProvider(TTSProviderBase):
         """清理资源"""
         await super().close()
         self.activate_session = False
-        # 取消监听任务
+        await self._cancel_monitor_task()
-        if self._monitor_task:
-            try:
-                self._monitor_task.cancel()
-                await self._monitor_task
-            except asyncio.CancelledError:
-                pass
-            except Exception as e:
-                logger.bind(tag=TAG).warning(f"关闭时取消监听任务错误: {e}")
-            self._monitor_task = None
 
         # 关闭WebSocket连接
         if self.ws:
@@ -325,6 +319,18 @@ class TTSProvider(TTSProviderBase):
                 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响应 - 长期运行"""

main/xiaozhi-server/core/providers/tts/aliyun_stream.py

@@ -193,6 +193,9 @@ class TTSProvider(TTSProviderBase):
                 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},
@@ -401,15 +404,7 @@ class TTSProvider(TTSProviderBase):
         """资源清理"""
         await super().close()
         self.activate_session = False
-        if self._monitor_task:
+        await self._cancel_monitor_task()
-            try:
-                self._monitor_task.cancel()
-                await self._monitor_task
-            except asyncio.CancelledError:
-                pass
-            except Exception as e:
-                logger.bind(tag=TAG).warning(f"关闭时取消监听任务错误: {e}")
-            self._monitor_task = None
 
         if self.ws:
             try:
@@ -504,6 +499,18 @@ class TTSProvider(TTSProviderBase):
             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:

main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py

@@ -229,6 +229,10 @@ class TTSProvider(TTSProviderBase):
                     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,
@@ -472,16 +476,7 @@ class TTSProvider(TTSProviderBase):
         """资源清理方法"""
         await super().close()
         self.activate_session = False
-        # 取消监听任务
+        await self._cancel_monitor_task()
-        if self._monitor_task:
-            try:
-                self._monitor_task.cancel()
-                await self._monitor_task
-            except asyncio.CancelledError:
-                pass
-            except Exception as e:
-                logger.bind(tag=TAG).warning(f"关闭时取消监听任务错误: {e}")
-            self._monitor_task = None
 
         if self.ws:
             try:
@@ -731,6 +726,18 @@ class TTSProvider(TTSProviderBase):
     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: