Введение в обработку событий

В этом руководстве рассказывается о том, как обрабатывать различные типы событий в ErisPulse.

Обзор типов событий

ErisPulse поддерживает следующие типы событий:

Обработка событий сообщений

Примечание: Рекомендуется использовать аннотацию типа Event в обработчиках событий, чтобы получить поддержку автодополнения и проверки типов в IDE.

from ErisPulse.Core.Event import Event  # Импорт типа события для аннотации

Отслеживание всех сообщений

from ErisPulse.Core.Event import message, Event

@message.on_message()
async def message_handler(event: Event):
    text = event.get_text()
    user_id = event.get_user_id()
    sdk.logger.info(f"Получено сообщение от {user_id}: {text}")

Отслеживание личных сообщений

@message.on_private_message()
async def private_handler(event: Event):
    user_id = event.get_user_id()
    await event.reply(f"Привет, {user_id}! Это личное сообщение.")

Отслеживание групповых сообщений

@message.on_group_message()
async def group_handler(event: Event):
    group_id = event.get_group_id()
    user_id = event.get_user_id()
    sdk.logger.info(f"В группе {group_id} пользователь {user_id} отправил сообщение")

Отслеживание сообщений с упоминанием

@message.on_at_message()
async def at_handler(event: Event):
    # Получить список упомянутых пользователей
    mentions = event.get_mentions()
    await event.reply(f"Вы упомянули этих пользователей: {mentions}")

Обработка с помощью подстановочных знаков и регулярных выражений

Четыре декоратора сообщения (on_message / on_private_message / on_group_message / on_at_message) поддерживают pattern (подстановочные знаки glob) и regex (регулярные выражения), не соответствующие сообщения не запускают обработчик:

# Подстановочные знаки glob: * любая строка, ? один символ, [seq] набор символов
@message.on_message(pattern="签到*")
async def signin_handler(event: Event):
    await event.reply("签到成功")

# Регулярное выражение: совпадение суммы
@message.on_message(regex=r"\d+\s*元")
async def price_handler(event: Event):
    await event.reply(f"Получена сумма: {event.get_text()}")

# pattern и regex одновременно → оба должны соответствовать
@message.on_message(pattern="*元", regex=r"\d+\s*元")
async def combined_handler(event: Event):
    pass

wait_reply также поддерживает эти два параметра (см. Функция ожидания ответа).

Обработка событий команд

Основные команды

from ErisPulse.Core.Event import command

@command("help", help="Отображает справочную информацию")
async def help_handler(event):
    help_text = """
Доступные команды:
/help - Показать справку
/ping - Тест подключения
/info - Просмотр информации
    """
    await event.reply(help_text)

Псевдонимы команд

@command(["help", "h"], aliases=["помощь"], help="Отображает справочную информацию")
async def help_handler(event):
    await event.reply("Справочная информация...")

Пользователь может вызывать команду любым из следующих способов:

• /help
• /h
• /помощь

Аргументы команд

@command("echo", help="Вернуть сообщение")
async def echo_handler(event):
    # Получить аргументы команды
    args = event.get_command_args()
    
    if not args:
        await event.reply("Введите сообщение для повтора")
    else:
        await event.reply(f"Вы сказали: {' '.join(args)}")

Аргументы сохраняют оригинальный регистр ввода пользователя (даже если настройки нечувствительны к регистру, нормализация имени команды не влияет на содержимое аргументов).

Декларативные аргументы и опции (args= / options=)

Ручная обработка аргументов требует ручного преобразования типов и сообщений об ошибках. При декларации args= / options= фреймворк автоматически анализирует аргументы команды и внедряет их по именам в обработчик после прохождения проверки прав доступа; при неверном вводе пользователем автоматически возвращаются локализованные сообщения и инструкции (без выброса исключений и аварийного завершения), а команда /help <команда> автоматически отображает инструкции:

@command(
    "roll",
    args="<count:int> [sides:int=6]",
    options={"verbose": "-v/--verbose", "label": "--label"},
    help="Бросить кубик",
)
async def roll_handler(event, count: int, sides: int = 6, verbose: bool = False, label: str = ""):
    total = sum(random.randint(1, sides) for _ in range(count))
    await event.reply(f"Бросил {count} раз {sides}-гранного кубика, общее количество очков: {total}")

Синтаксис позиционных аргументов args=: <count:int> обязательный, [sides:int=6] необязательный (с значением по умолчанию). Поддерживаемые типы:

options= опции объявляются в виде словаря: ключ — имя параметра обработчика, значение — формат флагов (множество алиасов через /). Параметры с аннотацией bool — это булевы флаги (появление означает True); остальные (по умолчанию как str) — опции со значениями, поддерживают --label hello и --label=hello для значений, тип соответствует аннотации обработчика. Опции сначала распознаются и удаляются, оставшиеся токены обрабатываются по args= (остаток rest покрывает оставшийся текст после удаления опций).

Особенности поведения:

• Проверка прав доступа выполняется до анализа аргументов — пользователи без прав не запускают анализ
• При неудачном анализе (неправильный тип / отсутствие параметров / слишком много параметров / неизвестная опция) автоматически возвращается локализованное сообщение об ошибке + инструкция, команда всё равно считается выполненной
• Имена декларированных параметров должны присутствовать в сигнатуре обработчика, иначе при регистрации возникает ValueError
• При отсутствии args= / options= поведение команды не меняется (обратная совместимость)

