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

Event Wrapper Class Detailed Explanation

The Event module provides a powerful Event wrapper class that simplifies event handling.

Adding Type Annotations to the event Parameter

The event parameter in event handlers is an Event wrapper class (a subclass of dict). It is strongly recommended to add type annotations to it:

from ErisPulse.Core.Event import Event

@message.on_private_message()
async def handler(event: Event):
    text = event.get_text()   # IDE auto-completes all convenient methods
    await event.reply(text)   # Spelling errors are detected during static checking

Without annotations, the IDE cannot recognize methods on Event (get_text() / reply() / wait_reply() / platform extension methods are not suggested), and you must rely on memory for spelling.

Note: The event in event handler callbacks is an Event wrapper class (annotated as Event); in module lifecycle methods on_load / on_unload, the event is a regular dict (annotated as dict), and these should not be confused.

Core Features

Core Field Methods

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: {platform}, Time: {time}")

Message Event Methods

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"Hello, {nickname}!")

Message Type Detection

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"Type: {'Private' if is_private else 'Group'}")

Reply Functionality

from ErisPulse.Core.Event import command

@command("ask")
async def ask_command(event: Event):
    await event.reply("Please enter your name:")
    reply = await event.wait_reply(timeout=30)
    if reply:
        name = reply.get_text()
        await event.reply(f"Hello, {name}!")

@command("price")
async def price_command(event: Event):
    await event.reply("Please enter the amount (e.g., 5 yuan):")
    # Reply must match the regex, otherwise continue waiting until timeout
    reply = await event.wait_reply(timeout=30, regex=r"\d+\s*元")
    if reply:
        await event.reply(f"Received amount: {reply.get_text()}")

Advanced Conversation Capabilities

Note

This section requires ErisPulse 2.8.0+.

# Session timeout reminder: Remind after 5 minutes of no reply, user reply cancels automatically
reminder = event.remind(300, "Are you still there? Reply 'exit' if you don't want to chat")
reminder.cancel()  # Can also be manually canceled

# Escalation: Guaranteed arrival at the deadline (not canceled by reply), e.g., notify the owner if long-term unhandled
event.escalate(1800, lambda e: notify_master("Ticket timeout"))

# Multi-path waiting: Wait for "agree" and "reject" simultaneously, first to arrive wins
which, reply = await event.select(
    event.expect(pattern="agree*", user="10001"),
    event.expect(pattern="reject*", user="10002"),
    timeout=60,
)
if which is None:
    await event.reply("No approval received within timeout")

# Session-level waiting: Any reply from the same group can match (group collaboration)
reply = await event.wait_reply(session=True, prompt="Any expert, please answer?")

# Session inbox: Recent 20 messages in the current session (including bot, AI context / anti-spam base)
messages = await event.history(20)

# Message transaction: Automatically recall messages sent within the transaction on exception
async with event.message_tx():
    await event.reply("Processing, please wait...")
    result = await do_something()
    await event.reply(f"Completed: {result}")

Command Information Retrieval

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"Command: {cmd_name}, Args: {cmd_args}")

Notice Event Methods

from ErisPulse.Core.Event import notice

@notice.on_friend_add()
async def friend_add_handler(event: Event):
    await event.reply("Welcome to add me as a friend!")

Method Quick Reference

Core Methods

Event Basic Information

Bot Information

Session Identifiers

Message Event Methods

Message Content

Sender Information

Group/Channel Information

Message Type Detection

Basic Detection

Notice Event Methods

Notice Operator

Notice Type Detection

Request Event Methods

Request Information

Request Type Detection

Reply Functionality

Basic Reply

Platform Capability Query

Forward Functionality

Note: Forward functionality must be implemented through the adapter's Send DSL; the Event wrapper class itself does not provide a direct forward method.

# Forward message to group
adapter = sdk.adapter.get(event.get_platform())
target_id = event.get_group_id()  # Or specify another group ID
await adapter.Send.To("group", target_id).Text(event.get_text())

Wait Reply Functionality

Interactive Methods

Interactive Method Examples

confirm() - Confirmation Dialog:

@command("delete", help="Delete data")
async def delete_handler(event: Event):
    if await event.confirm("Are you sure you want to delete all data?"):
        sdk.storage.delete("all_data")
        await event.reply("Data has been deleted")
    else:
        await event.reply("Cancelled")

confirm() - With Prompt Words:

