事件系统 API

本文档详细介绍了 ErisPulse 事件系统的 API。

事件系统将平台事件按类型分发到五类处理器：

flowchart LR
    A["平台事件<br/>（OneBot12 标准）"] --> B{"事件类型"}
    B --> C["command<br/>命令处理器"]
    B --> D["message<br/>消息处理器"]
    B --> E["notice<br/>通知处理器"]
    B --> F["request<br/>请求处理器"]
    B --> G["meta<br/>元事件处理器"]
    C & D & E & F & G --> H["Event 包装类<br/>reply / get_text / done 等"]

Command 命令模块

注册命令

from ErisPulse.Core.Event import command

# 基本命令
@command("hello", help="发送问候")
async def hello_handler(event):
    await event.reply("你好！")

# 带别名的命令
@command(["help", "h"], aliases=["帮助"], help="显示帮助")
async def help_handler(event):
    pass

# 带权限的命令
def is_admin(event):
    return event.get("user_id") in admin_ids

@command("admin", permission=is_admin, help="管理员命令")
async def admin_handler(event):
    pass

# 隐藏命令
@command("secret", hidden=True, help="秘密命令")
async def secret_handler(event):
    pass

# 命令组
@command("admin.reload", group="admin", help="重新加载模块")
async def reload_handler(event):
    pass

# 子命令（空格分隔的多 token 命令名）
# 匹配采用最长前缀：/admin add x 优先命中 admin add（args 为 ["x"]）；
# 子命令未声明 permission 时继承父链上最近声明权限的祖先命令
@command("admin add", help="添加管理员")
async def admin_add_handler(event):
    pass

命名冲突规则：命令名优先于别名。注册与既有命令重名的别名时该别名不生效并输出 告警；注册与既有别名重名的命令时命令名优先生效、死别名自动移除——两类冲突均有 WARNING 日志，不会静默劫持。

命令信息

所有命令查询 API 均支持可选的会话上下文：传 event=（Event 或 dict）或 显式 platform= / bot_id= / session_id=（与 event 叠加时显式参数优先）， 即按作用域模块维度过滤当前会话不可用模块的命令（详见 advanced/scope.md）； 全部为可选关键字参数，不传时保持原有全量行为。

# 获取命令帮助
help_text = command.help()

# 会话感知帮助：只列出当前会话可用的命令
help_text = command.help(event=event)

# 获取特定命令（返回合并覆盖后的生效参数；会话不可用时返回 None）
cmd_info = command.get_command("admin")
cmd_info = command.get_command("admin", event=event)

# 获取所有命令（会话感知时过滤不可用模块的命令）
all_commands = command.get_commands()
all_commands = command.get_commands(event=event)

# 获取命令组中的所有命令（支持会话感知过滤）
admin_commands = command.get_group_commands("admin")
admin_commands = command.get_group_commands("admin", event=event)

# 获取所有可见命令
visible_commands = command.get_visible_commands()

# 会话感知的可见命令（event 或显式关键字任一即可）
visible_commands = command.get_visible_commands(event=event)
visible_commands = command.get_visible_commands(
    platform=event.get("platform"),
    bot_id=event.get_self_account_id(),
    session_id=event.get_session_id(),
)

等待回复

# 等待用户回复
@command("ask", help="询问用户信息")
async def ask_command(event):
    reply = await command.wait_reply(
        event,
        prompt="请输入你的名字:",  # 已在上面发送
        timeout=30.0
    )
    
    if reply:
        name = reply.get_text()
        await event.reply(f"你好，{name}！")

# 带验证的等待回复
def validate_age(event_data):
    try:
        age = int(event_data.get_text())
        return 0 <= age <= 150
    except ValueError:
        return False

@command("age", help="询问用户年龄")
async def age_command(event):
    await event.reply("请输入你的年龄:")
    
    reply = await command.wait_reply(
        event,
        timeout=60,
        validator=validate_age
    )
    
    if reply:
        age = int(reply.get_text())
        await event.reply(f"你的年龄是 {age} 岁")

# 带回调的等待回复
async def handle_confirmation(reply_event):
    text = reply_event.get_text().lower()
    if text in ["是", "yes", "y"]:
        await event.reply("操作已确认！")
    else:
        await event.reply("操作已取消。")

@command("confirm", help="确认操作")
async def confirm_command(event):
    await command.wait_reply(
        event,
        prompt="请输入'是'或'否':",
        callback=handle_confirmation
    )

Message 消息模块

消息事件

from ErisPulse.Core.Event import message

# 监听所有消息
@message.on_message()
async def message_handler(event):
    sdk.logger.info(f"收到消息: {event.get_text()}")

# 监听私聊消息
@message.on_private_message()
async def private_handler(event):
    user_id = event.get_user_id()
    sdk.logger.info(f"私聊来自: {user_id}")

# 监听群聊消息
@message.on_group_message()
async def group_handler(event):
    group_id = event.get_group_id()
    sdk.logger.info(f"群聊来自: {group_id}")

# 监听@消息
@message.on_at_message()
async def at_handler(event):
    mentions = event.get_mentions()
    sdk.logger.info(f"被@的用户: {mentions}")

条件监听

# 使用优先级控制执行顺序
@message.on_message(priority=10)  # 数值越大优先级越高
async def high_priority_handler(event):
    pass

# 在处理器内部实现条件过滤
@message.on_message()
async def filtered_handler(event):
    if "关键词" not in event.get_text():
        return
    # 处理包含关键词的消息
    pass

Notice 通知模块

通知事件

from ErisPulse.Core.Event import notice

# 好友添加
@notice.on_friend_add()
async def friend_add_handler(event):
    user_id = event.get_user_id()
    await event.reply("欢迎添加我为好友！")

# 好友删除
@notice.on_friend_remove()
async def friend_remove_handler(event):
    user_id = event.get_user_id()
    sdk.logger.info(f"好友删除: {user_id}")

# 群成员增加
@notice.on_group_increase()
async def member_increase_handler(event):
    user_id = event.get_user_id()
    await event.reply(f"欢迎新成员！")

# 群成员减少
@notice.on_group_decrease()
async def member_decrease_handler(event):
    user_id = event.get_user_id()
    sdk.logger.info(f"群成员离开: {user_id}")

Request 请求模块

请求事件

from ErisPulse.Core.Event import request

# 好友请求
@request.on_friend_request()
async def friend_request_handler(event):
    user_id = event.get_user_id()
    comment = event.get_comment()
    sdk.logger.info(f"好友请求: {user_id}, 备注: {comment}")

# 群邀请请求
@request.on_group_request()
async def group_request_handler(event):
    group_id = event.get_group_id()
    user_id = event.get_user_id()
    sdk.logger.info(f"群邀请: {group_id}, 来自: {user_id}")

Meta 元事件模块

元事件

from ErisPulse.Core.Event import meta

# 连接事件
@meta.on_connect()
async def connect_handler(event):
    platform = event.get_platform()
    sdk.logger.info(f"平台 {platform} 连接成功")

# 断开连接事件
@meta.on_disconnect()
async def disconnect_handler(event):
    platform = event.get_platform()
    sdk.logger.info(f"平台 {platform} 断开连接")

# 心跳事件
@meta.on_heartbeat()
async def heartbeat_handler(event):
    sdk.logger.debug("收到心跳")

Bot 状态查询

当适配器发送 meta 事件后，框架会自动追踪 Bot 状态。查询 API 和生命周期事件监听请参考 适配器系统 API - Bot 状态管理。

Event 包装类

Event 模块的事件处理器接收一个 Event 包装类实例，它继承自 dict 并提供了便捷方法。

核心方法

# 获取事件信息
event_id = event.get_id()
event_time = event.get_time()
event_type = event.get_type()
detail_type = event.get_detail_type()
platform = event.get_platform()

# 获取机器人信息
self_platform = event.get_self_platform()
self_user_id = event.get_self_user_id()
self_info = event.get_self_info()

会话标识

# 统一目标 ID：群聊返回 group_id，私聊返回 user_id，以此类推
target_id = event.get_target_id()

# 会话唯一标识，格式: {platform}:{detail_type}:{target_id}
session_id = event.get_session_id()
# 示例: "telegram:private:12345"、"qq:group:67890"

get_target_id() 按以下顺序返回首个非空值：group_id → channel_id → guild_id → thread_id → user_id。适用于上下文管理、状态存储等需要统一标识会话的场景。

消息方法

# 获取消息内容
message_segments = event.get_message()
alt_message = event.get_alt_message()
text = event.get_text()

# 获取发送者信息
user_id = event.get_user_id()
nickname = event.get_user_nickname()
sender = event.get_sender()

# 获取群组信息
group_id = event.get_group_id()

# 判断消息类型
is_msg = event.is_message()
is_private = event.is_private_message()
is_group = event.is_group_message()

# @消息相关
is_at = event.is_at_message()
has_mention = event.has_mention()
mentions = event.get_mentions()

命令信息

# 获取命令信息
cmd_name = event.get_command_name()
cmd_args = event.get_command_args()
cmd_raw = event.get_command_raw()

# 判断是否为命令
is_cmd = event.is_command()

回复功能

# 基本回复
await event.reply("这是一条消息")

# 指定发送方法
await event.reply("http://example.com/image.jpg", method="Image")

# 带 @用户 和回复消息
await event.reply("你好", at_users=["user1"], reply_to="msg_id")

# @全体成员
await event.reply("公告", at_all=True)

# 使用平台专有修饰方法（via 参数）
await event.reply("看板内容", method="Board",
                  via=[("Expire", 3600), ("ForMember", "114514")])

# 获取发送链，自由追加修饰方法和发送方法（适合连续多个修饰 / 动作型方法）
await event.send_chain().Expire(3600).Board("看板内容")
await event.send_chain().DismissBoard()

# 使用 OneBot12 消息段回复
from ErisPulse.Core.Event import MessageBuilder
msg = MessageBuilder().text("Hello").image("url").build()
await event.reply_ob12(msg)

# 等待回复
reply = await event.wait_reply(timeout=30)

平台能力查询

# 检查当前平台是否支持某种发送方法
if event.supports("Image"):
    await event.reply(url, method="Image")

# 列出当前平台所有可用发送方法
methods = event.available_methods()
# ["Text", "Image", "Voice", ...]

回复方法

reply() 方法支持通过 method 参数指定发送类型，以及两个便捷的布尔参数：

# 简单文本回复
await event.reply("你好")

# 回复并@发送者
await event.reply("你好", at_sender=True)

# 回复并引用当前消息
await event.reply("收到", quote=True)

# 组合使用
await event.reply("收到", at_sender=True, quote=True)

# 发送图片（使用 method 参数）
if event.supports("Image"):
    await event.reply("http://example.com/img.jpg", method="Image")
else:
    await event.reply("[图片] http://example.com/img.jpg")

参数说明：

交互方法

# confirm — 确认对话（返回 True/False/None）
if await event.confirm("确定要执行此操作吗？"):
    await event.reply("已确认")

# 使用非 Text 方式发送确认提示
if await event.confirm("http://example.com/image.jpg", method="Image"):
    await event.reply("已确认图片提示")

# choose — 选择菜单（返回选项索引或 None）
choice = await event.choose("请选择颜色：", ["红色", "绿色", "蓝色"])

# options_format="auto"（默认）根据 method 自动选择样式：
# Markdown→无序列表（- 1.选项），Html→有序列表（<ol>），其他→纯文本列表
# 文本类方法（Markdown/Html 等）默认合并选项到末尾
# merge_prompt=True 可强制任意 method 合并；placeholder 可自定义占位符
choice = await event.choose(
    "## 请选择\n{options}", ["A", "B"],
    method="Markdown", merge_prompt=True,
)

# collect — 表单收集（返回 {key: value} 字典或 None）
data = await event.collect([
    {"key": "name", "prompt": "请输入姓名："},
    {"key": "age", "prompt": "请输入年龄：",
     "validator": lambda e: e.get_text().isdigit()},
    {"key": "avatar", "prompt": "请发送头像：", "method": "Image"},
])

# wait_for — 等待满足条件的任意事件
evt = await event.wait_for(event_type="notice", condition=lambda e: ..., timeout=120)

# conversation — 多轮对话上下文
conv = event.conversation(timeout=60)
await conv.say("欢迎！")

完整的交互方法参数说明和更多示例请参考 Event 包装类详解 和 Conversation 多轮对话。

工具方法

# 转换为字典（过滤以 _ 开头的内部键）
event_dict = event.to_dict()

# 获取原始数据
raw = event.get_raw()
raw_type = event.get_raw_type()

链路控制

event.done(claim=, stop=) 统一控制「认领」与「阻断」两个正交语义：

• 认领（claim）：标记事件已被处理（_processed），命令分发器据此跳过去重
• 阻断（stop）：阻止向低优先级处理器传播（_propagation_stopped）

# 认领 + 阻断（默认）
event.done()

# 仅认领，不阻断（低优先级观察者仍能看到）
event.done(stop=False)

# 仅阻断，不认领（如防火墙 / 限流）
event.done(claim=False)

# mark_processed 是主方法，done 是其别名
event.mark_processed()             # 等价 event.done()
event.mark_processed(stop=False)   # 等价 event.done(stop=False)

# 查询状态
event.is_processed()  # 是否已认领
event.is_stopped()    # 是否已阻断传播

平台扩展方法

适配器可以为 Event 注册平台专有方法，仅在对应平台的实例上可用。

用户：使用平台扩展方法

当适配器注册了平台专有方法后，你可以在事件处理器中直接调用。各平台的方法不同，请参阅对应的 平台文档。

from ErisPulse.Core.Event import message

@message.on_message()
async def handle_message(event):
    platform = event.get_platform()

    # 根据平台调用专有方法
    if platform == "email":
        subject = event.get_subject()           # 邮件专有
        attachments = event.get_attachments()   # 邮件专有

查询平台已注册方法

from ErisPulse.Core.Event import get_platform_event_methods

# 查看某平台注册了哪些方法
methods = get_platform_event_methods("email")
# ["get_subject", "get_from", "get_attachments", ...]

# 动态判断并调用
for method_name in get_platform_event_methods(event.get_platform()):
    method = getattr(event, method_name)
    print(f"{method_name}: {method()}")

平台方法隔离

不同平台注册的方法互不干扰：

# 邮件事件 - 只有邮件方法
event = Event({"platform": "email", "email_raw": {"subject": "Hello"}})
event.get_subject()      # ✅ "Hello"
event.get_chat_type()    # ❌ AttributeError

# Telegram 事件 - 只有 Telegram 方法
event = Event({"platform": "telegram", "telegram_raw": {"chat": {"type": "private"}}})
event.get_chat_type()    # ✅ "private"
event.get_subject()      # ❌ AttributeError

hasattr / dir 支持

hasattr(event, "get_subject")   # 仅当 platform="email" 时返回 True
"get_subject" in dir(event)     # 同上

适配器：注册平台扩展方法

适配器可以通过装饰器为 Event 注册平台专有方法，方法的第一个参数为 self（Event 实例），可以自由访问事件数据。

单个方法注册

from ErisPulse.Core.Event import register_event_method

@register_event_method("email")
def get_subject(self):
    """获取邮件主题"""
    return self.get("email_raw", {}).get("subject", "")

@register_event_method("email")
def get_from(self):
    """获取发件人"""
    return self.get("email_raw", {}).get("from", {})

批量注册（Mixin 类）

当方法较多时，推荐使用 Mixin 类批量注册：

from ErisPulse.Core.Event import register_event_mixin

class EmailEventMixin:
    def get_subject(self):
        return self.get("email_raw", {}).get("subject", "")

    def get_from(self):
        return self.get("email_raw", {}).get("from", {})

    def get_attachments(self):
        return self.get("email_raw", {}).get("attachments", [])

# 一次性注册所有方法
register_event_mixin("email", EmailEventMixin)

返回值规范

建议：非数据返回的方法返回 asyncio.Task，这样用户可以自行决定是否 await，即使不 await 操作也会执行完成。

@register_event_method("email")
def forward_email(self, to_address: str):
    """转发邮件 — 返回 Task，用户可自行决定是否 await"""
    import asyncio
    return asyncio.create_task(
        self._do_forward(to_address)
    )

# 用户可以 await 等待结果
await event.forward_email("[email protected]")

# 也可以不 await，操作在后台执行
event.forward_email("[email protected]")

注销方法

from ErisPulse.Core.Event import unregister_event_method, unregister_platform_event_methods

# 注销单个方法
unregister_event_method("email", "get_subject")

# 注销某平台全部方法（适配器 shutdown 时调用）
unregister_platform_event_methods("email")

覆写内置方法

register_event_mixin / register_event_method 支持覆写 Event 内置方法（如 confirm、choose、collect、wait_reply、reply 等）。注册的平台方法通过 Event.__getattribute__ 优先于内置方法生效，因此适配器可以提供平台特色的交互实现。

内置实现作为 _builtin_* 函数导出，覆写方可以调用它们作为回退：

from ErisPulse.Core.Event import register_event_mixin, _builtin_choose

class YunhuEventMixin:
    async def choose(self, prompt, options, timeout=60, method="Text"):
        # 云湖平台使用按钮组件
        buttons = [[{"text": opt} for opt in options]]
        await self.reply(prompt)
        # ...等待按钮回调或文本回复...
        # 回退到内置逻辑
        return await _builtin_choose(self, None, options, timeout, "Text")

register_event_mixin("yunhu", YunhuEventMixin)

跨平台扩展（通配符）

register_event_method 和 register_event_mixin 支持传 "*" 作为平台名，注册的方法在所有平台的 Event 实例上都可用。适合 AI 对话、上下文管理等需要跨平台复用的功能模块。

注册跨平台方法

from ErisPulse.Core.Event.wrapper import register_event_method

@register_event_method("*")
async def ai_chat(self, prompt: str):
    """self 为 Event 实例，可自由访问事件数据和内置方法"""
    await self.reply(f"AI: {prompt}")

注册后，所有平台的事件处理器都能调用：

from ErisPulse.Core.Event import message

@message.on_message()
async def handler(event):
    await event.ai_chat(event.get_text())

方法解析优先级

通过属性访问 Event 方法时，解析顺序为：

1. 平台特定方法（当前平台的覆写）
2. 通配符方法（"*" 注册的跨平台方法）
3. 内置方法（reply、confirm 等）
4. 字典键访问

因此通配符方法可以覆写内置方法（如 reply），但会被同名的平台特定方法进一步覆写。

优先级系统

事件处理器支持优先级，数值越大优先级越高：

# 高优先级处理器先执行
@message.on_message(priority=10)
async def high_priority_handler(event):
    pass

# 低优先级处理器后执行
@message.on_message(priority=0)
async def low_priority_handler(event):
    pass

相关文档

• 核心模块 API - 核心模块 API
• 适配器系统 API - Adapter 管理 API
• 模块开发指南 - 开发自定义模块