gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构并增强Napcat适配器的功能

Browse Source 
- 更新了`BaseAdapter`以简化子进程处理。
- 对`AdapterManager`进行了重构，以便根据适配器的`run_in_subprocess`属性来管理适配器。
- 增强了`NapcatAdapter`，以利用新的`CoreSinkManager`实现更优的进程管理。
- 在`utils.py`中实现了针对群组和成员信息的缓存机制。
- 改进了`message_handler.py`中的消息处理，以支持各种消息类型和格式。
- 已将插件配置版本更新至7.8.3。
This commit is contained in:
Windpicker-owo
2025-11-25 19:55:36 +08:00
parent 1ebdc37b22
commit 6b3b2a8245
38 changed files with 2082 additions and 3277 deletions
Download Patch File Download Diff File Expand all files Collapse all files

488
src/chat/message_receive/bot.py
View File

@@ -1,488 +0,0 @@
import os
import re
import traceback
from typing import Any
from mofox_bus.runtime import MessageRuntime
from mofox_bus import MessageEnvelope
from src.chat.message_manager import message_manager
from src.chat.message_receive.chat_stream import ChatStream, get_chat_manager
from src.chat.message_receive.storage import MessageStorage
from src.chat.utils.prompt import global_prompt_manager
from src.chat.utils.utils import is_mentioned_bot_in_message
from src.common.data_models.database_data_model import DatabaseMessages
from src.common.logger import get_logger
from src.config.config import global_config
from src.mood.mood_manager import mood_manager # 导入情绪管理器
from src.plugin_system.base import BaseCommand, EventType
from src.plugin_system.core import component_registry, event_manager, global_announcement_manager
# 获取项目根目录（假设本文件在src/chat/message_receive/下，根目录为上上上级目录）
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
# 配置主程序日志格式
logger = get_logger("chat")
def _check_ban_words(text: str, chat: ChatStream, userinfo: UserInfo) -> bool:
"""检查消息是否包含过滤词
Args:
text: 待检查的文本
chat: 聊天对象
userinfo: 用户信息
Returns:
bool: 是否包含过滤词
"""
for word in global_config.message_receive.ban_words:
if word in text:
chat_name = chat.group_info.group_name if chat.group_info else "私聊"
logger.info(f"[{chat_name}]{userinfo.user_nickname}:{text}")
logger.info(f"[过滤词识别]消息中含有{word}，filtered")
return True
return False
def _check_ban_regex(text: str, chat: ChatStream, userinfo: UserInfo) -> bool:
"""检查消息是否匹配过滤正则表达式
Args:
text: 待检查的文本
chat: 聊天对象
userinfo: 用户信息
Returns:
bool: 是否匹配过滤正则
"""
for pattern in global_config.message_receive.ban_msgs_regex:
if re.search(pattern, text):
chat_name = chat.group_info.group_name if chat.group_info else "私聊"
logger.info(f"[{chat_name}]{userinfo.user_nickname}:{text}")
logger.info(f"[正则表达式过滤]消息匹配到{pattern}，filtered")
return True
return False
runtime = MessageRuntime() # 获取mofox-bus运行时环境
class ChatBot:
def __init__(self):
self.bot = None # bot 实例引用
self._started = False
self.mood_manager = mood_manager # 获取情绪管理器单例
# 启动消息管理器
self._message_manager_started = False
async def _ensure_started(self):
"""确保所有任务已启动"""
if not self._started:
logger.debug("确保ChatBot所有任务已启动")
# 启动消息管理器
if not self._message_manager_started:
await message_manager.start()
self._message_manager_started = True
logger.info("消息管理器已启动")
self._started = True
async def _process_plus_commands(self, message: DatabaseMessages, chat: ChatStream):
"""独立处理PlusCommand系统"""
try:
text = message.processed_plain_text or ""
# 获取配置的命令前缀
from src.config.config import global_config
prefixes = global_config.command.command_prefixes
# 检查是否以任何前缀开头
matched_prefix = None
for prefix in prefixes:
if text.startswith(prefix):
matched_prefix = prefix
break
if not matched_prefix:
return False, None, True # 不是命令，继续处理
# 移除前缀
command_part = text[len(matched_prefix) :].strip()
# 分离命令名和参数
parts = command_part.split(None, 1)
if not parts:
return False, None, True # 没有命令名，继续处理
command_word = parts[0].lower()
args_text = parts[1] if len(parts) > 1 else ""
# 查找匹配的PlusCommand
plus_command_registry = component_registry.get_plus_command_registry()
matching_commands = []
for plus_command_name, plus_command_class in plus_command_registry.items():
plus_command_info = component_registry.get_registered_plus_command_info(plus_command_name)
if not plus_command_info:
continue
# 检查命令名是否匹配（命令名和别名）
all_commands = [plus_command_name.lower()] + [
alias.lower() for alias in plus_command_info.command_aliases
]
if command_word in all_commands:
matching_commands.append((plus_command_class, plus_command_info, plus_command_name))
if not matching_commands:
return False, None, True # 没有找到匹配的PlusCommand，继续处理
# 如果有多个匹配，按优先级排序
if len(matching_commands) > 1:
matching_commands.sort(key=lambda x: x[1].priority, reverse=True)
logger.warning(
f"文本 '{text}' 匹配到多个PlusCommand: {[cmd[2] for cmd in matching_commands]}，使用优先级最高的"
)
plus_command_class, plus_command_info, plus_command_name = matching_commands[0]
# 检查命令是否被禁用
if (
chat
and chat.stream_id
and plus_command_name
in global_announcement_manager.get_disabled_chat_commands(chat.stream_id)
):
logger.info("用户禁用的PlusCommand，跳过处理")
return False, None, True
message.is_command = True
# 获取插件配置
plugin_config = component_registry.get_plugin_config(plus_command_name)
# 创建PlusCommand实例
plus_command_instance = plus_command_class(message, plugin_config)
# 为插件实例设置 chat_stream 运行时属性
setattr(plus_command_instance, "chat_stream", chat)
try:
# 检查聊天类型限制
if not plus_command_instance.is_chat_type_allowed():
is_group = chat.group_info is not None
logger.info(
f"PlusCommand {plus_command_class.__name__} 不支持当前聊天类型: {'群聊' if is_group else '私聊'}"
)
return False, None, True # 跳过此命令，继续处理其他消息
# 设置参数
from src.plugin_system.base.command_args import CommandArgs
command_args = CommandArgs(args_text)
plus_command_instance.args = command_args
# 执行命令
success, response, intercept_message = await plus_command_instance.execute(command_args)
# 记录命令执行结果
if success:
logger.info(f"PlusCommand执行成功: {plus_command_class.__name__} (拦截: {intercept_message})")
else:
logger.warning(f"PlusCommand执行失败: {plus_command_class.__name__} - {response}")
# 根据命令的拦截设置决定是否继续处理消息
return True, response, not intercept_message # 找到命令，根据intercept_message决定是否继续
except Exception as e:
logger.error(f"执行PlusCommand时出错: {plus_command_class.__name__} - {e}")
logger.error(traceback.format_exc())
try:
await plus_command_instance.send_text(f"命令执行出错: {e!s}")
except Exception as send_error:
logger.error(f"发送错误消息失败: {send_error}")
# 命令出错时，根据命令的拦截设置决定是否继续处理消息
return True, str(e), False # 出错时继续处理消息
except Exception as e:
logger.error(f"处理PlusCommand时出错: {e}")
return False, None, True # 出错时继续处理消息
async def _process_commands_with_new_system(self, message: DatabaseMessages, chat: ChatStream):
# sourcery skip: use-named-expression
"""使用新插件系统处理命令"""
try:
text = message.processed_plain_text or ""
# 使用新的组件注册中心查找命令
command_result = component_registry.find_command_by_text(text)
if command_result:
command_class, matched_groups, command_info = command_result
plugin_name = command_info.plugin_name
command_name = command_info.name
if (
chat
and chat.stream_id
and command_name
in global_announcement_manager.get_disabled_chat_commands(chat.stream_id)
):
logger.info("用户禁用的命令，跳过处理")
return False, None, True
message.is_command = True
# 获取插件配置
plugin_config = component_registry.get_plugin_config(plugin_name)
# 创建命令实例
command_instance: BaseCommand = command_class(message, plugin_config)
command_instance.set_matched_groups(matched_groups)
# 为插件实例设置 chat_stream 运行时属性
setattr(command_instance, "chat_stream", chat)
try:
# 检查聊天类型限制
if not command_instance.is_chat_type_allowed():
is_group = chat.group_info is not None
logger.info(
f"命令 {command_class.__name__} 不支持当前聊天类型: {'群聊' if is_group else '私聊'}"
)
return False, None, True # 跳过此命令，继续处理其他消息
# 执行命令
success, response, intercept_message = await command_instance.execute()
# 记录命令执行结果
if success:
logger.info(f"命令执行成功: {command_class.__name__} (拦截: {intercept_message})")
else:
logger.warning(f"命令执行失败: {command_class.__name__} - {response}")
# 根据命令的拦截设置决定是否继续处理消息
return True, response, not intercept_message # 找到命令，根据intercept_message决定是否继续
except Exception as e:
logger.error(f"执行命令时出错: {command_class.__name__} - {e}")
logger.error(traceback.format_exc())
try:
await command_instance.send_text(f"命令执行出错: {e!s}")
except Exception as send_error:
logger.error(f"发送错误消息失败: {send_error}")
# 命令出错时，根据命令的拦截设置决定是否继续处理消息
return True, str(e), False # 出错时继续处理消息
# 没有找到命令，继续处理消息
return False, None, True
except Exception as e:
logger.error(f"处理命令时出错: {e}")
return False, None, True # 出错时继续处理消息
async def _handle_adapter_response_from_dict(self, seg_data: dict | None):
"""处理适配器命令响应（从字典数据）"""
try:
from src.plugin_system.apis.send_api import put_adapter_response
if isinstance(seg_data, dict):
request_id = seg_data.get("request_id")
response_data = seg_data.get("response")
else:
request_id = None
response_data = None
if request_id and response_data:
logger.info(f"[DEBUG bot.py] 收到适配器响应，request_id={request_id}")
put_adapter_response(request_id, response_data)
else:
logger.warning(f"适配器响应消息格式不正确: request_id={request_id}, response_data={response_data}")
except Exception as e:
logger.error(f"处理适配器响应时出错: {e}")
@runtime.on_message
async def message_process(self, envelope: MessageEnvelope) -> None:
"""处理转化后的统一格式消息"""
# 控制握手等消息可能缺少 message_info，这里直接跳过避免 KeyError
message_info = envelope.get("message_info")
if not isinstance(message_info, dict):
logger.debug(
"收到缺少 message_info 的消息，已跳过。可用字段: %s",
", ".join(envelope.keys()),
)
return
if message_info.get("group_info") is not None:
message_info["group_info"]["group_id"] = str( # type: ignore
message_info["group_info"]["group_id"] # type: ignore
)
if message_info.get("user_info") is not None:
message_info["user_info"]["user_id"] = str( # type: ignore
message_info["user_info"]["user_id"] # type: ignore
)
# 优先处理adapter_response消息（在echo检查之前！）
message_segment = envelope.get("message_segment")
if message_segment and isinstance(message_segment, dict):
if message_segment.get("type") == "adapter_response":
logger.info("[DEBUG bot.py message_process] 检测到adapter_response，立即处理")
await self._handle_adapter_response_from_dict(message_segment.get("data"))
return
# 先提取基础信息检查是否是自身消息上报
from mofox_bus import BaseMessageInfo
temp_message_info = BaseMessageInfo.from_dict(message_data.get("message_info", {}))
if temp_message_info.additional_config:
sent_message = temp_message_info.additional_config.get("echo", False)
if sent_message: # 这一段只是为了在一切处理前劫持上报的自身消息，用于更新message_id，需要ada支持上报事件，实际测试中不会对正常使用造成任何问题
# 直接使用消息字典更新，不再需要创建 MessageRecv
await MessageStorage.update_message(message_data)
return
message_segment = envelope.get("message_segment")
group_info = temp_message_info.group_info
user_info = temp_message_info.user_info
# 获取或创建聊天流
chat = await get_chat_manager().get_or_create_stream(
platform=temp_message_info.platform, # type: ignore
user_info=user_info, # type: ignore
group_info=group_info,
)
# 使用新的消息处理器直接生成 DatabaseMessages
from src.chat.message_receive.message_processor import process_message_from_dict
message = await process_message_from_dict(
message_dict=envelope,
stream_id=chat.stream_id,
platform=chat.platform
)
# 填充聊天流时间信息
message.chat_info.create_time = chat.create_time
message.chat_info.last_active_time = chat.last_active_time
# 注册消息到聊天管理器
get_chat_manager().register_message(message)
# 检测是否提及机器人
message.is_mentioned, _ = is_mentioned_bot_in_message(message)
# 在这里打印[所见]日志，确保在所有处理和过滤之前记录
chat_name = chat.group_info.group_name if chat.group_info else "私聊"
user_nickname = message.user_info.user_nickname if message.user_info else "未知用户"
logger.info(
f"[{chat_name}]{user_nickname}:{message.processed_plain_text}\u001b[0m"
)
# 在此添加硬编码过滤，防止回复图片处理失败的消息
failure_keywords = ["[表情包(描述生成失败)]", "[图片(描述生成失败)]"]
processed_text = message.processed_plain_text or ""
if any(keyword in processed_text for keyword in failure_keywords):
logger.info(f"[硬编码过滤] 检测到媒体内容处理失败（{processed_text}），消息被静默处理。")
return
# 过滤检查
# DatabaseMessages 使用 display_message 作为原始消息表示
raw_text = message.display_message or message.processed_plain_text or ""
if _check_ban_words(message.processed_plain_text, chat, user_info) or _check_ban_regex( # type: ignore
raw_text,
chat,
user_info, # type: ignore
):
return
# 命令处理 - 首先尝试PlusCommand独立处理
is_plus_command, plus_cmd_result, plus_continue_process = await self._process_plus_commands(message, chat)
# 如果是PlusCommand且不需要继续处理，则直接返回
if is_plus_command and not plus_continue_process:
await MessageStorage.store_message(message, chat)
logger.info(f"PlusCommand处理完成，跳过后续消息处理: {plus_cmd_result}")
return
# 如果不是PlusCommand，尝试传统的BaseCommand处理
if not is_plus_command:
is_command, cmd_result, continue_process = await self._process_commands_with_new_system(message, chat)
# 如果是命令且不需要继续处理，则直接返回
if is_command and not continue_process:
await MessageStorage.store_message(message, chat)
logger.info(f"命令处理完成，跳过后续消息处理: {cmd_result}")
return
result = await event_manager.trigger_event(EventType.ON_MESSAGE, permission_group="SYSTEM", message=message)
if result and not result.all_continue_process():
raise UserWarning(f"插件{result.get_summary().get('stopped_handlers', '')}于消息到达时取消了消息处理")
# TODO:暂不可用 - DatabaseMessages 不再有 message_info.template_info
# 确认从接口发来的message是否有自定义的prompt模板信息
# 这个功能需要在 adapter 层通过 additional_config 传递
template_group_name = None
async def preprocess():
# message 已经是 DatabaseMessages，直接使用
group_info = chat.group_info
# 先交给消息管理器处理，计算兴趣度等衍生数据
try:
# 在将消息添加到管理器之前进行最终的静默检查
should_process_in_manager = True
if group_info and str(group_info.group_id) in global_config.message_receive.mute_group_list:
# 检查消息是否为图片或表情包
is_image_or_emoji = message.is_picid or message.is_emoji
if not message.is_mentioned and not is_image_or_emoji:
logger.debug(f"群组 {group_info.group_id} 在静默列表中，且消息不是@、回复或图片/表情包，跳过消息管理器处理")
should_process_in_manager = False
elif is_image_or_emoji:
logger.debug(f"群组 {group_info.group_id} 在静默列表中，但消息是图片/表情包，静默处理")
should_process_in_manager = False
if should_process_in_manager:
await message_manager.add_message(chat.stream_id, message)
logger.debug(f"消息已添加到消息管理器: {chat.stream_id}")
except Exception as e:
logger.error(f"消息添加到消息管理器失败: {e}")
# 存储消息到数据库，只进行一次写入
try:
await MessageStorage.store_message(message, chat)
except Exception as e:
logger.error(f"存储消息到数据库失败: {e}")
traceback.print_exc()
# 情绪系统更新 - 在消息存储后触发情绪更新
try:
if global_config.mood.enable_mood:
# 获取兴趣度用于情绪更新
interest_rate = message.interest_value
if interest_rate is None:
interest_rate = 0.0
logger.debug(f"开始更新情绪状态，兴趣度: {interest_rate:.2f}")
# 获取当前聊天的情绪对象并更新情绪状态
chat_mood = mood_manager.get_mood_by_chat_id(chat.stream_id)
await chat_mood.update_mood_by_message(message, interest_rate)
logger.debug("情绪状态更新完成")
except Exception as e:
logger.error(f"更新情绪状态失败: {e}")
traceback.print_exc()
if template_group_name:
async with global_prompt_manager.async_message_scope(template_group_name):
await preprocess()
else:
await preprocess()
except Exception as e:
logger.error(f"预处理消息失败: {e}")
traceback.print_exc()
# 创建全局ChatBot实例
chat_bot = ChatBot()