Управление командами (cooldown= / rate_limit= / usage_limit= / deprecated=)

Ручная реализация таймеров, ограничения скорости, лимитов использования и устаревших команд может быть заменена декларациями, все четыре параметра могут быть комбинированы.

Таймер — синтаксис длительности аналогичен типу duration в args= (например, "30s"、"1h30m"、"1d"):

@command("daily", cooldown="1d", cooldown_key="user", cooldown_reply="Вы уже отметились сегодня")
async def daily_handler(event):
    await event.reply("Отметка успешно выполнена!")

Ограничение скорости — декларация скользящего окна в формате "количество/окно" (например, "5/minute"、"10/s"、"3/2m"):

@command("search", rate_limit="5/minute", rate_limit_key="user")
async def search_handler(event):
    await event.reply("Результаты поиска")

Лимит использования — максимальное количество вызовов в периоде (например, "100/day"、"10/hour"、"500/30d"), при превышении лимита выполнение отклоняется:

@command("translate", usage_limit="100/day", usage_limit_key="user",
         usage_limit_reply="Лимит переводов на сегодня исчерпан")
async def translate_handler(event):
    ...

В отличие от таймера и ограничения скорости, лимит использования сохраняется в хранилище (ключ erispulse.usage.<key>, не теряется при перезапуске): при недоступности хранилища автоматически переходит в чисто инкрементный счётчик в памяти с предупреждением (в этом случае лимит сбрасывается при перезапуске). Счётчик сбрасывается при смене периода лимита, при выгрузке модуля — очищается. Устаревание — при вызове автоматически возвращается сообщение устаревания, deprecated_reject=True отклоняет выполнение:

@command("oldcmd", deprecated="Пожалуйста, используйте /newcmd", deprecated_reject=True)
async def old_handler(event): ...

Гранулярность ключей (cooldown_key= / rate_limit_key=): "user" (по умолчанию, общие для одного пользователя), "session" (общие для одной сессии, например, в одной группе), "global" (общие для всех пользователей и сессий).

Особенности поведения:

• При срабатывании таймера / ограничения скорости / лимита использования по умолчанию молча отбрасываются (в соответствии с областью действия); при декларации cooldown_reply= / rate_limit_reply= / usage_limit_reply= при срабатывании возвращается соответствующее сообщение
• Команда считается выполненной при срабатывании — управление не пропускает команду низкому приоритету
• Проверка управления происходит после всех проверок прав доступа и анализа аргументов, перед фактическим выполнением: пользователи без прав не запускают анализ, ошибки аргументов не уменьшают лимит
• При одновременной декларации таймера и ограничения скорости сначала проверяется таймер (срабатывание таймера не занимает окно ограничения скорости); лимит использования проверяется после таймера и ограничения скорости
• deprecated= по умолчанию возвращает сообщение и продолжает выполнение; deprecated_reject=True отклоняет выполнение (钩子 command.executed записывает success=False, error="deprecated")
• В списке /help и справке по команде автоматически отображаются метки и сообщения устаревания
• Состояние таймера и ограничения скорости хранится в памяти процесса; при выгрузке модуля автоматически очищается; распределённое хранение и сохранение при перезапуске не поддерживаются (кроме счётчика лимита использования — сохраняется в хранилище, см. выше)
• Проверка деклараций происходит на этапе регистрации (fail-fast): синтаксис не соответствует, значение гранулярности ключа не в белом списке, reply не сопровождается основной декларацией — возникает ValueError

Ограничение скорости обработчиков (throttle=) и дебаунсинг (debounce=)

Декларация обработчиков сообщений для защиты от спама — события с одинаковым ключом обрабатываются максимум один раз за интервал, остальные молча отбрасываются:

from ErisPulse import sdk

@sdk.message.on_message(throttle="2s", throttle_key="user")
async def handler(event): ...

Дебаунсинг и ограничение скорости дополняют друг друга: в окне выполняется только последнее событие, предыдущие задачи автоматически отменяются (подходит для сценариев, где нужно обрабатывать только после окончания ввода, например, автозаполнение поиска):

@sdk.message.on_message(debounce="2s", debounce_key="user")
async def search(event): ...

on_message / on_private_message / on_group_message / on_at_message поддерживаются; throttle_key= / debounce_key= используют ту же гранулярность ключей, что и управление командами (user / session / global), синтаксис длительности аналогичен duration. Ограничение скорости и pattern= / regex= и другие условия работают в совокупности (все должны выполняться, чтобы сработало); в интервале отбрасываются события, записываются в лог TRACE; декларации проверяются на этапе регистрации; throttle= и debounce= семантически взаимоисключающие (одновременная декларация вызывает ValueError).

Дебаунсинг не прерывает бизнес-логику: только задачи, которые ещё не пересекли окно ожидания, отменяются; уже пересекшие окно и выполняющиеся обработчики не отменяются новыми событиями (избегает остановки на любом await-точке и частичных побочных эффектов).

Внедрение зависимостей (Depends)

Общие зависимости (сессия базы данных, чтение конфигурации и т.д.) можно вынести в функции зависимостей, обработчики объявляют параметры с Depends(функция_зависимости) в качестве значения по умолчанию, фреймворк автоматически вызывает функцию зависимости с объектом контекста и внедряет результат по именам перед вызовом:

from ErisPulse.Core import Depends

async def get_session(event):
    return await sdk.module.call("DB", "get_session")

