自主选择链接复用功能

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

@@ -163,7 +163,8 @@ class TTSProvider(TTSProviderBase):
         self.ws_url = config.get("ws_url")
         self.authorization = config.get("authorization")
         self.header = {"Authorization": f"{self.authorization}{self.access_token}"}
-        self.enable_two_way = True
+        enable_ws_reuse_value = config.get("enable_ws_reuse", True)
+        self.enable_ws_reuse = False if str(enable_ws_reuse_value).lower() in ('false', 'False') else True
         self.tts_text = ""
         self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
             sample_rate=16000, channels=1, frame_size_ms=60
@@ -184,8 +185,14 @@ class TTSProvider(TTSProviderBase):
         """建立新的WebSocket连接，并启动监听任务（仅第一次）"""
         try:
             if self.ws:
-                logger.bind(tag=TAG).info(f"使用已有链接...")
-                return self.ws
+                if self.enable_ws_reuse:
+                    logger.bind(tag=TAG).info(f"使用已有链接...")
+                    return self.ws
+                else:
+                    try:
+                        await self.finish_connection()
+                    except:
+                        pass
             logger.bind(tag=TAG).debug("开始建立新连接...")
             ws_header = {
                 "X-Api-App-Key": self.appId,
@@ -208,6 +215,22 @@ class TTSProvider(TTSProviderBase):
             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的文本处理线程"""
@@ -224,10 +247,16 @@ class TTSProvider(TTSProviderBase):
                 if self.conn.client_abort:
                     try:
                         logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
-                        asyncio.run_coroutine_threadsafe(
-                            self.cancel_session(self.conn.sentence_id),
-                            loop=self.conn.loop,
-                        )
+                        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)}")
@@ -432,6 +461,11 @@ class TTSProvider(TTSProviderBase):
                     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不匹配也要重置状态
@@ -461,6 +495,9 @@ class TTSProvider(TTSProviderBase):
                         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
@@ -479,6 +516,7 @@ class TTSProvider(TTSProviderBase):
                 self.ws = None
         # 监听任务退出时清理引用
         finally:
+            self.activate_session = False
             self._monitor_task = None
 
     async def send_event(