# hint=True appends "（Yes/No）" at the end of the prompt
if await event.confirm("Continue?", hint=True):
    await event.reply("Continued")
# User sees: Continue? (Yes/No)

choose() - Selection Menu:

@command("color", help="Choose color")
async def color_handler(event: Event):
    choice = await event.choose("Please choose a color:", ["Red", "Green", "Blue"])
    if choice is not None:
        colors = ["Red", "Green", "Blue"]
        await event.reply(f"You chose: {colors[choice]}")

choose() - Option Formatting and Message Merging:

# inline format: options displayed in a single line
choice = await event.choose("Please choose:", ["A", "B", "C"], options_format="inline")
# Output: 1.A | 2.B | 3.C

# Custom format
choice = await event.choose("Please choose:", ["Cat", "Dog"],
    options_format=lambda opts: " / ".join(opts))
# Output: Cat / Dog

# options_format="auto" (default): Automatically selects built-in style based on method
# Markdown → unordered list
choice = await event.choose(
    "## Please choose", ["Cat", "Dog"],
    method="Markdown",  # auto recognizes as md list
)
# Output:
# ## Please choose
# - 1. Cat
# - 2. Dog

# Html → ordered list
choice = await event.choose(
    "<h2>Please choose</h2>", ["Cat", "Dog"],
    method="Html", merge_prompt=True,  # auto recognizes as html list
)
# Output:
# <h2>Please choose</h2>
# <ol><li>1. Cat</li><li>2. Dog</li></ol>

# Merge mode + placeholder
choice = await event.choose(
    "## Please choose\n{options}\nPlease reply with number",
    ["Cat", "Dog"],
    method="Markdown", merge_prompt=True,
)

# Custom placeholder
choice = await event.choose(
    "Choose: [choices]",
    ["Cat", "Dog"],
    placeholder="[choices]",
)

collect() - Form Collection:

@command("register", help="Register")
async def register_handler(event: Event):
    data = await event.collect([
        {"key": "name", "prompt": "Please enter your name:"},
        {"key": "age", "prompt": "Please enter your age:",
         "validator": lambda e: e.get_text().isdigit()},
    ])
    if data:
        await event.reply(f"Registration successful! {data['name']}, {data['age']} years old")

Non-Text Methods in 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("Look at this image:").image("http://example.com/img.jpg").build()
await event.reply_ob12(segments)

For complete Conversation multi-turn dialog usage, see Conversation Multi-turn Dialog.

Command Information

Command Basics

Raw Data

Platform Extension Methods

Adapters can register platform-specific methods for the Event wrapper class. The methods are only available on Event instances of the corresponding platform; attempting to access them on other platforms raises AttributeError.

Platform methods take precedence over built-in methods via Event.__getattribute__, allowing them to override built-in interactive methods like confirm, choose, collect, wait_reply, providing platform-specific implementations (e.g., buttons, cards). Built-in implementations are exported as _builtin_* functions for overriding.

# Email event - only email methods
event = Event({"platform": "email", "email_raw": {"subject": "Hello"}})
event.get_subject()      # ✅ Returns "Hello"
event.get_chat_type()    # ❌ AttributeError

# Telegram event - only Telegram methods
event = Event({"platform": "telegram", "telegram_raw": {"chat": {"type": "private"}}})
event.get_chat_type()    # ✅ Returns "private"
event.get_subject()      # ❌ AttributeError

# Built-in methods are always available
event.get_text()         # ✅ Any platform
event.reply("hi")        # ✅ Any platform

Query Registered Methods

from ErisPulse.Core.Event import get_platform_event_methods

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

hasattr and dir Support

hasattr(event, "get_subject")   # Returns True only when platform="email"
"get_subject" in dir(event)     # Same as above

Cross-platform Extension (Wildcard)

register_event_method and register_event_mixin support passing "*" as the platform name, registering methods available on Event instances of all platforms. This is suitable for AI chat, context management, and other features requiring cross-platform reuse.

from ErisPulse.Core.Event.wrapper import register_event_method

@register_event_method("*")
async def ai_chat(self, prompt: str):
    # self is an Event instance, can access event data and built-in methods
    await self.reply(f"AI: {prompt}")

After registration, event.ai_chat(...) can be called from any platform's event handler.

Method resolution priority (from highest to lowest): platform-specific methods → wildcard methods → built-in methods → dictionary key access.

For adapter developers registering extension methods, see Event System API - Cross-platform Extension Wildcard.