@command("admin")
async def admin_handler(event, db=Depends(get_session)):
    ...

По умолчанию включено кэширование на уровне запроса: в рамках одной обработки события одна и та же функция зависимости вызывается только один раз, результаты всех точек внедрения разделяются (например, get_db в рамках одного события создает только одну сессию базы данных); между запросами кэш не используется. Можно отключить кэш для отдельной зависимости с Depends(get_db, use_cache=False).

Охватывает все точки внедрения фреймворка — обработчики команд, обработчики событий (message.on_message() и т.д.), жизненный цикл (sdk.lifecycle.on), обработчики маршрутов SSE. Первый параметр функции зависимости — объект контекста точки внедрения (в случае события — Event, в случае жизненного цикла — data, в случае маршрута — HttpRequest / SseEmitter); можно объявлять как синхронные, так и асинхронные функции зависимости.

Синтаксический сахар для объявления сервисов других модулей:

@command("query")
async def query_handler(event, session=Depends.module("DB", "get_session")):
    ...

Depends.module(имя_модуля, имя_метода, *постоянные_аргументы) эквивалентно вызову sdk.module.call(...) внутри функции зависимости. Инициализация модуля (__init__) не входит в область охвата — при инициализации нет объекта контекста; для маршрутов HTTP, использующих FastAPI, используйте оригинальный fastapi.Depends фреймворка FastAPI.

Особенности поведения:

• Декларации проверяются на этапе регистрации (fail-fast): при невозможности вызова зависимости или при повторном имени с args= / options= возникает ValueError
• Исключения, возникающие в функции зависимости, обрабатываются так же, как и исключения в обработчике (ошибка команды автоматически возвращается пользователю)
• Обработчики без декларации Depends не имеют накладных расходов (на этапе маршрутизации нет отражения)
• Для маршрутов HTTP, использующих FastAPI, используйте оригинальный fastapi.Depends

Группы команд

@command("admin.reload", group="admin", help="Перезагрузить модуль")
async def reload_handler(event):
    await event.reply("Модуль перезагружен")

@command("admin.stop", group="admin", help="Остановить бота")
async def stop_handler(event):
    await event.reply("Бот остановлен")

Параметр group используется только для группировки в справке; в примере выше admin.reload — это единое имя команды (точка — стиль именования, пользователь должен ввести /admin.reload).

Подкоманды

Имена команд поддерживают многотокенную структуру с разделением пробелами, реализуя подкоманды вроде /admin add, /admin user ban:

@command("admin", help="Команды администрирования")
async def admin_handler(event):
    await event.reply("Синтаксис: /admin add | /admin remove")

@command("admin add", help="Добавить администратора")
async def admin_add_handler(event):
    target = event.get_command_args()[0]
    await event.reply(f"Добавлен {target}")

@command("admin remove", aliases=["a remove"], help="Удалить администратора")
async def admin_remove_handler(event):
    await event.reply("Удалён")

Правила сопоставления (максимальное префиксное совпадение):

• /admin add x сначала попадает в admin add, event.get_command_args() возвращает ["x"] (параметры после подкоманды)
• Если зарегистрирована только admin, то /admin add x попадает в admin, get_command_args() возвращает ["add", "x"] (старое поведение не меняется)
• Алиасы поддерживают многотокенную форму (например, a remove), также можно использовать однотокенные алиасы (например, a) для указания подкоманды
• При регистрации родительской и подкоманды, не зарегистрированная подкоманда (например, /admin list x) возвращается к родительской команде

Наследование прав доступа: если подкоманда не декларирует permission, она автоматически наследует ближайшую родительскую команду, декларирующую permission — защита /admin автоматически защищает все её подкоманды; при декларации permission в подкоманде она имеет приоритет:

def is_admin(event):
    return event.get_user_id() in {"user123"}

@command("admin", permission=is_admin, help="Команды администрирования")
async def admin_handler(event):
    ...

# permission не нужно декларировать повторно, наследуется is_admin
@command("admin add", help="Добавить администратора")
async def admin_add_handler(event):
    ...

Примечание: master=True и hidden не наследуются, при необходимости декларируйте отдельно в подкоманде; пользовательские ACL (списки разрешений/запретов) сопоставляются по полному имени команды, шаблоны glob, например, "admin*" могут охватывать целую группу подкоманд.

В обзоре команд /help подкоманды автоматически отображаются вложенно под видимыми родительскими командами (например, admin → admin add с отступом на один уровень, admin user → admin user ban с отступом на два уровня).

Права доступа и управление доступом команд

Права доступа команд определяются на трёх уровнях, проверяются сверху вниз (если верхний уровень отклоняет, нижние уровни не проверяются):

# ① ACL команды (конфигурация со стороны пользователя): по белому и чёрному списку пользователей для команды, при отказе возвращается "Недостаточно прав"
# ② master=True — только владелец фреймворка может выполнить (фреймворк автоматически проверяет, при отказе возвращается "Недостаточно прав")
@command("restart", master=True, help="Перезапустить модуль")
async def restart_handler(event):
    await event.reply("Модуль перезапущен")

# ③ permission=вызов функции — логика управления доступом команды (выполняется, если функция возвращает True)
def is_admin(event):
    return event.get_user_id() in {"user123", "user456"}

@command("panel", permission=is_admin, help="Панель управления")
async def panel_handler(event):
    await event.reply("Добро пожаловать на панель управления")

