Registrar 与 Hook/Filter API 参考
约 1463 字大约 5 分钟
2026-03-19
HandlerDispatcher、Registrar 装饰器、Hook 抽象与所有内置 Hook/Filter 的完整签名。
HandlerDispatcher
核心分发引擎,管理处理器注册与事件分发。
class HandlerDispatcher:
def __init__(
self,
api: Optional[IAPIClient] = None,
service_manager: Optional[ServiceManager] = None,
platform_apis: Optional[Dict[str, IAPIClient]] = None,
global_hooks: Optional[List[Hook]] = None,
): ...
def start(self, event_dispatcher: AsyncEventDispatcher) -> None: ...
async def stop(self) -> None: ...
def register_handler(
self,
event_type: str,
func: Callable,
priority: int = 0,
plugin_name: str = "",
**metadata: Any,
) -> Optional[HandlerEntry]: ...
def unregister_handler(self, entry: HandlerEntry) -> bool: ...
def revoke_plugin(self, plugin_name: str) -> int: ...
def set_platform_api(self, platform: str, api: IAPIClient) -> None: ...构造参数
| 参数 | 类型 | 说明 |
|---|---|---|
api | Optional[IAPIClient] | 默认 API 实例(回退用) |
service_manager | Optional[ServiceManager] | 服务管理器(注入到 HookContext) |
platform_apis | Optional[Dict[str, IAPIClient]] | 平台 → API 映射(按事件平台选择 API) |
global_hooks | Optional[List[Hook]] | 全局 BEFORE_CALL Hooks,在每个 handler 的 handler 级 hooks 之前执行 |
register_handler()
| 参数 | 类型 | 说明 |
|---|---|---|
event_type | str | 事件类型字符串(如 "message.group") |
func | Callable | 异步处理函数(必须是 async 函数) |
priority | int | 优先级(越大越先执行),默认 0 |
plugin_name | str | 所属插件名(热重载时用于批量清理) |
**metadata | Any | 附加元信息(如 plugin_instance) |
返回 Optional[HandlerEntry]:成功注册返回条目,同步函数会被跳过并返回 None。
unregister_handler()
移除单个 handler 条目。参数为 register_handler() 返回的 HandlerEntry 实例。
set_platform_api()
注册平台特定 API,用于事件实体注入。在分发事件时,根据 event.data.platform 选择对应的 API 实例。
Registrar 装饰器
通过装饰器注册处理器,在插件中自动绑定到 HandlerDispatcher。Registrar 使用 ContextVar 实现插件隔离——装饰器标记 handler 元信息后,flush_pending() 将其批量注册到 HandlerDispatcher。
通用参数
所有跨平台装饰器支持:
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
priority | int | 0 | 执行优先级 |
platform | Optional[str] | None | 平台过滤(None = 所有平台,"qq" = 仅 QQ) |
跨平台事件装饰器
以下装饰器在 Registrar(registrar)上直接调用,接收所有平台的事件:
| 装饰器 | 监听事件类型 | 说明 |
|---|---|---|
on(event_type, priority=, platform=) | 任意 | 通用注册 |
on_message() | message.* | 所有消息 |
on_group_message() | message.group | 群消息 |
on_private_message() | message.private | 私聊消息 |
on_message_sent() | message_sent.* | Bot 自身发出的消息 |
on_notice() | notice.* | 所有通知 |
on_request() | request.* | 所有请求 |
on_meta() | meta_event.* | 元事件 |
跨平台命令装饰器
| 装饰器 | 等价于 | 说明 |
|---|---|---|
on_command(*names, ignore_case=False) | on_message() + CommandHook | 命令匹配 |
on_group_command(*names, ignore_case=False) | on_group_message() + CommandHook | 群命令 |
on_private_command(*names, ignore_case=False) | on_private_message() + CommandHook | 私聊命令 |
fork()
创建带不同默认 Hook 的新 Registrar 实例:
def fork(
self,
extra_hooks: Optional[List[Hook]] = None,
remove_hooks: Optional[List[Hook]] = None,
) -> Registrar: ...平台子注册器
QQ / Bilibili / GitHub 专属的便捷装饰器通过平台子注册器提供,自动附带平台过滤:
@registrar.qq.on_poke() # QQ 戳一戳
@registrar.bilibili.on_danmu() # Bilibili 弹幕
@registrar.github.on_push() # GitHub PushQQ 子注册器 (registrar.qq)
| 装饰器 | 监听事件类型 | 说明 |
|---|---|---|
on_group_increase() | notice.group_increase | 入群 |
on_group_decrease() | notice.group_decrease | 退群 |
on_group_recall() | notice.group_recall | 群撤回 |
on_group_admin() | notice.group_admin | 管理员变动 |
on_group_ban() | notice.group_ban | 禁言 |
on_friend_add() | notice.friend_add | 好友已添加 |
on_group_msg_emoji_like() | notice.group_msg_emoji_like | 群消息表情回应 |
on_poke() | notice.notify + SubTypeFilter(poke) | 戳一戳 |
on_friend_request() | request.friend | 加好友请求 |
on_group_request() | request.group | 加群请求 |
on_meta() | meta_event | 元事件 |
on_message_sent() | message_sent | 消息发送 |
Bilibili 子注册器 (registrar.bilibili)
| 装饰器 | 监听事件类型 | 说明 |
|---|---|---|
on_live() | live | 所有直播间事件 |
on_danmu() | live.danmu_msg | 弹幕 |
on_superchat() | live.super_chat_message | 醒目留言 |
on_gift() | live.send_gift | 礼物 |
on_guard_buy() | live.guard_buy | 大航海 |
on_interact() | live.interact_word_v2 | 进入/关注/分享 |
on_like() | live.like_info_v3_click | 点赞 |
on_comment() | comment | 评论 |
GitHub 子注册器 (registrar.github)
| 装饰器 | 监听事件类型 | 说明 |
|---|---|---|
on_push() | push | Push 事件 |
on_issue() | issue | Issue 事件 |
on_pr() | pull_request | PR 事件 |
on_star() | star | Star 事件 |
on_fork() | fork | Fork 事件 |
on_release() | release | Release 事件 |
on_comment() | comment | 评论事件 |
通用平台装饰器
所有平台子注册器还继承了基础装饰器:on()、on_message()、on_group_message()、on_private_message()、on_command()、on_group_command()、on_private_command()、on_notice()、on_request()。
Hook 抽象
Hook 基类
class Hook(ABC):
@abstractmethod
async def execute(self, ctx: HookContext) -> HookAction: ...HookStage — 执行阶段
class HookStage(Enum):
BEFORE_CALL = "before_call" # handler 执行前
AFTER_CALL = "after_call" # handler 执行后
ON_ERROR = "on_error" # handler 抛出异常时HookAction — 返回动作
class HookAction(Enum):
CONTINUE = "continue" # 继续执行
SKIP = "skip" # 仅 BEFORE_CALL: 跳过当前 handlerHookContext — 执行上下文
@dataclass
class HookContext:
event: Event # 当前事件
event_type: str # 事件类型字符串
handler_entry: HandlerEntry # handler 注册信息
api: BotAPIClient # API 实例
services: Optional[ServiceManager] = None # 服务管理器
kwargs: Dict[str, Any] = field(default_factory=dict) # 共享参数字典
result: Any = None # handler 返回值(AFTER_CALL 阶段)
error: Optional[Exception] = None # 异常对象(ON_ERROR 阶段)内置 Hook / Filter
消息过滤
| 类 | 参数 | 说明 |
|---|---|---|
StartsWithHook(prefix) | prefix: str | 消息以指定前缀开头 |
KeywordHook(*keywords) | *keywords: str | 消息包含任一关键词 |
RegexHook(pattern) | pattern: str | 消息匹配正则表达式 |
CommandHook(*names, ...) | *names: str, ignore_case: bool = False, priority: int = 95 | 命令匹配 + 参数绑定(预处理 → shlex 分词 → 统一前缀匹配 → binding stream → 缺失回复用法) |
类型过滤
| 类 | 参数 | 说明 |
|---|---|---|
MessageTypeFilter(type) | type: str ("group" / "private") | 按消息类型过滤 |
PostTypeFilter(type) | type: str | 按 post_type 过滤 |
SubTypeFilter(type) | type: str | 按 sub_type 过滤 |
NoticeTypeFilter(type) | type: str | 按 notice_type 过滤 |
RequestTypeFilter(type) | type: str | 按 request_type 过滤 |
PlatformFilter(platform) | platform: str ("qq" / "telegram" 等) | 按平台过滤 |
特殊过滤
| 类 | 参数 | 说明 |
|---|---|---|
SelfFilter() | — | 过滤 Bot 自身消息 |
CommandGroupHook — 命令组路由
CommandGroupHook 在 CommandHook 基础上增加了子命令注册与自动路由功能:
hook = CommandGroupHook("admin", "/admin", "a")
@hook.subcommand("ban", "禁言")
async def admin_ban(event, user_id: int, minutes: int = 60):
...
@hook.subcommand("kick")
async def admin_kick(event, user_id: int):
...
@registrar.on_group_message()
@hook
async def handle_admin(self, event, subcommand: str = ""):
...| 构造参数 | 类型 | 说明 |
|---|---|---|
*names | str | 命令名列表(支持别名) |
ignore_case | bool | 大小写不敏感,默认 False |
priority | int | Hook 优先级,默认 95 |
subcommand(*names) 装饰器注册子命令处理器,匹配 command subcommand [args...] 格式。
CommandGroup(向后兼容)
CommandGroup 是子命令/子命令组容器,提供 command(*names) 和 subgroup(group) 方法。建议使用 CommandGroupHook 替代。
延伸阅读
- 事件注册 — 装饰器用法入门
- Hooks 进阶 — Hook 链实战
- Predicate DSL — 条件组合 API
- Dispatcher 内部机制 — 分发引擎详解
版权所有
版权归属:MI
