291 lines
9.5 KiB
Python
291 lines
9.5 KiB
Python
"""
|
||
命令与事件管理器模块
|
||
|
||
该模块定义了 `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)
|
||
|