ACL пользователей команды (ErisPulse.event.command.acl): пользователь может настроить белый и чёрный списки пользователей для любой команды, имена команд поддерживают точное и шаблонное сопоставление (например, "roll*"), при отказе возвращается "Недостаточно прав":

# config.toml — разрешить только 123456 выполнить restart; 666 всегда запрещён
[ErisPulse.event.command.acl.restart]
allow = ["onebot11:123456"]
deny = ["onebot11:666"]

Порядок проверки: если deny совпадает → отказ; если allow не пуст и не совпадает → отказ; если ACL не настроен, используется event.command.default_allow (false = строгий режим, без ACL — отказ; true — передаётся разработчику по умолчанию master=True / permission). API во время выполнения (поддержка шаблонов для имен команд):

from ErisPulse.Core.Event import command

command.allow_user("restart", "onebot11", "123456")   # список разрешённых
command.deny_user("restart", "onebot11", "666")       # список запрещённых
command.remove_acl("restart")                          # очистить белый и чёрный списки
command.get_acl("restart")                             # получить текущий список

Обработчики команд импортируются из пакета событий: from ErisPulse.Core.Event import command; также доступны через пакет SDK: sdk.Event.command (оба являются одним и тем же экземпляром). Обычно в модуле уже импортируются вместе с декоратором команды (from ErisPulse.Core.Event import command).

Управление доступом на уровне события (для конкретного пользователя / группы / бота) — через идентификационные области действия (scope.identity); управление доступностью модуля — через модульные области (scope.platforms / bots / sessions). См. Области действия (scope).

Рекомендация: для взаимодействия с бизнес-логикой внутри команды используйте master=True / permission; для управления доступом по пользователю / группе используйте идентификационные области; для управления доступностью модуля используйте модульные области.

Приоритет команд

# Чем больше значение приоритета, тем раньше выполнение
@message.on_message(priority=10)
async def high_priority_handler(event):
    await event.reply("Обработчик высокого приоритета")

@message.on_message(priority=1)
async def low_priority_handler(event):
    await event.reply("Обработчик низкого приоритета")

Параллельная обработка событий

Система событий ErisPulse использует модель параллельной обработки в пределах одного приоритета, последовательной между разными приоритетами:

Событие поступило
    ↓
группа с приоритетом 10: [обработчик C || обработчик D] параллельно → объединить результаты
    ↓ (если не прервано)
группа с приоритетом 0: [обработчик A || обработчик B] параллельно → объединить результаты
    ↓
...

• Параллельно в пределах одного приоритета: обработчики с одинаковым приоритетом выполняются одновременно, повышая пропускную способность
• Последовательно между приоритетами: группы с разными приоритетами выполняются по порядку (чем больше значение, тем раньше), обеспечивая выполнение обработчиков высокого приоритета первыми
• Копирование по требованию: обработчики не создают копию, если не изменяют данные, обеспечивая нулевые накладные расходы
• Обработка конфликтов: при одновременном изменении одного поля несколькими обработчиками используется последнее значение и записывается предупреждение в лог
• Механизм прерывания: при вызове event.done() (по умолчанию) или event.done(claim=False) любым обработчиком, выполнение последующих групп с низким приоритетом пропускается. Разница между присвоением и блокировкой описана в разделе Управление цепочкой: присвоение и блокировка

# Пример: параллельное выполнение обработчиков с одинаковым приоритетом
@message.on_message(priority=0)
async def handler_a(event):
    # Обработка задачи A
    event['result_a'] = process_a()

@message.on_message(priority=0)
async def handler_b(event):
    # Выполняется параллельно с handler_a
    event['result_b'] = process_b()

# Последовательное выполнение с разными приоритетами
@message.on_message(priority=10)
async def handler_c(event):
    # Наивысший приоритет, выполняется первым
    pass

Ограничение параллелизма: все соответствующие задачи немедленно создаются, но ограничиваются сигналом, ограничивающим количество одновременно выполняемых задач, по умолчанию 64 (ErisPulse.framework.handler_max_concurrency, поддержка горячего обновления). Задачи, превышающие лимит, ждут в очереди на сигнале, пока предыдущие не завершатся. Это ваш "сброс давления" при пиковых нагрузках.

Медленные логи: если обработка одной задачи занимает более 1 секунды, фреймворк записывает предупреждение в лог (handler_slow). Время ожидания wait_reply исключается из времени выполнения, чтобы "ожидание ответа" не приводило к ложным предупреждениям о медленной обработке.

Мидлвары: изменение или отклонение до распределения

Мидлвары выполняются до распределения событий по порядку, это правильное место для реализации сценариев, таких как брандмауэр, ограничение скорости, удаление конфиденциальных данных из событий:

from ErisPulse.Core import adapter

@adapter.middleware
async def firewall(data):
    if _is_banned(data.get("user_id")):
        return False          # Отклонение: событие отбрасывается, не попадает ни в один обработчик, без побочных эффектов
    data["checked"] = True    # Возврат dict: изменение данных события (сохранение старого поведения)
    # Возврат None: пропуск, данные события не меняются (старое поведение)
    return data

При отклонении фреймворк выводит трассировочный лог и запускает хук жизненного цикла adapter.event.blocked (с именем мидлвары и полным событием), что полезно для аудита "почему событие не ответило".

Цепочка принятия решений при распределении команд: почему команда не сработала

Сообщение команды последовательно проходит через: определение текста команды → соответствие имени команды/алиаса (при несовпадении предлагается исправление) → при совпадении команда считается принятой → область действия → пользовательский ACL → владелец → права → холодное ожидание/ограничение скорости → квоты (использование) → устаревание (deprecated, notice/rejected) → разбор параметров → выполнение (промежуточные обработчики могут отклонить на уровне событий, см. предыдущий раздел). Если какое-либо условие не выполняется, процесс прерывается; при нарушении правил (холодное ожидание/ограничение скорости/квоты) действия по умолчанию игнорируются, при отказе по правам пользователь получает ответ, а устаревшие команды отвечают согласно заявленным правилам (ответ/отказ).

В тестовой среде ErisPulse-Testing метод dispatch() напрямую возвращает эту цепочку решений (объект DispatchTrace, trace.explain() выводит построчный отчёт с причинами). В продакшен-среде можно использовать ErisPulse.Core.Event.start_dispatch_trace() для получения аналогичной записи.

Кроме того, ErisPulse.runtime предоставляет два набора API для диагностики: explain_module(имя_модуля) отвечает на вопрос "почему модуль не загрузился" (не зарегистрирован / ленивая загрузка / отключён конфигурацией / отсутствуют зависимости / не соответствует версии SDK, причины перечисляются по порядку), explain_event(событие) отвечает на вопрос "почему событие не обработано" (адаптер не зарегистрирован / аккаунт заблокирован / модуль отключил сессию / команда не соответствует); в связке с format_report() выводится удобочитаемый отчёт.

Фильтрация области: почему мой модуль не получил сообщение

После поступления события есть два молчаливых фильтра (ни один не отвечает, не выдает ошибки):

1. Идентификационное измерение (ErisPulse.scope.identity): при входе события в точку распределения проверяется, получает ли оно сообщение по пользователю > группе > боту > адаптеру. Отклоненные все события немедленно отбрасываются, ни один обработчик (включая распределитель команд) не срабатывает.
2. Модульное измерение (ErisPulse.scope): при поступлении события в обработчик модуля проверяется, доступен ли модуль по сеансу > боту > платформе, если не проходит, молча пропускается.

# Пример 1: все сообщения в группе не распространяются
[ErisPulse.scope.identity.sessions.onebot11."group_123"]
deny = true

# Пример 2: блокировка MyModule для определенного бота
[ErisPulse.scope.bots.onebot11."123456"]
blocked = ["MyModule"]

В этом случае события в группе не будут доставлены обработчикам команд и модуля MyModule. Это не ошибка, а механизм фильтрации — при диагностике "модуль не реагирует" сначала проверьте фильтрацию области по идентификатору и привязке модуля.

• Логи фильтрации видны только на уровне TRACE (core.scope.identity_denied / core.scope.denied), по умолчанию на уровне INFO следов нет
• Фреймворк-обработчики (например, распределитель команд scope_exempt=True) не подвержены влиянию модульного измерения, но подвержены влиянию идентификационного измерения (все событие уже отброшено)
• Перед выполнением команды есть третий фильтр: пользовательский ACL команды (при отклонении отображается "Недостаточно прав", см. предыдущий раздел)
• Четвертый фильтр — перезапись события (см. следующий раздел)

Конфигурация области, сопоставление, API времени выполнения смотрите в Области (scope).

Перезапись событий: без изменения кода модуля, перезаписать поведение любого типа событий

Параметры, объявленные при регистрации обработчика событий (например, pattern / regex / master / hidden), являются по умолчанию для разработчика. Система унифицированной перезаписи позволяет пользователю перезаписать поведение любого модуля по типу события — стандартные типы OneBot12 (meta / message / notice / request) и расширенные типы ErisPulse (command) имеют собственные перезаписываемые параметры:

# message: перезапись условия срабатывания текста (и с условиями в коде AND)
[ErisPulse.event.overrides.message.ChatModule]
pattern = "闲聊*"

# notice: только определенные подтипы уведомлений
[ErisPulse.event.overrides.notice.MyModule]
detail_types = ["group_increase"]

# command: перезапись параметров реализации (приоритет для пользователя — можно ужесточить или ослабить настройки разработчика)
[ErisPulse.event.overrides.command.MyModule.restart]
master = true
hidden = true

# acl: черно-белые списки пользователей команды (для нескольких команд glob)
[ErisPulse.event.overrides.acl."roll*"]
allow = ["onebot11:u_vip"]

# ACL по умолчанию (false = строгий режим: без ACL — отклонить)
acl_default_allow = true

API во время выполнения (from ErisPulse.Core.Event import overrides или sdk.Event.overrides, типы подпространства имен — симметричные set / get / delete для каждого типа)：

from ErisPulse.Core.Event import overrides

overrides.message.set("ChatModule", pattern="闲聊*")   # условие текста message
overrides.notice.set("MyModule", detail_types=["group_increase"])
overrides.command.set("MyModule", "restart", master=True)  # параметры команды
overrides.acl.set("roll*", deny=["onebot11:u_bad"])    # черный список пользователей команды

overrides.message.get("ChatModule")     # {"pattern": "闲聊*"}
overrides.message.delete("ChatModule")  # восстановить настройки разработчика

