From 3cfc1cf97a38d058eba2f56ecfde8481d4b18737 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Wed, 3 Sep 2025 09:43:34 +0800 Subject: [PATCH] =?UTF-8?q?textHandle.py=20=E6=B6=88=E6=81=AF=E8=A7=A3?= =?UTF-8?q?=E8=80=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/handle/textHandle.py | 171 +----------------- .../handle/textHandler/abortMessageHandler.py | 16 ++ .../handle/textHandler/helloMessageHandler.py | 16 ++ .../handle/textHandler/iotMessageHandler.py | 20 ++ .../textHandler/listenMessageHandler.py | 63 +++++++ .../handle/textHandler/mcpMessageHandler.py | 20 ++ .../textHandler/serverMessageHandler.py | 92 ++++++++++ .../core/handle/textMessageHandler.py | 21 +++ .../core/handle/textMessageHandlerRegistry.py | 45 +++++ .../core/handle/textMessageProcessor.py | 41 +++++ .../core/handle/textMessageType.py | 11 ++ 11 files changed, 353 insertions(+), 163 deletions(-) create mode 100644 main/xiaozhi-server/core/handle/textHandler/abortMessageHandler.py create mode 100644 main/xiaozhi-server/core/handle/textHandler/helloMessageHandler.py create mode 100644 main/xiaozhi-server/core/handle/textHandler/iotMessageHandler.py create mode 100644 main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py create mode 100644 main/xiaozhi-server/core/handle/textHandler/mcpMessageHandler.py create mode 100644 main/xiaozhi-server/core/handle/textHandler/serverMessageHandler.py create mode 100644 main/xiaozhi-server/core/handle/textMessageHandler.py create mode 100644 main/xiaozhi-server/core/handle/textMessageHandlerRegistry.py create mode 100644 main/xiaozhi-server/core/handle/textMessageProcessor.py create mode 100644 main/xiaozhi-server/core/handle/textMessageType.py diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index 1bcf4545..b5e87783 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -1,169 +1,14 @@ -import json -import time -from core.handle.abortHandle import handleAbortMessage -from core.handle.helloHandle import handleHelloMessage -from core.providers.tools.device_mcp import handle_mcp_message -from core.utils.util import remove_punctuation_and_length, filter_sensitive_info -from core.handle.receiveAudioHandle import startToChat, handleAudioMessage -from core.handle.sendAudioHandle import send_stt_message, send_tts_message -from core.providers.tools.device_iot import handleIotDescriptors, handleIotStatus -from core.handle.reportHandle import enqueue_asr_report -import asyncio +from core.handle.textMessageHandlerRegistry import TextMessageHandlerRegistry +from core.handle.textMessageProcessor import TextMessageProcessor TAG = __name__ +# 全局处理器注册表 +message_registry = TextMessageHandlerRegistry() + +# 创建全局消息处理器实例 +message_processor = TextMessageProcessor(message_registry) async def handleTextMessage(conn, message): """处理文本消息""" - try: - msg_json = json.loads(message) - if isinstance(msg_json, int): - conn.logger.bind(tag=TAG).info(f"收到文本消息：{message}") - await conn.websocket.send(message) - return - if msg_json["type"] == "hello": - conn.logger.bind(tag=TAG).info(f"收到hello消息：{message}") - await handleHelloMessage(conn, msg_json) - elif msg_json["type"] == "abort": - conn.logger.bind(tag=TAG).info(f"收到abort消息：{message}") - await handleAbortMessage(conn) - elif msg_json["type"] == "listen": - conn.logger.bind(tag=TAG).info(f"收到listen消息：{message}") - if "mode" in msg_json: - conn.client_listen_mode = msg_json["mode"] - conn.logger.bind(tag=TAG).debug( - f"客户端拾音模式：{conn.client_listen_mode}" - ) - if msg_json["state"] == "start": - conn.client_have_voice = True - conn.client_voice_stop = False - elif msg_json["state"] == "stop": - conn.client_have_voice = True - conn.client_voice_stop = True - if len(conn.asr_audio) > 0: - await handleAudioMessage(conn, b"") - elif msg_json["state"] == "detect": - conn.client_have_voice = False - conn.asr_audio.clear() - if "text" in msg_json: - conn.last_activity_time = time.time() * 1000 - original_text = msg_json["text"] # 保留原始文本 - filtered_len, filtered_text = remove_punctuation_and_length( - original_text - ) - - # 识别是否是唤醒词 - is_wakeup_words = filtered_text in conn.config.get("wakeup_words") - # 是否开启唤醒词回复 - enable_greeting = conn.config.get("enable_greeting", True) - - if is_wakeup_words and not enable_greeting: - # 如果是唤醒词，且关闭了唤醒词回复，就不用回答 - await send_stt_message(conn, original_text) - await send_tts_message(conn, "stop", None) - conn.client_is_speaking = False - elif is_wakeup_words: - conn.just_woken_up = True - # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） - enqueue_asr_report(conn, "嘿，你好呀", []) - await startToChat(conn, "嘿，你好呀") - else: - # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） - enqueue_asr_report(conn, original_text, []) - # 否则需要LLM对文字内容进行答复 - await startToChat(conn, original_text) - elif msg_json["type"] == "iot": - conn.logger.bind(tag=TAG).info(f"收到iot消息：{message}") - if "descriptors" in msg_json: - asyncio.create_task(handleIotDescriptors(conn, msg_json["descriptors"])) - if "states" in msg_json: - asyncio.create_task(handleIotStatus(conn, msg_json["states"])) - elif msg_json["type"] == "mcp": - conn.logger.bind(tag=TAG).info(f"收到mcp消息：{message[:100]}") - if "payload" in msg_json: - asyncio.create_task( - handle_mcp_message(conn, conn.mcp_client, msg_json["payload"]) - ) - elif msg_json["type"] == "server": - # 记录日志时过滤敏感信息 - conn.logger.bind(tag=TAG).info( - f"收到服务器消息：{filter_sensitive_info(msg_json)}" - ) - # 如果配置是从API读取的，则需要验证secret - if not conn.read_config_from_api: - return - # 获取post请求的secret - post_secret = msg_json.get("content", {}).get("secret", "") - secret = conn.config["manager-api"].get("secret", "") - # 如果secret不匹配，则返回 - if post_secret != secret: - await conn.websocket.send( - json.dumps( - { - "type": "server", - "status": "error", - "message": "服务器密钥验证失败", - } - ) - ) - return - # 动态更新配置 - if msg_json["action"] == "update_config": - try: - # 更新WebSocketServer的配置 - if not conn.server: - await conn.websocket.send( - json.dumps( - { - "type": "server", - "status": "error", - "message": "无法获取服务器实例", - "content": {"action": "update_config"}, - } - ) - ) - return - - if not await conn.server.update_config(): - await conn.websocket.send( - json.dumps( - { - "type": "server", - "status": "error", - "message": "更新服务器配置失败", - "content": {"action": "update_config"}, - } - ) - ) - return - - # 发送成功响应 - await conn.websocket.send( - json.dumps( - { - "type": "server", - "status": "success", - "message": "配置更新成功", - "content": {"action": "update_config"}, - } - ) - ) - except Exception as e: - conn.logger.bind(tag=TAG).error(f"更新配置失败: {str(e)}") - await conn.websocket.send( - json.dumps( - { - "type": "server", - "status": "error", - "message": f"更新配置失败: {str(e)}", - "content": {"action": "update_config"}, - } - ) - ) - # 重启服务器 - elif msg_json["action"] == "restart": - await conn.handle_restart(msg_json) - else: - conn.logger.bind(tag=TAG).error(f"收到未知类型消息：{message}") - except json.JSONDecodeError: - await conn.websocket.send(message) + await message_processor.process_message(conn, message) diff --git a/main/xiaozhi-server/core/handle/textHandler/abortMessageHandler.py b/main/xiaozhi-server/core/handle/textHandler/abortMessageHandler.py new file mode 100644 index 00000000..dc540d24 --- /dev/null +++ b/main/xiaozhi-server/core/handle/textHandler/abortMessageHandler.py @@ -0,0 +1,16 @@ +from typing import Dict, Any + +from core.handle.abortHandle import handleAbortMessage +from core.handle.textMessageHandler import TextMessageHandler +from core.handle.textMessageType import TextMessageType + + +class AbortTextMessageHandler(TextMessageHandler): + """Abort消息处理器""" + + @property + def message_type(self) -> TextMessageType: + return TextMessageType.ABORT + + async def handle(self, conn, msg_json: Dict[str, Any]) -> None: + await handleAbortMessage(conn) diff --git a/main/xiaozhi-server/core/handle/textHandler/helloMessageHandler.py b/main/xiaozhi-server/core/handle/textHandler/helloMessageHandler.py new file mode 100644 index 00000000..1839814e --- /dev/null +++ b/main/xiaozhi-server/core/handle/textHandler/helloMessageHandler.py @@ -0,0 +1,16 @@ +from typing import Dict, Any + +from core.handle.helloHandle import handleHelloMessage +from core.handle.textMessageHandler import TextMessageHandler +from core.handle.textMessageType import TextMessageType + + +class HelloTextMessageHandler(TextMessageHandler): + """Hello消息处理器""" + + @property + def message_type(self) -> TextMessageType: + return TextMessageType.HELLO + + async def handle(self, conn, msg_json: Dict[str, Any]) -> None: + await handleHelloMessage(conn, msg_json) \ No newline at end of file diff --git a/main/xiaozhi-server/core/handle/textHandler/iotMessageHandler.py b/main/xiaozhi-server/core/handle/textHandler/iotMessageHandler.py new file mode 100644 index 00000000..335d08b0 --- /dev/null +++ b/main/xiaozhi-server/core/handle/textHandler/iotMessageHandler.py @@ -0,0 +1,20 @@ +import asyncio +from typing import Dict, Any + +from core.handle.textMessageHandler import TextMessageHandler +from core.handle.textMessageType import TextMessageType +from core.providers.tools.device_iot import handleIotStatus, handleIotDescriptors + + +class IotTextMessageHandler(TextMessageHandler): + """IOT消息处理器""" + + @property + def message_type(self) -> TextMessageType: + return TextMessageType.IOT + + async def handle(self, conn, msg_json: Dict[str, Any]) -> None: + if "descriptors" in msg_json: + asyncio.create_task(handleIotDescriptors(conn, msg_json["descriptors"])) + if "states" in msg_json: + asyncio.create_task(handleIotStatus(conn, msg_json["states"])) \ No newline at end of file diff --git a/main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py b/main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py new file mode 100644 index 00000000..97286dfe --- /dev/null +++ b/main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py @@ -0,0 +1,63 @@ +import time +from typing import Dict, Any + +from core.handle.receiveAudioHandle import handleAudioMessage, startToChat +from core.handle.reportHandle import enqueue_asr_report +from core.handle.sendAudioHandle import send_stt_message, send_tts_message +from core.handle.textMessageHandler import TextMessageHandler +from core.handle.textMessageType import TextMessageType +from core.utils.util import remove_punctuation_and_length + +TAG = __name__ + +class ListenTextMessageHandler(TextMessageHandler): + """Listen消息处理器""" + + @property + def message_type(self) -> TextMessageType: + return TextMessageType.LISTEN + + async def handle(self, conn, msg_json: Dict[str, Any]) -> None: + if "mode" in msg_json: + conn.client_listen_mode = msg_json["mode"] + conn.logger.bind(tag=TAG).debug( + f"客户端拾音模式：{conn.client_listen_mode}" + ) + if msg_json["state"] == "start": + conn.client_have_voice = True + conn.client_voice_stop = False + elif msg_json["state"] == "stop": + conn.client_have_voice = True + conn.client_voice_stop = True + if len(conn.asr_audio) > 0: + await handleAudioMessage(conn, b"") + elif msg_json["state"] == "detect": + conn.client_have_voice = False + conn.asr_audio.clear() + if "text" in msg_json: + conn.last_activity_time = time.time() * 1000 + original_text = msg_json["text"] # 保留原始文本 + filtered_len, filtered_text = remove_punctuation_and_length( + original_text + ) + + # 识别是否是唤醒词 + is_wakeup_words = filtered_text in conn.config.get("wakeup_words") + # 是否开启唤醒词回复 + enable_greeting = conn.config.get("enable_greeting", True) + + if is_wakeup_words and not enable_greeting: + # 如果是唤醒词，且关闭了唤醒词回复，就不用回答 + await send_stt_message(conn, original_text) + await send_tts_message(conn, "stop", None) + conn.client_is_speaking = False + elif is_wakeup_words: + conn.just_woken_up = True + # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） + enqueue_asr_report(conn, "嘿，你好呀", []) + await startToChat(conn, "嘿，你好呀") + else: + # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） + enqueue_asr_report(conn, original_text, []) + # 否则需要LLM对文字内容进行答复 + await startToChat(conn, original_text) \ No newline at end of file diff --git a/main/xiaozhi-server/core/handle/textHandler/mcpMessageHandler.py b/main/xiaozhi-server/core/handle/textHandler/mcpMessageHandler.py new file mode 100644 index 00000000..65876f24 --- /dev/null +++ b/main/xiaozhi-server/core/handle/textHandler/mcpMessageHandler.py @@ -0,0 +1,20 @@ +import asyncio +from typing import Dict, Any + +from core.handle.textMessageHandler import TextMessageHandler +from core.handle.textMessageType import TextMessageType +from core.providers.tools.device_mcp import handle_mcp_message + + +class McpTextMessageHandler(TextMessageHandler): + """MCP消息处理器""" + + @property + def message_type(self) -> TextMessageType: + return TextMessageType.MCP + + async def handle(self, conn, msg_json: Dict[str, Any]) -> None: + if "payload" in msg_json: + asyncio.create_task( + handle_mcp_message(conn, conn.mcp_client, msg_json["payload"]) + ) \ No newline at end of file diff --git a/main/xiaozhi-server/core/handle/textHandler/serverMessageHandler.py b/main/xiaozhi-server/core/handle/textHandler/serverMessageHandler.py new file mode 100644 index 00000000..b9a23588 --- /dev/null +++ b/main/xiaozhi-server/core/handle/textHandler/serverMessageHandler.py @@ -0,0 +1,92 @@ +import asyncio +import json +from typing import Dict, Any + +from core.handle.textMessageHandler import TextMessageHandler +from core.handle.textMessageType import TextMessageType +from core.providers.tools.device_mcp import handle_mcp_message + +TAG = __name__ + +class ServerTextMessageHandler(TextMessageHandler): + """MCP消息处理器""" + + @property + def message_type(self) -> TextMessageType: + return TextMessageType.SERVER + + async def handle(self, conn, msg_json: Dict[str, Any]) -> None: + # 如果配置是从API读取的，则需要验证secret + if not conn.read_config_from_api: + return + # 获取post请求的secret + post_secret = msg_json.get("content", {}).get("secret", "") + secret = conn.config["manager-api"].get("secret", "") + # 如果secret不匹配，则返回 + if post_secret != secret: + await conn.websocket.send( + json.dumps( + { + "type": "server", + "status": "error", + "message": "服务器密钥验证失败", + } + ) + ) + return + # 动态更新配置 + if msg_json["action"] == "update_config": + try: + # 更新WebSocketServer的配置 + if not conn.server: + await conn.websocket.send( + json.dumps( + { + "type": "server", + "status": "error", + "message": "无法获取服务器实例", + "content": {"action": "update_config"}, + } + ) + ) + return + + if not await conn.server.update_config(): + await conn.websocket.send( + json.dumps( + { + "type": "server", + "status": "error", + "message": "更新服务器配置失败", + "content": {"action": "update_config"}, + } + ) + ) + return + + # 发送成功响应 + await conn.websocket.send( + json.dumps( + { + "type": "server", + "status": "success", + "message": "配置更新成功", + "content": {"action": "update_config"}, + } + ) + ) + except Exception as e: + conn.logger.bind(tag=TAG).error(f"更新配置失败: {str(e)}") + await conn.websocket.send( + json.dumps( + { + "type": "server", + "status": "error", + "message": f"更新配置失败: {str(e)}", + "content": {"action": "update_config"}, + } + ) + ) + # 重启服务器 + elif msg_json["action"] == "restart": + await conn.handle_restart(msg_json) \ No newline at end of file diff --git a/main/xiaozhi-server/core/handle/textMessageHandler.py b/main/xiaozhi-server/core/handle/textMessageHandler.py new file mode 100644 index 00000000..f94a0bac --- /dev/null +++ b/main/xiaozhi-server/core/handle/textMessageHandler.py @@ -0,0 +1,21 @@ +from abc import abstractmethod, ABC +from typing import Dict, Any + +from core.handle.textMessageType import TextMessageType + +TAG = __name__ + + +class TextMessageHandler(ABC): + """消息处理器抽象基类""" + + @abstractmethod + async def handle(self, conn, msg_json: Dict[str, Any]) -> None: + """处理消息的抽象方法""" + pass + + @property + @abstractmethod + def message_type(self) -> TextMessageType: + """返回处理的消息类型""" + pass diff --git a/main/xiaozhi-server/core/handle/textMessageHandlerRegistry.py b/main/xiaozhi-server/core/handle/textMessageHandlerRegistry.py new file mode 100644 index 00000000..e90d7231 --- /dev/null +++ b/main/xiaozhi-server/core/handle/textMessageHandlerRegistry.py @@ -0,0 +1,45 @@ +from typing import Dict, Optional + +from core.handle.textHandler.abortMessageHandler import AbortTextMessageHandler +from core.handle.textHandler.helloMessageHandler import HelloTextMessageHandler +from core.handle.textHandler.iotMessageHandler import IotTextMessageHandler +from core.handle.textHandler.listenMessageHandler import ListenTextMessageHandler +from core.handle.textHandler.mcpMessageHandler import McpTextMessageHandler +from core.handle.textMessageHandler import TextMessageHandler +from core.handle.textHandler.serverMessageHandler import ServerTextMessageHandler + +TAG = __name__ + + +class TextMessageHandlerRegistry: + """消息处理器注册表""" + + def __init__(self): + self._handlers: Dict[str, TextMessageHandler] = {} + self._register_default_handlers() + + def _register_default_handlers(self) -> None: + """注册默认的消息处理器""" + handlers = [ + HelloTextMessageHandler(), + AbortTextMessageHandler(), + ListenTextMessageHandler(), + IotTextMessageHandler(), + McpTextMessageHandler(), + ServerTextMessageHandler(), + ] + + for handler in handlers: + self.register_handler(handler) + + def register_handler(self, handler: TextMessageHandler) -> None: + """注册消息处理器""" + self._handlers[handler.message_type.value] = handler + + def get_handler(self, message_type: str) -> Optional[TextMessageHandler]: + """获取消息处理器""" + return self._handlers.get(message_type) + + def get_supported_types(self) -> list: + """获取支持的消息类型""" + return list(self._handlers.keys()) diff --git a/main/xiaozhi-server/core/handle/textMessageProcessor.py b/main/xiaozhi-server/core/handle/textMessageProcessor.py new file mode 100644 index 00000000..0cae5e09 --- /dev/null +++ b/main/xiaozhi-server/core/handle/textMessageProcessor.py @@ -0,0 +1,41 @@ +import json + +from core.handle.textMessageHandlerRegistry import TextMessageHandlerRegistry + +TAG = __name__ + + +class TextMessageProcessor: + """消息处理器主类""" + + def __init__(self, registry: TextMessageHandlerRegistry): + self.registry = registry + + async def process_message(self, conn, message: str) -> None: + """处理消息的主入口""" + try: + # 解析JSON消息 + msg_json = json.loads(message) + + # 处理JSON消息 + if isinstance(msg_json, dict): + message_type = msg_json.get("type") + + # 记录日志 + conn.logger.bind(tag=TAG).info(f"收到{message_type}消息：{message}") + + # 获取并执行处理器 + handler = self.registry.get_handler(message_type) + if handler: + await handler.handle(conn, msg_json) + else: + conn.logger.bind(tag=TAG).error(f"收到未知类型消息：{message}") + # 处理纯数字消息 + elif isinstance(msg_json, int): + conn.logger.bind(tag=TAG).info(f"收到数字消息：{message}") + await conn.websocket.send(message) + + except json.JSONDecodeError: + # 非JSON消息直接转发 + conn.logger.bind(tag=TAG).error(f"解析到错误的消息：{message}") + await conn.websocket.send(message) diff --git a/main/xiaozhi-server/core/handle/textMessageType.py b/main/xiaozhi-server/core/handle/textMessageType.py new file mode 100644 index 00000000..53e71b71 --- /dev/null +++ b/main/xiaozhi-server/core/handle/textMessageType.py @@ -0,0 +1,11 @@ +from enum import Enum + + +class TextMessageType(Enum): + """消息类型枚举""" + HELLO = "hello" + ABORT = "abort" + LISTEN = "listen" + IOT = "iot" + MCP = "mcp" + SERVER = "server"