16
src/chat/message_receive/chat_stream.py
View File

@@ -2,11 +2,11 @@ import asyncio
import hashlib
import time
from mofox_bus import GroupInfo, UserInfo
from rich.traceback import install
from sqlalchemy.dialects.mysql import insert as mysql_insert
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
from src.common.data_models.database_data_model import DatabaseGroupInfo,DatabaseUserInfo
from src.common.data_models.database_data_model import DatabaseMessages
from src.common.data_models.message_manager_data_model import StreamContext
from src.plugin_system.base.component_types import ChatMode, ChatType
@@ -30,8 +30,8 @@ class ChatStream:
self,
stream_id: str,
platform: str,
user_info: UserInfo | None = None,
group_info: GroupInfo | None = None,
user_info: DatabaseUserInfo | None = None,
group_info: DatabaseGroupInfo | None = None,
data: dict | None = None,
):
self.stream_id = stream_id
@@ -77,8 +77,8 @@ class ChatStream:
@classmethod
def from_dict(cls, data: dict) -> "ChatStream":
"""从字典创建实例"""
user_info = UserInfo.from_dict(data.get("user_info", {})) if data.get("user_info") else None
group_info = GroupInfo.from_dict(data.get("group_info", {})) if data.get("group_info") else None
user_info = DatabaseUserInfo.from_dict(data.get("user_info", {})) if data.get("user_info") else None
group_info = DatabaseGroupInfo.from_dict(data.get("group_info", {})) if data.get("group_info") else None
instance = cls(
stream_id=data["stream_id"],
@@ -369,7 +369,7 @@ class ChatManager:
# logger.debug(f"注册消息到聊天流: {stream_id}")
@staticmethod
def _generate_stream_id(platform: str, user_info: UserInfo | None, group_info: GroupInfo | None = None) -> str:
def _generate_stream_id(platform: str, user_info: DatabaseUserInfo | None, group_info: DatabaseGroupInfo | None = None) -> str:
"""生成聊天流唯一ID"""
if not user_info and not group_info:
raise ValueError("用户信息或群组信息必须提供")
@@ -392,7 +392,7 @@ class ChatManager:
return hashlib.sha256(key.encode()).hexdigest()
async def get_or_create_stream(
self, platform: str, user_info: UserInfo, group_info: GroupInfo | None = None
self, platform: str, user_info: DatabaseUserInfo, group_info: DatabaseGroupInfo | None = None
) -> ChatStream:
"""获取或创建聊天流 - 优化版本使用缓存机制"""
try:
@@ -483,7 +483,7 @@ class ChatManager:
return stream
def get_stream_by_info(
self, platform: str, user_info: UserInfo, group_info: GroupInfo | None = None
self, platform: str, user_info: DatabaseUserInfo, group_info: DatabaseGroupInfo | None = None
) -> ChatStream | None:
"""通过信息获取聊天流"""
stream_id = self._generate_stream_id(platform, user_info, group_info)

6
src/chat/message_receive/message.py
View File

@@ -1,12 +1,11 @@
import time
from abc import ABCMeta, abstractmethod
from dataclasses import dataclass
from typing import Optional
from typing import Optional, TYPE_CHECKING
import urllib3
from rich.traceback import install
from src.chat.message_receive.chat_stream import ChatStream
from src.chat.utils.self_voice_cache import consume_self_voice_text
from src.chat.utils.utils_image import get_image_manager
from src.chat.utils.utils_voice import get_voice_text
@@ -14,6 +13,9 @@ from src.common.data_models.database_data_model import DatabaseMessages
from src.common.logger import get_logger
from src.config.config import global_config
if TYPE_CHECKING:
from src.chat.message_receive.chat_stream import ChatStream
install(extra_lines=3)

678
src/chat/message_receive/message_handler.py
View File

@@ -1,39 +1,566 @@
import os
import traceback
"""
统一消息处理器 (Message Handler)
利用 mofox_bus.MessageRuntime 的路由功能，简化消息处理链条：
1. 使用 @runtime.on_message() 装饰器注册按消息类型路由的处理器
2. 使用 before_hook 进行消息预处理（ID标准化、过滤等）
3. 使用 after_hook 进行消息后处理（存储、情绪更新等）
4. 使用 error_hook 统一处理异常
消息流向：
适配器 → CoreSinkManager → MessageRuntime
↓
[before_hook] 消息预处理、过滤
↓
[on_message] 按类型路由处理（命令、普通消息等）
↓
[after_hook] 存储、情绪更新等
↓
回复生成 → CoreSinkManager.send_outgoing() → 适配器
重构说明（2025-11）：
- 移除手动的消息处理链，改用 MessageRuntime 路由
- MessageHandler 变成处理器注册器，在初始化时注册各种处理器
- 利用 runtime 的钩子机制简化前置/后置处理
"""
from __future__ import annotations
import asyncio
import os
import re
import time
import traceback
from functools import partial
from typing import TYPE_CHECKING, Any
from mofox_bus import MessageEnvelope, MessageRuntime
from mofox_bus.runtime import MessageRuntime
from mofox_bus import MessageEnvelope
from src.chat.message_manager import message_manager
from src.chat.message_receive.chat_stream import get_chat_manager
from src.chat.message_receive.storage import MessageStorage
from src.chat.utils.prompt import global_prompt_manager
from src.chat.utils.utils import is_mentioned_bot_in_message
from src.common.data_models.database_data_model import DatabaseMessages
from src.common.logger import get_logger
from src.config.config import global_config
from src.chat.message_receive.chat_stream import ChatStream, get_chat_manager
from src.common.data_models.database_data_model import DatabaseGroupInfo, DatabaseUserInfo, DatabaseMessages
from src.mood.mood_manager import mood_manager
from src.plugin_system.base import BaseCommand, EventType
from src.plugin_system.core import component_registry, event_manager, global_announcement_manager
runtime = MessageRuntime()
if TYPE_CHECKING:
from src.common.core_sink_manager import CoreSinkManager
from src.chat.message_receive.chat_stream import ChatStream
# 获取项目根目录（假设本文件在src/chat/message_receive/下，根目录为上上上级目录）
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
logger = get_logger("message_handler")
# 项目根目录
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))
def _check_ban_words(text: str, chat: "ChatStream", userinfo) -> bool:
"""检查消息是否包含过滤词"""
for word in global_config.message_receive.ban_words:
if word in text:
chat_name = chat.group_info.group_name if chat.group_info else "私聊"
logger.info(f"[{chat_name}]{userinfo.user_nickname}:{text}")
logger.info(f"[过滤词识别]消息中含有{word}，filtered")
return True
return False
def _check_ban_regex(text: str, chat: "ChatStream", userinfo) -> bool:
"""检查消息是否匹配过滤正则表达式"""
for pattern in global_config.message_receive.ban_msgs_regex:
if re.search(pattern, text):
chat_name = chat.group_info.group_name if chat.group_info else "私聊"
logger.info(f"[{chat_name}]{userinfo.user_nickname}:{text}")
logger.info(f"[正则表达式过滤]消息匹配到{pattern}，filtered")
return True
return False
# 配置主程序日志格式
logger = get_logger("chat")
class MessageHandler:
"""
统一消息处理器
利用 MessageRuntime 的路由功能，将消息处理逻辑注册为路由和钩子。
架构说明：
- 在 register_handlers() 中向 MessageRuntime 注册各种处理器
- 使用 @runtime.on_message(message_type=...) 按消息类型路由
- 使用 before_hook 进行消息预处理
- 使用 after_hook 进行消息后处理
- 使用 error_hook 统一处理异常
主要功能：
1. 消息预处理：ID标准化、过滤检查
2. 适配器响应处理：处理 adapter_response 类型消息
3. 命令处理：PlusCommand 和 BaseCommand
4. 普通消息处理：触发事件、存储、情绪更新
"""
def __init__(self):
self._started = False
self._message_manager_started = False
self._core_sink_manager: CoreSinkManager | None = None
self._shutting_down = False
self._runtime: MessageRuntime | None = None
async def preprocess(self, chat: ChatStream, message: DatabaseMessages):
# message 已经是 DatabaseMessages，直接使用
group_info = chat.group_info
def set_core_sink_manager(self, manager: "CoreSinkManager") -> None:
"""设置 CoreSinkManager 引用"""
self._core_sink_manager = manager
# 先交给消息管理器处理
def register_handlers(self, runtime: MessageRuntime) -> None:
"""
向 MessageRuntime 注册消息处理器和钩子
这是核心方法，在系统初始化时调用，将所有处理逻辑注册到 runtime。
Args:
runtime: MessageRuntime 实例
"""
self._runtime = runtime
# 注册前置钩子：消息预处理和过滤
runtime.register_before_hook(self._before_hook)
# 注册后置钩子：存储、情绪更新等
runtime.register_after_hook(self._after_hook)
# 注册错误钩子：统一异常处理
runtime.register_error_hook(self._error_hook)
# 注册适配器响应处理器（最高优先级）
def _is_adapter_response(env: MessageEnvelope) -> bool:
segment = env.get("message_segment")
if isinstance(segment, dict):
return segment.get("type") == "adapter_response"
return False
runtime.add_route(
predicate=_is_adapter_response,
handler=self._handle_adapter_response_route,
name="adapter_response_handler",
message_type="adapter_response",
)
# 注册默认消息处理器（处理所有其他消息）
runtime.add_route(
predicate=lambda _: True, # 匹配所有消息
handler=self._handle_normal_message,
name="default_message_handler",
)
logger.info("MessageHandler 已向 MessageRuntime 注册处理器和钩子")
async def ensure_started(self) -> None:
"""确保所有依赖任务已启动"""
if not self._started:
logger.debug("确保 MessageHandler 所有任务已启动")
# 启动消息管理器
if not self._message_manager_started:
await message_manager.start()
self._message_manager_started = True
logger.info("消息管理器已启动")
self._started = True
async def _before_hook(self, envelope: MessageEnvelope) -> None:
"""
前置钩子：消息预处理
1. 标准化 ID 为字符串
2. 检查是否为 echo 消息（自身发送的消息上报）
3. 附加预处理数据到 envelope（chat_stream, message 等）
"""
if self._shutting_down:
raise UserWarning("系统正在关闭，拒绝处理消息")
# 确保依赖服务已启动
await self.ensure_started()
# 提取消息信息
message_info = envelope.get("message_info")
if not isinstance(message_info, dict):
logger.debug(
"收到缺少 message_info 的消息，已跳过。可用字段: %s",
", ".join(envelope.keys()),
)
raise UserWarning("消息缺少 message_info")
# 标准化 ID 为字符串
if message_info.get("group_info") is not None:
message_info["group_info"]["group_id"] = str( # type: ignore
message_info["group_info"]["group_id"] # type: ignore
)
if message_info.get("user_info") is not None:
message_info["user_info"]["user_id"] = str( # type: ignore
message_info["user_info"]["user_id"] # type: ignore
)
# 处理自身消息上报（echo）
additional_config = message_info.get("additional_config", {})
if additional_config and isinstance(additional_config, dict):
sent_message = additional_config.get("echo", False)
if sent_message:
# 更新消息ID
await MessageStorage.update_message(dict(envelope))
raise UserWarning("Echo 消息已处理")
async def _after_hook(self, envelope: MessageEnvelope) -> None:
"""
后置钩子：消息后处理
在消息处理完成后执行的清理工作
"""
# 后置处理逻辑（如有需要）
pass
async def _error_hook(self, envelope: MessageEnvelope, exc: BaseException) -> None:
"""
错误钩子：统一异常处理
"""
if isinstance(exc, UserWarning):
# UserWarning 是预期的流程控制，只记录 debug 日志
logger.debug(f"消息处理流程控制: {exc}")
else:
message_id = envelope.get("message_info", {}).get("message_id", "UNKNOWN")
logger.error(f"处理消息 {message_id} 时出错: {exc}", exc_info=True)
async def _handle_adapter_response_route(self, envelope: MessageEnvelope) -> MessageEnvelope | None:
"""
处理适配器响应消息的路由处理器
"""
message_segment = envelope.get("message_segment")
if message_segment and isinstance(message_segment, dict):
seg_data = message_segment.get("data")
if isinstance(seg_data, dict):
await self._handle_adapter_response(seg_data)
return None
async def _handle_normal_message(self, envelope: MessageEnvelope) -> MessageEnvelope | None:
"""
默认消息处理器：处理普通消息
1. 获取或创建聊天流
2. 转换为 DatabaseMessages
3. 过滤检查
4. 命令处理
5. 触发事件、存储、情绪更新
"""
try:
# 在将消息添加到管理器之前进行最终的静默检查
message_info = envelope.get("message_info")
if not isinstance(message_info, dict):
return None
# 获取用户和群组信息
group_info = message_info.get("group_info")
user_info = message_info.get("user_info")
# 获取或创建聊天流
platform = message_info.get("platform", "unknown")
chat = await get_chat_manager().get_or_create_stream(
platform=platform,
user_info=user_info, # type: ignore
group_info=group_info,
)
# 将消息信封转换为 DatabaseMessages
from src.chat.message_receive.message_processor import process_message_from_dict
message = await process_message_from_dict(
message_dict=envelope,
stream_id=chat.stream_id,
platform=chat.platform
)
# 填充聊天流时间信息
message.chat_info.create_time = chat.create_time
message.chat_info.last_active_time = chat.last_active_time
# 注册消息到聊天管理器
get_chat_manager().register_message(message)
# 检测是否提及机器人
message.is_mentioned, _ = is_mentioned_bot_in_message(message)
# 打印接收日志
chat_name = chat.group_info.group_name if chat.group_info else "私聊"
user_nickname = message.user_info.user_nickname if message.user_info else "未知用户"
logger.info(f"[{chat_name}]{user_nickname}:{message.processed_plain_text}\u001b[0m")
# 硬编码过滤
failure_keywords = ["[表情包(描述生成失败)]", "[图片(描述生成失败)]"]
processed_text = message.processed_plain_text or ""
if any(keyword in processed_text for keyword in failure_keywords):
logger.info(f"[硬编码过滤] 检测到媒体内容处理失败（{processed_text}），消息被静默处理。")
return None
# 过滤检查
raw_text = message.display_message or message.processed_plain_text or ""
if _check_ban_words(processed_text, chat, user_info) or _check_ban_regex(
raw_text, chat, user_info
):
return None
# 处理命令和后续流程
await self._process_commands(message, chat)
except UserWarning as uw:
logger.info(str(uw))
except Exception as e:
logger.error(f"处理消息时出错: {e}")
logger.error(traceback.format_exc())
return None
# 保留旧的 process_message 方法用于向后兼容
async def process_message(self, envelope: MessageEnvelope) -> None:
"""
处理接收到的消息信封（向后兼容）
注意：此方法已被 MessageRuntime 路由取代。
如果直接调用此方法，它会委托给 runtime.handle_message()。
Args:
envelope: 消息信封（来自适配器）
"""
if self._runtime:
await self._runtime.handle_message(envelope)
else:
# 如果 runtime 未设置，使用旧的处理流程
await self._handle_normal_message(envelope)
async def _process_commands(self, message: DatabaseMessages, chat: "ChatStream") -> None:
"""处理命令和继续消息流程"""
try:
# 首先尝试 PlusCommand
is_plus_command, plus_cmd_result, plus_continue_process = await self._process_plus_commands(message, chat)
if is_plus_command and not plus_continue_process:
await MessageStorage.store_message(message, chat)
logger.info(f"PlusCommand处理完成，跳过后续消息处理: {plus_cmd_result}")
return
# 如果不是 PlusCommand，尝试传统 BaseCommand
if not is_plus_command:
is_command, cmd_result, continue_process = await self._process_base_commands(message, chat)
if is_command and not continue_process:
await MessageStorage.store_message(message, chat)
logger.info(f"命令处理完成，跳过后续消息处理: {cmd_result}")
return
# 触发消息事件
result = await event_manager.trigger_event(
EventType.ON_MESSAGE,
permission_group="SYSTEM",
message=message
)
if result and not result.all_continue_process():
raise UserWarning(
f"插件{result.get_summary().get('stopped_handlers', '')}于消息到达时取消了消息处理"
)
# 预处理消息
await self._preprocess_message(message, chat)
except UserWarning as uw:
logger.info(str(uw))
except Exception as e:
logger.error(f"处理命令时出错: {e}")
logger.error(traceback.format_exc())
async def _process_plus_commands(
self,
message: DatabaseMessages,
chat: "ChatStream"
) -> tuple[bool, Any, bool]:
"""处理 PlusCommand 系统"""
try:
text = message.processed_plain_text or ""
# 获取配置的命令前缀
prefixes = global_config.command.command_prefixes
# 检查是否以任何前缀开头
matched_prefix = None
for prefix in prefixes:
if text.startswith(prefix):
matched_prefix = prefix
break
if not matched_prefix:
return False, None, True
# 移除前缀
command_part = text[len(matched_prefix):].strip()
# 分离命令名和参数
parts = command_part.split(None, 1)
if not parts:
return False, None, True
command_word = parts[0].lower()
args_text = parts[1] if len(parts) > 1 else ""
# 查找匹配的 PlusCommand
plus_command_registry = component_registry.get_plus_command_registry()
matching_commands = []
for plus_command_name, plus_command_class in plus_command_registry.items():
plus_command_info = component_registry.get_registered_plus_command_info(plus_command_name)
if not plus_command_info:
continue
all_commands = [plus_command_name.lower()] + [
alias.lower() for alias in plus_command_info.command_aliases
]
if command_word in all_commands:
matching_commands.append((plus_command_class, plus_command_info, plus_command_name))
if not matching_commands:
return False, None, True
# 按优先级排序
if len(matching_commands) > 1:
matching_commands.sort(key=lambda x: x[1].priority, reverse=True)
plus_command_class, plus_command_info, plus_command_name = matching_commands[0]
# 检查是否被禁用
if (
chat
and chat.stream_id
and plus_command_name in global_announcement_manager.get_disabled_chat_commands(chat.stream_id)
):
logger.info("用户禁用的PlusCommand，跳过处理")
return False, None, True
message.is_command = True
# 获取插件配置
plugin_config = component_registry.get_plugin_config(plus_command_name)
# 创建实例
plus_command_instance = plus_command_class(message, plugin_config)
setattr(plus_command_instance, "chat_stream", chat)
try:
if not plus_command_instance.is_chat_type_allowed():
is_group = chat.group_info is not None
logger.info(
f"PlusCommand {plus_command_class.__name__} 不支持当前聊天类型: {'群聊' if is_group else '私聊'}"
)
return False, None, True
from src.plugin_system.base.command_args import CommandArgs
command_args = CommandArgs(args_text)
plus_command_instance.args = command_args
success, response, intercept_message = await plus_command_instance.execute(command_args)
if success:
logger.info(f"PlusCommand执行成功: {plus_command_class.__name__} (拦截: {intercept_message})")
else:
logger.warning(f"PlusCommand执行失败: {plus_command_class.__name__} - {response}")
return True, response, not intercept_message
except Exception as e:
logger.error(f"执行PlusCommand时出错: {plus_command_class.__name__} - {e}")
logger.error(traceback.format_exc())
try:
await plus_command_instance.send_text(f"命令执行出错: {e!s}")
except Exception:
pass
return True, str(e), False
except Exception as e:
logger.error(f"处理PlusCommand时出错: {e}")
return False, None, True
async def _process_base_commands(
self,
message: DatabaseMessages,
chat: "ChatStream"
) -> tuple[bool, Any, bool]:
"""处理传统 BaseCommand 系统"""
try:
text = message.processed_plain_text or ""
command_result = component_registry.find_command_by_text(text)
if command_result:
command_class, matched_groups, command_info = command_result
plugin_name = command_info.plugin_name
command_name = command_info.name
if (
chat
and chat.stream_id
and command_name in global_announcement_manager.get_disabled_chat_commands(chat.stream_id)
):
logger.info("用户禁用的命令，跳过处理")
return False, None, True
message.is_command = True
plugin_config = component_registry.get_plugin_config(plugin_name)
command_instance: BaseCommand = command_class(message, plugin_config)
command_instance.set_matched_groups(matched_groups)
setattr(command_instance, "chat_stream", chat)
try:
if not command_instance.is_chat_type_allowed():
is_group = chat.group_info is not None
logger.info(
f"命令 {command_class.__name__} 不支持当前聊天类型: {'群聊' if is_group else '私聊'}"
)
return False, None, True
success, response, intercept_message = await command_instance.execute()
if success:
logger.info(f"命令执行成功: {command_class.__name__} (拦截: {intercept_message})")
else:
logger.warning(f"命令执行失败: {command_class.__name__} - {response}")
return True, response, not intercept_message
except Exception as e:
logger.error(f"执行命令时出错: {command_class.__name__} - {e}")
logger.error(traceback.format_exc())
try:
await command_instance.send_text(f"命令执行出错: {e!s}")
except Exception:
pass
return True, str(e), False
return False, None, True
except Exception as e:
logger.error(f"处理命令时出错: {e}")
return False, None, True
async def _preprocess_message(self, message: DatabaseMessages, chat: "ChatStream") -> None:
"""预处理消息：存储、情绪更新等"""
try:
group_info = chat.group_info
# 检查是否需要处理消息
should_process_in_manager = True
if group_info and str(group_info.group_id) in global_config.message_receive.mute_group_list:
# 检查消息是否为图片或表情包
is_image_or_emoji = message.is_picid or message.is_emoji
if not message.is_mentioned and not is_image_or_emoji:
logger.debug(f"群组 {group_info.group_id} 在静默列表中，且消息不是@、回复或图片/表情包，跳过消息管理器处理")
logger.debug(
f"群组 {group_info.group_id} 在静默列表中，且消息不是@、回复或图片/表情包，跳过消息管理器处理"
)
should_process_in_manager = False
elif is_image_or_emoji:
logger.debug(f"群组 {group_info.group_id} 在静默列表中，但消息是图片/表情包，静默处理")
@@ -43,68 +570,81 @@ class MessageHandler:
await message_manager.add_message(chat.stream_id, message)
logger.debug(f"消息已添加到消息管理器: {chat.stream_id}")
except Exception as e:
logger.error(f"消息添加到消息管理器失败: {e}")
# 存储消息
try:
await MessageStorage.store_message(message, chat)
except Exception as e:
logger.error(f"存储消息到数据库失败: {e}")
traceback.print_exc()
# 情绪系统更新
try:
if global_config.mood.enable_mood:
interest_rate = message.interest_value or 0.0
logger.debug(f"开始更新情绪状态，兴趣度: {interest_rate:.2f}")
chat_mood = mood_manager.get_mood_by_chat_id(chat.stream_id)
await chat_mood.update_mood_by_message(message, interest_rate)
logger.debug("情绪状态更新完成")
except Exception as e:
logger.error(f"更新情绪状态失败: {e}")
traceback.print_exc()
# 存储消息到数据库，只进行一次写入
try:
await MessageStorage.store_message(message, chat)
except Exception as e:
logger.error(f"存储消息到数据库失败: {e}")
logger.error(f"预处理消息失败: {e}")
traceback.print_exc()
# 情绪系统更新 - 在消息存储后触发情绪更新
async def _handle_adapter_response(self, seg_data: dict | None) -> None:
"""处理适配器命令响应"""
try:
if global_config.mood.enable_mood:
# 获取兴趣度用于情绪更新
interest_rate = message.interest_value
if interest_rate is None:
interest_rate = 0.0
logger.debug(f"开始更新情绪状态，兴趣度: {interest_rate:.2f}")
from src.plugin_system.apis.send_api import put_adapter_response
if isinstance(seg_data, dict):
request_id = seg_data.get("request_id")
response_data = seg_data.get("response")
else:
request_id = None
response_data = None
if request_id and response_data:
logger.debug(f"收到适配器响应，request_id={request_id}")
put_adapter_response(request_id, response_data)
else:
logger.warning(
f"适配器响应消息格式不正确: request_id={request_id}, response_data={response_data}"
)
# 获取当前聊天的情绪对象并更新情绪状态
chat_mood = mood_manager.get_mood_by_chat_id(chat.stream_id)
await chat_mood.update_mood_by_message(message, interest_rate)
logger.debug("情绪状态更新完成")
except Exception as e:
logger.error(f"更新情绪状态失败: {e}")
traceback.print_exc()
logger.error(f"处理适配器响应时出错: {e}")
async def shutdown(self) -> None:
"""关闭消息处理器"""
self._shutting_down = True
logger.info("MessageHandler 正在关闭...")
async def handle_message(self, envelope: MessageEnvelope):
# 控制握手等消息可能缺少 message_info，这里直接跳过避免 KeyError
message_info = envelope.get("message_info")
if not isinstance(message_info, dict):
logger.debug(
"收到缺少 message_info 的消息，已跳过。可用字段: %s",
", ".join(envelope.keys()),
)
return
# 全局单例
_message_handler: MessageHandler | None = None
if message_info.get("group_info") is not None:
message_info["group_info"]["group_id"] = str( # type: ignore
message_info["group_info"]["group_id"] # type: ignore
)
if message_info.get("user_info") is not None:
message_info["user_info"]["user_id"] = str( # type: ignore
message_info["user_info"]["user_id"] # type: ignore
)
group_info = message_info.get("group_info")
user_info = message_info.get("user_info")
def get_message_handler() -> MessageHandler:
"""获取 MessageHandler 单例"""
global _message_handler
if _message_handler is None:
_message_handler = MessageHandler()
return _message_handler
chat_stream = await get_chat_manager().get_or_create_stream(
platform=envelope["platform"], # type: ignore
user_info=user_info, # type: ignore
group_info=group_info,
)
# 生成 DatabaseMessages
from src.chat.message_receive.message_processor import process_message_from_dict
message = await process_message_from_dict(
message_dict=envelope,
stream_id=chat_stream.stream_id,
platform=chat_stream.platform
)
async def shutdown_message_handler() -> None:
"""关闭 MessageHandler"""
global _message_handler
if _message_handler:
await _message_handler.shutdown()
_message_handler = None
__all__ = [
"MessageHandler",
"get_message_handler",
"shutdown_message_handler",
]

