事件类型体系、装饰器路由模式、优先级机制与通知/请求事件处理。

---

目录

• 事件类型体系
• 模式 A：装饰器注册（推荐）

---

事件类型体系

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="你好！👋")

取自 examples/qq/01_event_registration/main.py

参数绑定

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}")

取自 examples/common/01_hello_world/main.py

更复杂的参数绑定示例：

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 携带完整直播间信息，event._data.live_time 为开播时间戳，可用于区分同一场直播的重复 LIVE 事件和新开播事件：

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
        print(f"live_time={data.live_time}")
        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。live_time 缺失时默认为 0。

所有装饰器均支持 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()

取自 examples/qq/05_notice_and_request/main.py

---

下一步

• 事件高级用法 — 事件流、wait_event、事件实体
• Hook 基础 — 深入了解过滤器和中间件

版权所有

版权归属：MI

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