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 Push

QQ 子注册器 (`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: 跳过当前 handler

HookContext — 执行上下文

@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 替代。

版权所有

版权归属：MI

许可证：署名 4.0 国际 (CC-BY-4.0)

通用

QQ

Bilibili

GitHub

AI

Misc