• Условия перезаписи и условия в обработчике кода одновременно действуют (AND семантика); command параметры и объявления разработчика глубоко объединяются (приоритет перезаписи)
• detail_types: событие без detail_type разрешается (не ошибочно удаляет неизвестные события)
• pattern / regex: события без текста (connect / heartbeat и т.д.) не ограничиваются, сразу разрешаются
• command ключ перезаписи master синхронизируется с ключом хранения must_master; отключение команды происходит через acl deny
• Ключ маппинга: overrides.command.set("My", "restart", master=True) ключ master является просто псевдонимом конфигурации, фактический ключ хранения и ключ возвращаемого значения get() унифицирован как must_master (в get() возвращается {"must_master": true}) — при чтении во время выполнения используется ключ хранения, не используйте ключ master для чтения
• Изменения конфигурации применяются немедленно (горячее обновление), формат проверяется (неизвестные параметры / плохие записи игнорируются)

Управление цепочкой: присвоение и блокировка

ErisPulse разделяет два ортогональных понятия "присвоение" и "блокировка", объединяя их через event.done(), что удобно для добавления слоев наблюдения (логирование, аудит, права доступа) вокруг обработки команд.

Точное определение двух понятий:

• Присвоение (claim): пометить событие как обработанное данным обработчиком (записать _processed). Распределитель команд видит уже присвоенное событие и пропускает повторную обработку — предотвращает дублирование обработки одного сообщения несколькими обработчиками команд. Типичный сценарий: после успешного сопоставления команды присвоить, чтобы распределитель команд больше не вмешивался.
• Блокировка (stop): предотвратить распространение события ниже по приоритету обработчиков (записать _propagation_stopped). Обработчики с более низким приоритетом больше не увидят это событие. Типичный сценарий: высокоприоритетный обработчик полностью обработал событие, не желая, чтобы низкоприоритетные обработчики выполнялись.

event.done(claim=, stop=) является псевдонимом event.mark_processed(claim=, stop=), параметры и поведение полностью эквивалентны.

@command("help")
async def help_cmd(event):
    event.done()            # Присвоение + блокировка (стандартный подход после обработки команды)

@message.on_message(priority=50)
async def observer(event):
    event.done(stop=False)  # Только присвоение: низкоприоритетные обработчики по-прежнему выполняются (логирование / статистика)

@message.on_message(priority=100)
async def firewall(event):
    if denied(event):
        event.done(claim=False)  # Только блокировка: низкоприоритетные обработчики не выполняются, но дедупликация не делается

Настройка блокировки для команд и ответов

Присвоение при сопоставлении команды: как только сообщение сопоставляется с зарегистрированным именем команды (включая подкоманды/псевдонимы), оно будет присвоено и по умолчанию заблокировано распространению — команды, отклоненные правами, больше не будут переданы низкоприоритетным обработчикам сообщений (устранение "повторного ответа" при сопоставлении команды, затем повторном срабатывании on_message).

Можно разрешить блокировку, чтобы низкоприоритетные наблюдатели (логирование / аудит / права) по-прежнему видели эти сообщения:

[ErisPulse.event.command]
block = false   # Сообщения команд продолжают распространяться до низкоприоритетных обработчиков (присвоение не влияет, повторная обработка невозможна)

[ErisPulse.event.wait_reply]
block = false   # Ответы, потребленные wait_reply, продолжают распространяться до низкоприоритетных обработчиков

Примечание: block контролирует только блокировку (stop), не влияет на присвоение (claim) — сопоставленные команды никогда не будут повторно обработаны обработчиками сообщений; сообщения, не сопоставленные ни одной команде, по-прежнему распространяются до обработчиков сообщений.

Обработка событий уведомлений

Добавление в друзья

from ErisPulse.Core.Event import notice

@notice.on_friend_add()
async def friend_add_handler(event):
    user_id = event.get_user_id()
    nickname = event.get_user_nickname() or "Новый друг"
    await event.reply(f"Добро пожаловать, {nickname}, добавьте меня в друзья!")

Увеличение участников группы

@notice.on_group_increase()
async def member_increase_handler(event):
    group_id = event.get_group_id()
    user_id = event.get_user_id()
    await event.reply(f"Добро пожаловать, {user_id}, в группу {group_id}")

Уменьшение участников группы

@notice.on_group_decrease()
async def member_decrease_handler(event):
    group_id = event.get_group_id()
    user_id = event.get_user_id()
    await event.reply(f"Пользователь {user_id} покинул группу {group_id}")

Обработка событий запросов

Запрос на добавление в друзья

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}")
    
    # Можно обработать запрос через API адаптера
    # Конкретная реализация см. в документации каждого адаптера

Запрос на группу

@request.on_group_request()
async def group_request_handler(event):
    group_id = event.get_group_id()
    user_id = event.get_user_id()
    
    await event.reply(f"Получено приглашение в группу {group_id} от {user_id}")

Обработка мета-событий

Событие подключения

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.warning(f"{platform} платформа отключена")

Событие сердцебиения

@meta.on_heartbeat()
async def heartbeat_handler(event):
    platform = event.get_platform()
    sdk.logger.debug(f"{platform} сердцебиение")

Запрос состояния бота

После того как адаптер отправляет мета-событие, фреймворк автоматически отслеживает состояние бота, и вы можете в любой момент проверить его:

from ErisPulse import sdk

# Проверить, онлайн ли определенный бот
if sdk.adapter.is_bot_online("telegram", "123456"):
    telegram = sdk.adapter.get("telegram")
    await telegram.Send.To("user", "123456").Text("Бот онлайн")