101
src/chat/message_receive/uni_message_sender.py
View File

@@ -1,13 +1,26 @@
"""
统一消息发送器
重构说明（2025-11）：
- 使用 CoreSinkManager 发送消息，而不是直接通过 WS 连接
- MessageServer 仅作为与旧适配器的兼容层
- 所有发送的消息都通过 CoreSinkManager.send_outgoing() 路由到适配器
"""
import asyncio
import traceback
import time
import uuid
from typing import Any, cast
from rich.traceback import install
from mofox_bus import MessageEnvelope
from src.chat.message_receive.message import MessageSending
from src.chat.message_receive.storage import MessageStorage
from src.chat.utils.utils import calculate_typing_time, truncate_message
from src.common.logger import get_logger
from src.common.message.api import get_global_api
install(extra_lines=3)
@@ -15,12 +28,30 @@ logger = get_logger("sender")
async def send_message(message: MessageSending, show_log=True) -> bool:
"""合并后的消息发送函数，包含WS发送和日志记录"""
"""
合并后的消息发送函数
重构后使用 CoreSinkManager 发送消息，而不是直接调用 MessageServer
Args:
message: 要发送的消息
show_log: 是否显示日志
Returns:
bool: 是否发送成功
"""
message_preview = truncate_message(message.processed_plain_text, max_length=120)
try:
# 直接调用API发送消息
await get_global_api().send_message(message)
# 将 MessageSending 转换为 MessageEnvelope
envelope = _message_sending_to_envelope(message)
# 通过 CoreSinkManager 发送
from src.common.core_sink_manager import get_core_sink_manager
manager = get_core_sink_manager()
await manager.send_outgoing(envelope)
if show_log:
logger.info(f"已将消息 '{message_preview}' 发往平台'{message.message_info.platform}'")
@@ -44,7 +75,67 @@ async def send_message(message: MessageSending, show_log=True) -> bool:
except Exception as e:
logger.error(f"发送消息 '{message_preview}' 发往平台'{message.message_info.platform}' 失败: {e!s}")
traceback.print_exc()
raise e # 重新抛出其他异常
raise e
def _message_sending_to_envelope(message: MessageSending) -> MessageEnvelope:
"""
将 MessageSending 转换为 MessageEnvelope
Args:
message: MessageSending 对象
Returns:
MessageEnvelope: 消息信封
"""
# 构建消息信息
message_info: dict[str, Any] = {
"message_id": message.message_info.message_id,
"time": message.message_info.time or time.time(),
"platform": message.message_info.platform,
"user_info": {
"user_id": message.message_info.user_info.user_id,
"user_nickname": message.message_info.user_info.user_nickname,
"platform": message.message_info.user_info.platform,
} if message.message_info.user_info else None,
}
# 添加群组信息（如果有）
if message.chat_stream and message.chat_stream.group_info:
message_info["group_info"] = {
"group_id": message.chat_stream.group_info.group_id,
"group_name": message.chat_stream.group_info.group_name,
"platform": message.chat_stream.group_info.group_platform,
}
# 构建消息段
message_segment: dict[str, Any]
if message.message_segment:
message_segment = {
"type": message.message_segment.type,
"data": message.message_segment.data,
}
else:
# 默认为文本消息
message_segment = {
"type": "text",
"data": message.processed_plain_text or "",
}
# 添加回复信息（如果有）
if message.reply_to:
message_segment["reply_to"] = message.reply_to
# 构建消息信封
envelope = cast(MessageEnvelope, {
"id": str(uuid.uuid4()),
"direction": "outgoing",
"platform": message.message_info.platform,
"message_info": message_info,
"message_segment": message_segment,
})
return envelope
class HeartFCSender: