简体中文 English 繁體中文 日本語 Русский
本文为静态镜像，内容以交互版为准 在交互式文档中心打开 →

Event 包装类详解

Event 模块提供了功能强大的 Event 包装类，简化事件处理。

为 event 参数添加类型注解

事件处理器的 event 参数是 Event 包装类（dict 子类）。强烈建议为它添加类型注解：

from ErisPulse.Core.Event import Event

@message.on_private_message()
async def handler(event: Event):
    text = event.get_text()   # IDE 自动补全所有便捷方法
    await event.reply(text)   # 拼写错误在静态检查时即可发现

不加注解时 IDE 无法识别 Event 上的方法（get_text() / reply() / wait_reply() / 平台扩展方法均不提示），只能靠记忆拼写。

注意区分：事件处理器回调的 event 是 Event 包装类（注解为 Event）；模块生命周期方法 on_load / on_unload 的 event 是普通 dict（注解为 dict），二者不要混淆。

核心特性

核心字段方法

from ErisPulse.Core.Event import command

@command("info")
async def info_command(event: Event):
    event_id = event.get_id()
    platform = event.get_platform()
    time = event.get_time()
    print(f"ID: {event_id}, 平台: {platform}, 时间: {time}")

消息事件方法

from ErisPulse.Core.Event import message

@message.on_private_message()
async def private_handler(event: Event):
    text = event.get_text()
    user_id = event.get_user_id()
    nickname = event.get_user_nickname()
    await event.reply(f"你好，{nickname}！")

消息类型判断

from ErisPulse.Core.Event import message

@message.on_group_message()
async def group_handler(event: Event):
    is_private = event.is_private_message()
    is_group = event.is_group_message()
    is_at = event.is_at_message()
    await event.reply(f"类型: {'私聊' if is_private else '群聊'}")

回复功能

from ErisPulse.Core.Event import command

@command("ask")
async def ask_command(event: Event):
    await event.reply("请输入你的名字:")
    reply = await event.wait_reply(timeout=30)
    if reply:
        name = reply.get_text()
        await event.reply(f"你好，{name}！")

@command("price")
async def price_command(event: Event):
    await event.reply("请输入金额（如：5元）:")
    # 回复必须匹配正则，否则继续等待直到超时
    reply = await event.wait_reply(timeout=30, regex=r"\d+\s*元")
    if reply:
        await event.reply(f"收到金额：{reply.get_text()}")

交互会话进阶

Note

本节能力需要 ErisPulse **2.8.0+**。

# 会话定时提醒：5 分钟无回复则提醒，用户回复后自动取消
reminder = event.remind(300, "还在吗？不想聊了回复「退出」")
reminder.cancel()  # 也可手动取消

# 超时升级：到点必达（不被回复取消），如长时间未处理通知主人
event.escalate(1800, lambda e: notify_master("工单超时"))

# 多路等待：同时等"同意"与"拒绝"，先到先得
which, reply = await event.select(
    event.expect(pattern="同意*", user="10001"),
    event.expect(pattern="拒绝*", user="10002"),
    timeout=60,
)
if which is None:
    await event.reply("超时未收到审批")

# 会话级等待：同群任何人的回复均可命中（群协作）
reply = await event.wait_reply(session=True, prompt="哪位大佬帮忙答一下？")

# 会话收件箱：当前会话最近 20 条消息（含机器人，AI 上下文 / 防复读底座）
messages = await event.history(20)

# 消息事务：异常时自动撤回事务内已发送的消息
async with event.message_tx():
    await event.reply("处理中，请稍候...")
    result = await do_something()
    await event.reply(f"完成：{result}")

命令信息获取

from ErisPulse.Core.Event import command

@command("cmdinfo")
async def cmdinfo_command(event: Event):
    cmd_name = event.get_command_name()
    cmd_args = event.get_command_args()
    await event.reply(f"命令: {cmd_name}, 参数: {cmd_args}")

通知事件方法

from ErisPulse.Core.Event import notice

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

方法速查表

核心方法

事件基础信息

机器人信息

会话标识

消息事件方法

消息内容

发送者信息

群组/频道信息

@消息相关

消息类型判断

基础判断

通知事件方法

通知操作者

通知类型判断

请求事件方法

请求信息

请求类型判断

回复功能

基础回复

平台能力查询

转发功能

注意：转发功能需要通过适配器的 Send DSL 实现，Event 包装类本身不提供直接的转发方法。