# Вывести список всех онлайн ботов
bots = sdk.adapter.list_bots()
for platform, bot_list in bots.items():
    for bot_id, info in bot_list.items():
        print(f"{platform}/{bot_id}: {info['status']}")

# Получить полную сводку состояния
summary = sdk.adapter.get_status_summary()

Интерактивная обработка

Использование метода reply для отправки ответа

Метод event.reply() поддерживает различные модификаторы, удобные для отправки сообщений с упоминанием, ответом и т.д.:

# Простой ответ
await event.reply("Привет")

# Отправка различных типов сообщений
await event.reply("http://example.com/image.jpg", method="Image")  # изображение
await event.reply("http://example.com/voice.mp3", method="Voice")  # голосовое сообщение

# Упоминание одного пользователя
await event.reply("Привет", at_users=["user123"])

# Упоминание нескольких пользователей
await event.reply("Всем привет", at_users=["user1", "user2", "user3"])

# Ответ на сообщение
await event.reply("Содержание ответа", reply_to="msg_id")

# Упоминание всех пользователей
await event.reply("Анонс", at_all=True)

# Комбинирование: упоминание пользователей + ответ на сообщение
await event.reply("Содержание", at_users=["user1"], reply_to="msg_id")

Ожидание ответа пользователя

@command("ask", help="Спросить пользователя")
async def ask_handler(event):
    await event.reply("Введите ваше имя:")
    
    # Ожидание ответа пользователя, таймаут 30 секунд
    reply = await event.wait_reply(timeout=30)
    
    if reply:
        name = reply.get_text()
        await event.reply(f"Привет, {name}!")
    else:
        await event.reply("Таймаут ожидания, пожалуйста, повторите ввод.")

Ожидание с проверкой

@command("age", help="Спросить возраст")
async def age_handler(event):
    def validate_age(event_data):
        """Проверка возраста на валидность"""
        try:
            age = int(event_data.get_text())
            return 0 <= age <= 150
        except ValueError:
            return False
    
    await event.reply("Введите ваш возраст (0-150):")
    
    reply = await event.wait_reply(
        timeout=60,
        validator=validate_age
    )
    
    if reply:
        age = int(reply.get_text())
        await event.reply(f"Ваш возраст: {age} лет")
    else:
        await event.reply("Неверный ввод или таймаут")

Ожидание с обратным вызовом

@command("confirm", help="Подтвердить действие")
async def confirm_handler(event):
    async def handle_confirmation(reply_event):
        text = reply_event.get_text().lower()
        
        if text in ["是", "yes", "y"]:
            await event.reply("Действие подтверждено!")
        else:
            await event.reply("Действие отменено.")
    
    await event.reply("Подтвердить выполнение этого действия? (是/否)")
    
    await event.wait_reply(
        timeout=30,
        callback=handle_confirmation
    )

Подтверждение диалога (confirm)

Ожидание подтверждения или отрицания пользователя, автоматически распознаются встроенные слова подтверждения на китайском и английском:

@command("confirm", help="Подтвердить действие")
async def confirm_handler(event):
    if await event.confirm("Вы уверены, что хотите выполнить это действие?"):
        await event.reply("Подтверждено, выполняется...")
    else:
        await event.reply("Отменено")

# Пользовательские слова подтверждения
if await event.confirm("Продолжить?", yes_words={"go", "继续"}, no_words={"stop", "停止"}):
    pass

Выбор меню (choose)

Пользователь может ответить номером опции или текстом опции:

@command("choose", help="Выбрать")
async def choose_handler(event):
    choice = await event.choose(
        "Выберите цвет:",
        ["красный", "зеленый", "синий"]
    )
    
    if choice is not None:
        colors = ["красный", "зеленый", "синий"]
        await event.reply(f"Вы выбрали: {colors[choice]}")
    else:
        await event.reply("Таймаут выбора")

Объединенный режим: merge_prompt=True вставляет опции в сообщение-подсказку, отправляя все в одном сообщении с указанным методом:

# Отправить объединенную подсказку + опции в Markdown
choice = await event.choose(
    "## Выберите цвет\n{options}\nПожалуйста, введите номер",
    ["красный", "зеленый", "синий"],
    method="Markdown",
    merge_prompt=True,
)

{options} — подстановочный знак для указания позиции вставки опций; если не указан, добавляется в конец подсказки. Можно настроить подстановочный знак через параметр placeholder (например, placeholder="[choices]"). options_format="auto" (по умолчанию) автоматически выбирает стиль в зависимости от метода: Markdown → маркированный список, Html → нумерованный список, другие → простой список. Для текстовых методов (Text/Markdown/Html и т.д.) по умолчанию объединяются опции в конец; для не-текстовых методов (Image и т.д.) по умолчанию отправляются две отдельные сообщения.

Сбор анкеты (collect)

Сбор ввода пользователя в несколько шагов:

@command("register", help="Регистрация")
async def register_handler(event):
    data = await event.collect([
        {"key": "name", "prompt": "Введите имя:"},
        {"key": "age", "prompt": "Введите возраст:", 
         "validator": lambda e: e.get_text().isdigit()},
        {"key": "email", "prompt": "Введите email:"}
    ])
    
    if data:
        await event.reply(f"Регистрация прошла успешно!\nИмя: {data['name']}\nВозраст: {data['age']}\nEmail: {data['email']}")
    else:
        await event.reply("Таймаут регистрации или неверный ввод")

