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

Подробное описание обёртки событий

Модуль 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); event в методах жизненного цикла модуля on_load / on_unload является обычным 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 сообщений текущей сессии (включая сообщения бота, контекст ИИ / база для предотвращения повторения)
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("Добро пожаловать, добавьте меня в друзья!")

Справочник методов

Основные методы

Базовая информация о событии

Информация о боте

Идентификаторы сессии

Методы событий сообщений

Текст сообщения

Информация об отправителе

Информация о группе/канале

Упоминания

Типы сообщений

Базовые проверки

Методы уведомлений

Информация об операторе

Типы уведомлений

Методы запросов

Информация о запросе

Типы запросов

Ответы

Основные ответы

Проверка поддержки платформой

Пересылка сообщений

Важно: Функция пересылки реализуется через 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" (по умолчанию): автоматически выбирается встроенный стиль в зависимости от метода
# 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']}岁")

reply с не-Text методами:

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 системы событий - Расширение для всех платформ (шаблон).

Связанные документы