# 转发消息到群组
adapter = sdk.adapter.get(event.get_platform())
target_id = event.get_group_id()  # 或指定其他群组ID
await adapter.Send.To("group", target_id).Text(event.get_text())

等待回复功能

交互方法

交互方法示例

confirm() - 确认对话：

@command("delete", help="删除数据")
async def delete_handler(event: Event):
    if await event.confirm("确定要删除所有数据吗？"):
        sdk.storage.delete("all_data")
        await event.reply("数据已删除")
    else:
        await event.reply("已取消")

confirm() - 带提示词：

# hint=True 会在提示末尾追加 "（是/否）"
if await event.confirm("确定继续？", hint=True):
    await event.reply("已继续")
# 用户看到：确定继续？（是/否）

choose() - 选择菜单：

@command("color", help="选择颜色")
async def color_handler(event: Event):
    choice = await event.choose("请选择颜色：", ["红色", "绿色", "蓝色"])
    if choice is not None:
        colors = ["红色", "绿色", "蓝色"]
        await event.reply(f"你选择了：{colors[choice]}")

choose() - 选项格式化与消息合并：

# inline 格式：选项显示在同一行
choice = await event.choose("请选择：", ["A", "B", "C"], options_format="inline")
# 输出：1.A | 2.B | 3.C

# 自定义格式
choice = await event.choose("请选择：", ["猫", "狗"],
    options_format=lambda opts: " / ".join(opts))
# 输出：猫 / 狗

# options_format="auto"（默认）：根据 method 自动选择内置样式
# Markdown → 无序列表
choice = await event.choose(
    "## 请选择", ["猫", "狗"],
    method="Markdown",  # auto 自动识别为 md 列表
)
# 输出：
# ## 请选择
# - 1. 猫
# - 2. 狗

# Html → 有序列表
choice = await event.choose(
    "<h2>请选择</h2>", ["猫", "狗"],
    method="Html", merge_prompt=True,  # auto 自动识别为 html 列表
)
# 输出：
# <h2>请选择</h2>
# <ol><li>1. 猫</li><li>2. 狗</li></ol>

# 合并模式 + 占位符
choice = await event.choose(
    "## 请选择\n{options}\n请回复编号",
    ["猫", "狗"],
    method="Markdown", merge_prompt=True,
)

# 自定义占位符
choice = await event.choose(
    "请选择: [choices]",
    ["猫", "狗"],
    placeholder="[choices]",
)

collect() - 表单收集：

@command("register", help="注册")
async def register_handler(event: Event):
    data = await event.collect([
        {"key": "name", "prompt": "请输入姓名："},
        {"key": "age", "prompt": "请输入年龄：",
         "validator": lambda e: e.get_text().isdigit()},
    ])
    if data:
        await event.reply(f"注册成功！{data['name']}，{data['age']}岁")

非 Text 方法的 reply：

await event.reply("http://example.com/img.jpg", method="Image")
await event.reply("http://example.com/audio.mp3", method="Voice")

from ErisPulse.Core.Event import MessageBuilder
segments = MessageBuilder.text("看这张图：").image("http://example.com/img.jpg").build()
await event.reply_ob12(segments)

完整的 Conversation 多轮对话用法请参考 Conversation 多轮对话。

命令信息

命令基础

原始数据

平台扩展方法

适配器可以为 Event 包装类注册平台专有方法。方法仅在对应平台的 Event 实例上可用，其他平台访问时抛出 AttributeError。

平台方法通过 Event.__getattribute__ 优先于内置方法生效，因此可以覆写 confirm、choose、collect、wait_reply 等内置交互方法，提供平台特色实现（如按钮、卡片等）。内置实现作为 _builtin_* 函数导出供覆写方调用。

# 邮件事件 - 只有邮件方法
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

# 内置方法始终可用
event.get_text()         # ✅ 任何平台
event.reply("hi")        # ✅ 任何平台

查询已注册方法

from ErisPulse.Core.Event import get_platform_event_methods

methods = get_platform_event_methods("email")
# ["get_subject", "get_from", ...]

hasattr 和 dir 支持

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

跨平台扩展（通配符）

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

注册后，任何平台的事件处理器都能调用 event.ai_chat(...)。

方法解析优先级（从高到低）：平台特定方法 → 通配符方法 → 内置方法 → 字典键访问。

适配器开发者注册扩展方法的方式请参阅 事件系统 API - 跨平台扩展。

相关文档