事件注册与装饰器
约 978 字大约 3 分钟
2026-03-19
事件类型体系、装饰器路由模式、优先级机制与通知/请求事件处理。
目录
事件类型体系
NcatBot 基于 OneBot v11 协议，将事件分为四大类：
| 大类 | 事件类型字符串 | 说明 |
|---|---|---|
| message | message.group / message.private | 群消息 / 私聊消息 |
| notice | notice.group_increase / notice.group_decrease / notice.group_recall / notice.poke 等 | 通知事件 |
| request | request.friend / request.group | 好友请求 / 群请求 |
| meta | meta_event.lifecycle / meta_event.heartbeat | 元事件 |
事件路由支持前缀匹配——注册 "message" 可以匹配所有消息类型（message.group 和 message.private）。
模式 A：装饰器注册（推荐）
最常用的方式——使用 @registrar 装饰器将方法注册为事件处理器，框架自动路由匹配的事件：
命令装饰器
from ncatbot.core import registrar
from ncatbot.event.qq import GroupMessageEvent, PrivateMessageEvent
class MyPlugin(NcatBotPlugin):
name = "my_plugin"
version = "1.0.0"
@registrar.on_group_command("hello", ignore_case=True)
async def on_hello(self, event: GroupMessageEvent):
"""群里发 'hello' 时触发"""
await self.api.qq.post_group_msg(event.group_id, text="Hello, World! 👋")
@registrar.on_group_command("hi", ignore_case=True)
async def on_hi(self, event: GroupMessageEvent):
"""event.reply() 自动引用 + @发送者"""
await event.reply(text="你好呀！🎉")
@registrar.on_private_command("hello", ignore_case=True)
async def on_private_hello(self, event: PrivateMessageEvent):
"""私聊命令"""
await event.reply(text="你好！👋")参数绑定
on_group_command / on_private_command 内置了 CommandHook，自动从消息中提取参数并绑定到处理器函数的参数：
from ncatbot.types import At
@registrar.on_group_command("禁言")
async def on_ban(
self, event: GroupMessageEvent, target: At = None, duration: int = 60
):
"""'禁言 @xxx 60' → target=At(user_id=xxx), duration=60"""
if target is None:
await event.reply("请 @一个用户，例如: 禁言 @xxx 60")
return
await self.api.qq.manage.set_group_ban(event.group_id, target.user_id, duration)
await event.reply(f"已禁言 {duration} 秒")
@registrar.on_group_command("设置前缀")
async def on_set_prefix(self, event: GroupMessageEvent, new_prefix: str):
"""'设置前缀 !' → new_prefix='!'"""
self.set_config("prefix", new_prefix)
await event.reply(f"命令前缀已更新为: {new_prefix}")取自 examples/qq/04_bot_api/main.py 和 examples/common/02_config_and_data/main.py
支持的参数类型：
| 类型 | 说明 | 示例 |
|---|---|---|
str | 命令后的文本 | "回声 你好" → content="你好" |
int | 自动转为整数 | "禁言 @xxx 60" → duration=60 |
At | 消息中的 @段 | "踢 @xxx" → target=At(user_id=xxx) |
便捷装饰器一览
| 装饰器 | 事件类型 | 自动添加的 Hook |
|---|---|---|
on_group_command(*names) | message | MessageTypeFilter("group") + CommandHook |
on_private_command(*names) | message | MessageTypeFilter("private") + CommandHook |
on_command(*names) | message | CommandHook（群/私聊均可） |
on_group_message() | message | MessageTypeFilter("group") |
on_private_message() | message | MessageTypeFilter("private") |
on_message() | message | （无额外过滤） |
on_message_sent() | message_sent | （无额外过滤） |
on_notice() | notice | （无额外过滤） |
on_request() | request | （无额外过滤） |
on_meta() | meta_event | （无额外过滤） |
on_group_increase() | notice | NoticeTypeFilter("group_increase") |
on_group_decrease() | notice | NoticeTypeFilter("group_decrease") |
on_poke() | notice | NoticeTypeFilter("notify") + SubTypeFilter("poke") |
on_friend_request() | request | RequestTypeFilter("friend") |
on_group_request() | request | RequestTypeFilter("group") |
on(event_type) | 自定义 | 精确/前缀匹配 |
所有装饰器均支持 platform 参数，用于限定只接收特定平台的事件：
# 仅处理 QQ 平台的群消息
@registrar.on_group_message(platform="qq")
async def qq_only(self, event: GroupMessageEvent):
await event.reply(text="QQ 平台的消息")
# 处理所有平台的消息（默认）
@registrar.on_message()
async def all_platforms(self, event):
print(f"来自 {event.platform} 的消息")详见 多平台开发指南
Handler 优先级
通过 priority 参数控制同事件多个 Handler 的执行优先级——数值越大，优先级越高：
@registrar.on_group_message(priority=100)
async def count_message(self, event: GroupMessageEvent):
"""高优先级：每条群消息都计数"""
self.data["total_messages"] += 1
@registrar.on_group_command("ping", priority=10)
async def on_ping(self, event: GroupMessageEvent):
"""标准优先级"""
await event.reply("pong 🏓")
@registrar.on_group_command("状态", priority=0)
async def on_status(self, event: GroupMessageEvent):
"""低优先级"""
await event.reply("运行中 ✅")取自 examples/qq/02_event_handling/main.py 和 examples/common/02_config_and_data/main.py
通知与请求事件
from ncatbot.event.qq import (
GroupIncreaseEvent, NoticeEvent,
FriendRequestEvent, GroupRequestEvent,
)
@registrar.qq.on_group_increase()
async def on_member_join(self, event: GroupIncreaseEvent):
"""新成员入群 → 发送欢迎消息"""
msg = MessageArray()
msg.add_at(event.user_id)
msg.add_text(" 欢迎加入本群！📜")
await self.api.qq.post_group_array_msg(event.group_id, msg)
@registrar.qq.on_poke()
async def on_poke(self, event: NoticeEvent):
"""戳一戳 → 回戳"""
target_id = getattr(event.data, "target_id", None)
if str(target_id) == str(event.self_id):
await self.api.qq.send_poke(event.group_id, event.user_id)
@registrar.qq.on_friend_request()
async def on_friend_request(self, event: FriendRequestEvent):
"""好友请求 → 自动通过"""
await event.approve()
@registrar.qq.on_group_recall()
async def on_recall(self, event: NoticeEvent):
"""消息撤回"""
operator_id = getattr(event.data, "operator_id", None)
await self.api.qq.post_group_msg(
event.group_id,
text=f"有人撤回了一条消息 👀 (操作者: {operator_id})"
)下一步
版权所有
版权归属：huan-yp