""" 命令与事件管理器模块 该模块定义了 `CommandManager` 类，它是整个机器人框架事件处理的核心。 它通过装饰器模式，为插件提供了注册消息指令、通知事件处理器和 请求事件处理器的能力。 主要职责： - 提供 `@matcher.command()` 装饰器来注册命令。 - 提供 `@matcher.on_notice()` 装饰器来注册通知处理器。 - 提供 `@matcher.on_request()` 装饰器来注册请求处理器。 - 负责解析收到的消息，匹配命令前缀并分发给对应的处理器。 - 统一处理所有类型的事件，并将其分发给所有已注册的处理器。 - 内置一个 `/help` 命令，用于展示所有已加载插件的帮助信息。 """ import inspect from typing import Any, Callable, Dict, List, Tuple from .config_loader import global_config # 从配置中获取命令前缀 comm_prefixes = global_config.bot.get("command", ("/",)) class CommandManager: """ 命令管理器，负责注册和分发所有类型的事件。 这是一个单例对象（`matcher`），在整个应用中共享。 """ def __init__(self, prefixes: Tuple[str, ...]): """ 初始化命令管理器。 Args: prefixes (Tuple[str, ...]): 一个包含所有合法命令前缀的元组。 """ # --- 初始化所有处理器列表 --- self.prefixes = prefixes self.commands: Dict[str, Callable] = {} self.message_handlers: List[Callable] = [] self.notice_handlers: List[Dict] = [] self.request_handlers: List[Dict] = [] self.plugins: Dict[str, Dict[str, Any]] = {} # --- 注册内置指令 --- self.commands["help"] = self._help_command self.plugins["core.help"] = { "name": "帮助", "description": "显示所有可用指令的帮助信息", "usage": "/help", } def on_message(self) -> Callable: """ 装饰器：用于注册一个通用的消息处理器。 被此装饰器注册的函数，会在每次收到消息时（在指令匹配前）被调用。 如果函数返回 True，则表示该消息已被“消费”，后续的指令匹配将不会进行。 Example: @matcher.on_message() async def code_input_handler(bot, event): if is_waiting_for_code(event.user_id): await process_code(event.raw_message) return True # 消费事件 """ def decorator(func: Callable) -> Callable: self.message_handlers.append(func) return func return decorator async def _help_command(self, bot, event): """ 内置的 `/help` 命令的实现。 该命令会遍历所有已加载插件的元数据，并生成一段格式化的帮助文本。 Args: bot: Bot 实例。 event: 消息事件对象。 """ help_text = "--- 可用指令列表 ---\n" for plugin_name, meta in self.plugins.items(): name = meta.get("name", "未命名插件") description = meta.get("description", "暂无描述") usage = meta.get("usage", "暂无用法说明") help_text += f"\n{name}:\n" help_text += f" 功能: {description}\n" help_text += f" 用法: {usage}\n" await bot.send(event, help_text.strip()) def command(self, name: str) -> Callable: """ 装饰器：用于注册一个消息指令处理器。 Example: @matcher.command("echo") async def handle_echo(bot, event, args): await bot.send(event, " ".join(args)) Args: name (str): 指令的名称（不包含命令前缀）。 Returns: Callable: 原函数，使其可以继续被调用。 """ def decorator(func: Callable) -> Callable: self.commands[name] = func return func return decorator def on_notice(self, notice_type: str = None) -> Callable: """ 装饰器：用于注册一个通知事件处理器。 如果 `notice_type` 未指定，则该处理器会接收所有类型的通知事件。 Args: notice_type (str, optional): 要处理的通知类型 (e.g., "group_increase")。 Defaults to None. Returns: Callable: 原函数。 """ def decorator(func: Callable) -> Callable: self.notice_handlers.append({"type": notice_type, "func": func}) return func return decorator def on_request(self, request_type: str = None) -> Callable: """ 装饰器：用于注册一个请求事件处理器。 如果 `request_type` 未指定，则该处理器会接收所有类型的请求事件。 Args: request_type (str, optional): 要处理的请求类型 (e.g., "friend", "group")。 Defaults to None. Returns: Callable: 原函数。 """ def decorator(func: Callable) -> Callable: self.request_handlers.append({"type": request_type, "func": func}) return func return decorator async def handle_event(self, bot, event): """ 统一的事件分发入口。 由 `WS` 客户端在接收到事件后调用。该方法会根据事件的 `post_type` 将其分发给对应的具体处理方法。 Args: bot: Bot 实例。 event: 已解析的事件对象。 """ # --- 全局过滤机器人自身消息 --- # 仅对消息事件生效 if event.post_type == 'message' and global_config.bot.get('ignore_self_message', False): if hasattr(event, 'user_id') and hasattr(event, 'self_id') and event.user_id == event.self_id: return post_type = event.post_type if post_type == 'message': await self.handle_message(bot, event) elif post_type == 'notice': await self.handle_notice(bot, event) elif post_type == 'request': await self.handle_request(bot, event) async def handle_message(self, bot, event): """ 处理消息事件，优先执行通用处理器，然后解析并分发指令。 """ # --- 1. 执行通用消息处理器 --- for handler in self.message_handlers: # 如果任何一个处理器返回 True，则中断后续处理 consumed = await self._run_handler(handler, bot, event) if consumed: return # --- 2. 检查并执行指令 --- if not event.raw_message: return raw_text = event.raw_message.strip() prefix_found = None for p in self.prefixes: if raw_text.startswith(p): prefix_found = p break if not prefix_found: return full_cmd = raw_text[len(prefix_found) :].split() if not full_cmd: return cmd_name = full_cmd[0] args = full_cmd[1:] if cmd_name in self.commands: func = self.commands[cmd_name] await self._run_handler(func, bot, event, args) async def handle_notice(self, bot, event): """ 分发通知事件给所有匹配的处理器。 Args: bot: Bot 实例。 event: 通知事件对象。 """ for handler in self.notice_handlers: if handler["type"] is None or handler["type"] == event.notice_type: await self._run_handler(handler["func"], bot, event) async def handle_request(self, bot, event): """ 分发请求事件给所有匹配的处理器。 Args: bot: Bot 实例。 event: 请求事件对象。 """ for handler in self.request_handlers: if handler["type"] is None or handler["type"] == event.request_type: await self._run_handler(handler["func"], bot, event) async def _run_handler(self, func: Callable, bot, event, args: List[str] = None): """ 智能执行事件处理器，并返回事件是否被消费。 该方法会检查目标处理器的函数签名，并根据签名动态地传入所需的参数 （如 `bot`, `event`, `args`），实现了依赖注入。 Args: func (Callable): 目标处理器函数。 bot: Bot 实例。 event: 事件对象。 args (List[str], optional): 指令参数列表（仅对消息事件有效）。 Returns: bool: 如果处理器函数返回 True，则返回 True，否则返回 False。 """ sig = inspect.signature(func) params = sig.parameters kwargs = {} if "bot" in params: kwargs["bot"] = bot if "event" in params: kwargs["event"] = event if "args" in params and args is not None: kwargs["args"] = args # 执行函数并获取返回值 result = await func(**kwargs) return result is True # --- 全局单例 --- # 确保前缀配置是元组格式 if isinstance(comm_prefixes, list): comm_prefixes = tuple(comm_prefixes) elif isinstance(comm_prefixes, str): comm_prefixes = (comm_prefixes,) # 实例化全局唯一的命令管理器 matcher = CommandManager(prefixes=comm_prefixes)