NcatBot 文档

外观

事件的订阅和发布

约 918 字大约 3 分钟

2025-03-06

事件

事件对象

事件是基本的可处理对象, 一个事件由 ncatbot.plugin_system.event.event.NcatBotEvent 类表示.

NcatBotEvent 类主要包含两个成员变量

  • data: Any 事件携带的数据。
  • type: str 事件类型

事件类型

事件在被发布时会携带上事件类型, 事件类型用于订阅和处理事件.

事件类型命名规范为 [插件名].[事件名].

官方事件 (群聊消息, 私聊消息, 请求消息, 通知消息, 启动事件, 心跳事件, 关闭事件) 的事件名封装如下:

  • ncatbot.utils.assets.literals.OFFICIAL_GROUP_MESSAGE_EVENT = "ncatbot.group_message_event"
  • ncatbot.utils.assets.literals.OFFICIAL_PRIVATE_MESSAGE_EVENT = "ncatbot.private_message_event"
  • ncatbot.utils.assets.literals.OFFICIAL_REQUEST_EVENT = "ncatbot.request_event"
  • ncatbot.utils.assets.literals.OFFICIAL_NOTICE_EVENT = "ncatbot.notice_event"
  • ncatbot.utils.assets.literals.OFFICIAL_STARTUP_EVENT = "ncatbot.startup_event"
  • ncatbot.utils.assets.literals.OFFICIAL_HEARTBEAT_EVENT = "ncatbot.heartbeat_event"
  • ncatbot.utils.assets.literals.OFFICIAL_SHUTDOWN_EVENT = "ncatbot.shutdown_event"

官方事件携带的数据一定是 BaseEventData 的子类。

插件也可以自行发布事件, 具体请继续阅读.

事件传播

事件沿**事件总线**传播, 处理事件时可以主动停止事件传播或者添加事件处理结果, 相关函数:

  • NcatBotEvent.stop_propagation()
  • NcatBotEvent.add_result(result)

订阅事件

插件可以通过订阅事件来处理事件.

订阅事件时，需要提供事件名和事件回调函数。

NcatBotPlugin 有一个 event_bus 成员变量, 该成员变量持有 EventBus 对象的引用, 用于管理事件的订阅和发布。你可以用 event_bus 来订阅事件。但是我们提供一个更方便的接口 register_handler 来订阅事件。

示例代码

from ncatbot.plugin_system.event import NcatBotEvent
from ncatbot.plugin_system import NcatBotPlugin

class MyPlugin(NcatBotPlugin):
    async def on_load(self):
        # 支持正则匹配,re:前缀
        self.hid1 = self.register_handler("re:test\.", self.handle_test) # 订阅 test 插件发布的所有事件
        self.hid2 = self.register_handler("exact.match", self.handle_exact) # 订阅 exact 插件发布的 match 事件

    async def handle_test(self, event: NcatBotEvent):
        print(f"正则匹配处理器: {event.data}")

    async def handle_exact(self, event: NcatBotEvent):
        print(f"精确匹配处理器: {event.data}")

    async def close_plugin(self):
        # 注销事件处理器
        self.event_bus.unregister_handler(self.hid1)
        self.event_bus.unregister_handler(self.hid2)
  • 理论上讲在任何时间都可以订阅事件，但一般在 on_load 函数中订阅。

事件回调函数

订阅事件时需要指定一个回调函数, 回调函数需要接受一个 NcatBotEvent 类型的参数。

本示例中, handle_test 和 handle_exact 都是事件回调函数，且他们都是 MyPlugin 类的成员函数。因此可以使用

其它有关方法

class NcatBotPlugin(...):
    def unregister_handler(self, handler_id: UUID) -> bool:
        """注销事件处理器。
        Args:
            handler_id: 要注销的事件处理器UUID
        Returns:
            是否成功注销
        """
        ...

    def unregister_all_handler(self) -> None:
        """注销本插件所有事件处理器。"""
        ...
        
    ...

参数绑定失败事件

当用户输入的命令参数与该命令实际要求的参数不匹配时（例如参数缺失、参数类型错误、参数数量不符等），NcatBot 会发布ncatbot.param_bind_failed 事件。插件可通过订阅该事件，自定义参数错误的提示逻辑，向用户反馈清晰的错误信息。

    async def on_load(self):
        self.event_bus.subscribe(
            event_type="ncatbot.param_bind_failed",
            handler=self.handle_param_error
        )
        LOG.info(f"{self.name} 已订阅ncatbot.param_bind_failed事件")

    @command_registry.command("hello", description="简单问候指令（需要姓名参数）")
    async def hello_cmd(self, event: BaseMessageEvent, name: str):
        await event.reply(f"Hello!, {name}!")

    async def handle_param_error(self, event: NcatBotEvent):
        await self.api.post_private_msg(
            user_id=event.data["event"].user_id,
            text=f"❌命令「{event.data['cmd']}」\n{event.data['msg']}"
        )

当用户发送/hello TestUser 123时，则会返回命令「hello」\n参数解析异常：list index out of range

发布事件

提示

一般用于跨插件的通讯

在 NcatBotPlugin 上下文中任意位置均可发布事件。

插件发布的事件会被所有订阅该事件的插件接收和处理。就像官方事件一样。

class MyPlugin(NcatBotPlugin):
    async def some_func(self):
        event = Event("MyPlugin.event", {"message": "hello"})
        await self.event_bus.publish_async(event)  # 异步发布不等待结果

版权归属：huan-yp

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

© 2025 NcatBot