NcatBot 文档

外观

事件的订阅和发布

约 994 字大约 3 分钟

2025-03-06

事件

事件对象

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

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

  • 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"

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

事件传播

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

  • Event.stop_propagation()
  • Event.add_result(result)

事件总线

咕咕咕.

订阅事件

可以使用兼容回调函数注册器来订阅事件，具体参考了解 NcatBot 插件.

兼容回调函数注册器的本质仍然是订阅事件. 使用兼容注册器注册后, 会在插件加载时为注册的函数订阅对应的事件.

下面介绍通过事件名订阅的方法。

示例代码

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

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

    async def handle_exact(self, event: Event):
        print(f"精确匹配处理器: {event.data}")
  • BasePlugin 提供了 register_handler 方法用于订阅事件, 该方法接受两个参数:
    • type: str 事件类型, 支持全名称和正则表达式，请参考上方的示例。
    • func: Callable[[Event], Any] 事件回调函数, 事件回调函数接受一个 Event 类型的参数，且一般是插件类的成员函数。
  • 理论上讲在任何时间都可以订阅事件，但一般在 on_load 函数中订阅。

事件回调函数

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

在事件回调函数中可以对事件进行处理，一般有下面的一些操作：

  • 调用 self.api 发送一些什么东西。（插件实例会有一个 BotAPI 成员，用于调用 API）
  • 调用 event.stop_propagation() 停止事件传播。
  • 调用 event.add_result(result) 添加事件处理结果。

Event 原型

class Event:
    def __init__(self, type: str, data: Any, source: EventSource = None):
        """初始化事件"""
        self.type = type # 事件类型, 定义见上方
        self.data = data # 事件数据
        self.source = source  # 事件源, 描述事件触发的群聊和用户

    def stop_propagation(self):
        """停止事件的传播:当调用此方法后，后续的处理器将不会被执行"""

    def add_result(self, result: Any):
        """添加事件处理结果"""

    def __repr__(self):
        return f'Event(type="{self.type}",data={self.data})'

Event.data 的规定

对于官方事件类型, 对应的 Event.data 保证和 回调函数参数一致.

对于自定义事件类型, 插件作者应该自己定义规范.

发布事件

在 BasePlugin 上下文中任意位置均可发布事件.

事件分为同步和异步, 同步事件会优先处理并迅速返回结果, 异步事件将会挂到异步事件循环中处理, 如果可能请一定使用异步事件, 同步事件未经测试不稳定

事件的处理结果(如果有), 应该使用 Event.add_result(result) 在回调函数中添加.

请务必严格按照 [插件名].[事件名] 的格式填写事件类型.

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

版权归属：huan-yp

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

© 2025 NcatBot