事件注册与装饰器
约 1702 字大约 6 分钟
2026-03-19
事件类型体系、装饰器路由模式、优先级机制与通知/请求事件处理。
目录
事件类型体系
NcatBot 基于 OneBot v11 协议,将事件分为四大类:
| 大类 | 事件类型字符串 | 说明 |
|---|---|---|
| message | message.group / message.private | 群消息 / 私聊消息 |
| message_sent | message_sent.group / message_sent.private | Bot 自身发出的消息 |
| notice | notice.group_increase / notice.group_decrease / notice.group_recall / notice.notify 等 | 通知事件 |
| 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,自动从消息中提取参数并绑定到处理器函数的参数。实现含参数命令时,建议使用自动参数绑定:
# ✅ 推荐:自动参数绑定 — 框架自动提取并转换参数
@registrar.on_command("echo")
async def on_echo(self, event, content: str):
"""'echo 你好' → content='你好'"""
await event.reply(text=f"🔊 {content}")
# ❌ 不推荐:手动解析 — 代码更繁琐,且不支持引号处理、类型转换、缺失提示
@registrar.on_command("echo-manual")
async def on_echo_manual(self, event):
parts = event.raw_message.split(maxsplit=1)
content = parts[1] if len(parts) > 1 else ""
if not content:
await event.reply("用法: echo-manual <内容>")
return
await event.reply(f"🔊 {content}")更复杂的参数绑定示例:
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/02_command_binding/main.py 和 examples/common/02_config_and_data/main.py
支持的参数类型:
| 类型 | 说明 | 示例 |
|---|---|---|
str | 命令后的文本 | "回声 你好" → content="你好" |
int | 自动转为整数 | "禁言 @xxx 60" → duration=60 |
float | 自动转为浮点数 | "设置 3.14" → value=3.14 |
At | 消息中的 @段 | "踢 @xxx" → target=At(user_id=xxx) |
绑定机制细节:
- 消息预处理:消息中首个纯文本段(
PlainText)会被自动移到最前,解决 Reply 开头的消息(如引用回复)导致命令匹配不到的问题。 - 统一前缀匹配:只要消息首 token 匹配命令名即触发,无论 Handler 是否有额外参数。
- 引号支持:双引号或单引号包裹的部分视为单个 token(如
echo "hello world"→content="hello world")。 - 段跳过:参数逐一从消息段流中匹配,不匹配当前参数类型的段会被跳过(永久消耗,不参与后续参数匹配)。
- 必选参数缺失:若必选参数无法绑定,框架会 WARNING 并自动回复正确的参数格式提示,然后跳过该 Handler。
便捷装饰器一览
跨平台(registrar.*):
| 装饰器 | 事件类型 | 自动添加的 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(event_type) | 自定义 | 精确/前缀匹配 |
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 | 加群 |
Bilibili 平台(registrar.bilibili.*):
| 装饰器 | 路由键 | 事件类型 | 说明 |
|---|---|---|---|
on_live() | live | BiliLiveEvent | 所有直播间事件 |
on_danmu() | live.danmu_msg | DanmuMsgEvent | 弹幕消息 |
on_superchat() | live.super_chat_message | SuperChatEvent | 醒目留言 (SC) |
on_gift() | live.send_gift | GiftEvent | 礼物 |
on_guard_buy() | live.guard_buy | GuardBuyEvent | 大航海 |
on_interact() | live.interact_word_v2 | InteractEvent | 进入/关注/分享 |
on_like() | live.like_info_v3_click | LikeEvent | 点赞 |
on_live_start() | live.live | LiveNoticeEvent | 开播通知 |
on_live_end() | live.preparing | LiveNoticeEvent | 下播通知 |
on_comment() | comment | BiliCommentEvent | 评论事件 |
on_live_start / on_live_end 的事件类型均为 LiveNoticeEvent,开播时 event._data.room_info 携带完整直播间信息:
from ncatbot.core import registrar
from ncatbot.event.bilibili import LiveNoticeEvent
class LiveMonitor(NcatBotPlugin):
name = "live_monitor"
version = "1.0.0"
@registrar.bilibili.on_live_start()
async def on_start(self, event: LiveNoticeEvent):
data = event._data # LiveStatusEventData
if data.room_info:
title = data.room_info.room_info.title
anchor = data.room_info.anchor_info.name
area = data.room_info.room_info.area_name
await self.api.qq.post_group_msg(
GROUP_ID,
text=f"【开播】{anchor} 正在直播: {title}({area})",
)
else:
await self.api.qq.post_group_msg(GROUP_ID, text="直播间已开播!")
@registrar.bilibili.on_live_end()
async def on_end(self, event: LiveNoticeEvent):
await self.api.qq.post_group_msg(GROUP_ID, text="直播已结束。")注意:
room_info仅在开播(on_live_start)时填充;下播事件(on_live_end)中该字段为None。
所有装饰器均支持 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/01_event_registration/main.py 和 examples/common/02_config_and_data/main.py
通知与请求事件
from ncatbot.event.qq import (
GroupIncreaseEvent, GroupAdminEvent, GroupBanEvent,
GroupRecallEvent, GroupUploadEvent, PokeNotifyEvent,
FriendAddEvent, 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_group_admin()
async def on_admin_change(self, event: GroupAdminEvent):
"""管理员变动 → 记录"""
LOG.info("群 %s 管理员变动: %s (%s)", event.group_id, event.user_id, event.sub_type)
@registrar.qq.on_group_ban()
async def on_ban(self, event: GroupBanEvent):
"""禁言 → 记录"""
LOG.info("群 %s 禁言: %s 被 %s %s", event.group_id, event.user_id, event.operator_id, event.sub_type)
@registrar.qq.on_poke()
async def on_poke(self, event: PokeNotifyEvent):
"""戳一戳 → 回戳"""
if str(event.target_id) == str(event.self_id):
await self.api.qq.send_poke(event.group_id, event.user_id)
@registrar.qq.on_friend_add()
async def on_friend_add(self, event: FriendAddEvent):
"""好友已添加 → 记录"""
LOG.info("新好友: %s", event.user_id)
@registrar.qq.on_group_recall()
async def on_recall(self, event: GroupRecallEvent):
"""消息撤回"""
LOG.info("群 %s 消息 %s 被 %s 撤回", event.group_id, event.message_id, event.operator_id)
# 无专用快捷方法的事件用 registrar.qq.on() 注册
@registrar.qq.on("notice.group_upload")
async def on_upload(self, event: GroupUploadEvent):
"""群文件上传"""
LOG.info("群 %s 文件上传: %s", event.group_id, event.file.name)
@registrar.qq.on_friend_request()
async def on_friend_request(self, event: FriendRequestEvent):
"""好友请求 → 自动通过"""
await event.approve()下一步
版权所有
版权归属:MI