Ожидание произвольного события (wait_for)

Ожидание события, соответствующего определенным условиям, независимо от пользователя:

@command("wait_member", help="Ожидание нового участника")
async def wait_member_handler(event):
    await event.reply("Ожидание нового участника в группе...")
    
    evt = await event.wait_for(
        event_type="notice",
        condition=lambda e: e.get_detail_type() == "group_member_increase",
        timeout=120
    )
    
    if evt:
        await event.reply(f"Добро пожаловать, {evt.get_user_id()}!")
    else:
        await event.reply("Таймаут ожидания")

Многошаговый диалог (conversation)

Создание интерактивного многошагового диалогового контекста:

@command("survey", help="Опрос")
async def survey_handler(event):
    conv = event.conversation(timeout=60)
    
    await conv.say("Добро пожаловать в опрос!")
    
    while conv.is_active:
        reply = await conv.wait()
        
        if reply is None:
            await conv.say("Диалог истек, до свидания!")
            break
        
        text = reply.get_text()
        
        if text == "выход":
            await conv.say("До свидания!")
            break
        
        await conv.say(f"Вы сказали: {text}, продолжайте ввод или ответьте 'выход' для завершения")

Встроенные слова подтверждения

ErisPulse включает в себя набор встроенных слов подтверждения на китайском и английском:

• Слова подтверждения (CONFIRM_YES_WORDS): 是, yes, y, confirmed, confirmed, ok, true, correct, agreed, no problem...
• Слова отрицания (CONFIRM_NO_WORDS): 否, no, n, canceled, not, don't, not, cancel, false, wrong, rejected, not allowed...

Доступ к данным события

Часто используемые методы объекта Event

@command("info")
async def info_handler(event):
    # Основная информация
    event_id = event.get_id()
    event_time = event.get_time()
    event_type = event.get_type()
    detail_type = event.get_detail_type()
    
    # Информация отправителя
    user_id = event.get_user_id()
    nickname = event.get_user_nickname()
    
    # Содержимое сообщения
    message_segments = event.get_message()
    alt_message = event.get_alt_message()
    text = event.get_text()
    
    # Информация о группе
    group_id = event.get_group_id()
    
    # Информация о боте
    self_id = event.get_self_user_id()
    self_platform = event.get_self_platform()
    
    # Оригинальные данные
    raw_data = event.get_raw()
    raw_type = event.get_raw_type()
    
    # Информация о платформе
    platform = event.get_platform()
    
    # Определение типа сообщения
    is_private = event.is_private_message()
    is_group = event.is_group_message()
    is_at = event.is_at_message()
    
    # Информация о команде
    if event.is_command():
        cmd_name = event.get_command_name()
        cmd_args = event.get_command_args()
        cmd_raw = event.get_command_raw()

Платформенно-специфические методы

Помимо встроенных методов, каждый адаптер платформы регистрирует платформенно-специфические методы, удобные для доступа к платформенно-специфическим данным.

from ErisPulse.Core.Event import message

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

    # Вызов платформенно-специфических методов в зависимости от платформы
    if platform == "telegram":
        chat_type = event.get_chat_type()      # Telegram специфичный метод
    elif platform == "email":
        subject = event.get_subject()           # Email специфичный метод

Если не уверены, зарегистрирован ли для платформы определенный метод, можно запросить, какие методы зарегистрированы для определенной платформы:

from ErisPulse.Core.Event import get_platform_event_methods

methods = get_platform_event_methods("telegram")
# ["get_chat_type", "is_bot_message", ...]

Список платформенно-специфических методов см. в соответствующей документации платформы。

Лучшие практики обработки событий

1. Обработка исключений

@command("process")
async def process_handler(event):
    try:
        # бизнес-логика
        result = await do_some_work()
        await event.reply(f"Результат: {result}")
    except ValueError as e:
        # ожидаемая бизнес-ошибка
        await event.reply(f"Ошибка параметра: {e}")
    except Exception as e:
        # неожиданная ошибка
        sdk.logger.error(f"Обработка не удалась: {e}")
        await event.reply("Произошла ошибка, попробуйте позже")

2. Логирование

@message.on_message()
async def message_handler(event):
    user_id = event.get_user_id()
    text = event.get_text()
    
    sdk.logger.info(f"Обработка сообщения: {user_id} - {text}")
    
    # Использование собственного логгера модуля
    from ErisPulse import sdk
    logger = sdk.logger.get_child("MyHandler")
    logger.debug(f"Подробная отладочная информация")

3. Условная обработка

@message.on_message(priority=0)
async def conditional_handler(event):
    """Условная обработка - проверка внутри обработчика"""
    # Обрабатывать только сообщения определенных пользователей
    if event.get_user_id() in ["bot1", "bot2"]:
        return
    
    # Обрабатывать только сообщения с определенным ключевым словом
    if "ключевое слово" not in event.get_text():
        return
    
    await event.reply("Условие выполнено, обрабатываем сообщение")

Далее

• Примеры распространенных задач - Узнайте, как реализовать часто используемые функции (включая продвинутую отправку сообщений: повтор, таймаут, пакетная отправка)
• Руководство по особенностям платформы - Полное объяснение Send DSL цепочечной отправки, правил отправки, пакетного построения
• Подробное руководство по обертке Event - Глубокое понимание объекта Event
• Руководство пользователя - Узнайте о настройке и управлении модулями