NcatBot 文档

外观

事件的订阅和发布

约 694 字大约 2 分钟

2025-03-06

事件

事件对象

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

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

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

事件类型

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

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

四大基本事件 (群聊消息, 私聊消息, 请求消息, 通知消息) 的事件名封装如下:

  • ncatbot.utils.literals.OFFICIAL_GROUP_MESSAGE_EVENT = "ncatbot.group_message_event"
  • ncatbot.utils.literals.OFFICIAL_PRIVATE_MESSAGE_EVENT = "ncatbot.private_message_event"
  • ncatbot.utils.literals.OFFICIAL_REQUEST_EVENT = "ncatbot.request_event"
  • ncatbot.utils.literals.OFFICIAL_NOTICE_EVENT = "ncatbot.notice_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}")

事件回调函数

订阅事件时需要指定一个回调函数, 回调函数需要接受一个 Event 类型的参数, 并且必须定义为异步函数.

在事件回调函数中可以对事件进行处理, 也可以调用 Event 的方法停止事件传播或者添加事件处理结果.

发布事件

在 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