name: erispulse-getting-started description: "适用于：第一次接触 ErisPulse，想从零跑起一个机器人并理解核心概念。覆盖场景：ErisPulse — это асинхронный модульный робот-фреймворк, основанный на событийном подходе. Бизнес-логику нужно писать только один раз, а с помощью адаптеров развертывать на различных платформах, таких как QQ, Telegram, Kook, Yunhu и других. В комплекте предоставляется система плагинов, горячая перезагрузка и полный набор инструментов для разработчиков.。1. Установка (Python >= 3.10):。Когда стоит читать эту статью: бот уже запускается, и вы хотите написать первый обработчик команды /hello, а также разобраться, что именно делают sdk.run и декораторы команд. Если вы ещё не установили ErisPulse и не инициализировали проект, сначала выполните три шага из быстрого старта за 5 минут: «Установка», «Инициализация проекта» и «Запуск проекта»; если всё готово — пропустите.。Какие компоненты проходит сообщение от отправки пользователем до ответа бота? За что отвечают адаптеры и за что модули? Прочитав эту статью, вы сможете самостоятельно отвечать..."

ErisPulse 快速上手与核心概念

This skill is auto-built from the ErisPulse documentation. Repository: https://github.com/ErisPulse/ErisPulse Source paths are annotated per section; see docs/ in the repository for the latest content.

内容大纲

AI 按需读取：只读需要的章节即可，不必全文加载。

• 文档总入口与阅读路径: 87-97 行
• Выбор вашего пути: 99-210 行
• Справочник по расширенным функциям по версиям: 212-222 行
• Способы разработки: 224-229 行
• Другое: 231-235 行
• Получите помощь: 237-241 行
• Связанные ссылки: 243-247 行
• 快速开始: 249-255 行
• Эффект за 60 секунд: заставьте бота сначала ответить вам: 257-296 行
• Установка ErisPulse: три способа: 298-363 行
• Инициализация проекта: 365-397 行
• Установка модуля: 399-419 行
• Запуск проекта: 421-429 行
• Включение автодополнения в IDE (по желанию): 431-448 行
• Структура проекта: 450-460 行
• Конфигурационный файл: 462-476 行
• Далее: 478-503 行
• 第一个机器人: 505-514 行
• Шаг первый: написание первой команды: 516-563 行
• Шаг второй: Запуск робота: 565-573 行
• Шаг 3: Тестирование бота: 575-583 行
• Описание кода: 585-617 行
• Расширение: добавление дополнительных функций: 619-625 行
• Часто задаваемые вопросы: 627-663 行
• Следующие шаги: 665-669 行
• 基础概念: 671-680 行
• Событийно-ориентированная архитектура: 682-720 行
• Основные компоненты: 722-845 行
• Типы событий: 847-859 行
• Описание основных модулей: 861-963 行
• Отправка сообщений SendDSL: 965-1010 行
• Система ленивой загрузки: 1012-1033 行
• Следующие шаги: 1035-1038 行
• 事件处理: 1040-1049 行
• Обзор типов событий: 1051-1061 行
• Обработка событий сообщений: 1063-1133 行
• Обработка событий команд: 1135-1489 行
• Мидлвары: изменение или отклонение до распределения: 1491-1513 行
• Цепочка принятия решений при распределении команд: почему команда не сработала: 1515-1521 行
• Фильтрация области: почему мой модуль не получил сообщение: 1523-1551 行
• Перезапись событий: без изменения кода модуля, перезаписать поведение любого типа событий: 1553-1612 行
• Управление цепочкой: присвоение и блокировка: 1614-1663 行
• Обработка событий уведомлений: 1665-1697 行
• Обработка событий запросов: 1699-1726 行
• Обработка мета-событий: 1728-1775 行
• Интерактивная обработка: 1777-2001 行
• Доступ к данным события: 2003-2078 行
• Лучшие практики обработки событий: 2080-2131 行
• Далее: 2133-2137 行
• 常见任务: 2139-2146 行
• Содержание: 2148-2158 行
• Персистентность данных: 2160-2217 行
• Задачи по расписанию: 2219-2287 行
• Фильтрация сообщений: 2289-2328 行
• Адаптация к разным платформам: 2330-2371 行
• Продвинутая отправка сообщений (повторы/таймауты/пакетная отправка): 2373-2426 行
• Контроль прав доступа: 2428-2478 行
• Статистика сообщений: 2480-2524 行
• Функция поиска: 2526-2580 行
• Обработка изображений: 2582-2645 行
• Дальнейшие шаги: 2647-2651 行
• IDE 补全: 2653-2663 行
• Основные принципы проектирования: 2665-2671 行
• Основное использование: 2673-2681 行
• Использование в коде: 2683-2695 行
• Работа: 2697-2721 行
• Параметры командной строки: 2723-2730 行
• Когда перегенерировать: 2732-2736 行
• Отношение к стандартным методам SendDSL: 2738-2741 行
• Далее: 2743-2747 行

文档总入口与阅读路径

来源: docs/ru/README.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ru/README.md）

Документация ErisPulse

ErisPulse — это асинхронный модульный робот-фреймворк, основанный на событийном подходе. Бизнес-логику нужно писать только один раз, а с помощью адаптеров развертывать на различных платформах, таких как QQ, Telegram, Kook, Yunhu и других. В комплекте предоставляется система плагинов, горячая перезагрузка и полный набор инструментов для разработчиков.

Впервые используете? Начните с 5-минутного краткого руководства — от установки до запуска первого робота, всё пошагово.

Выбор вашего пути

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

1. Я хочу использовать бота

Запустите бота, установите модули и настройте.

2. Я хочу разрабатывать модули / адаптеры

Написание расширений для ErisPulse, которые можно распространять.

3. Я хочу глубже понять принципы работы

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

Архитектура и API

Обязательные статьи уровня L2 (обязательно для написания производственных модулей)

По сценариям (по мере необходимости)

Сводная таблица понятий

Когда сложно понять, какая подсистема отвечает за что, воспользуйтесь этой таблицей:

4. Экосистема и официальный клиент

Официальный клиент + модули экосистемы, которые можно устанавливать по мере необходимости (не входят в состав фреймворка).

5. Я хочу внести вклад в ErisPulse

Сделать фреймворк лучше

Справочник по расширенным функциям по версиям

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

Способы разработки

ErisPulse поддерживает два способа разработки:

• Разработка модуля (рекомендуется): создание отдельного пакета модуля, устанавливаемого через менеджер пакетов, удобно для распространения и управления.
• Встраиваемая разработка: непосредственное написание обработчиков в проекте, подходит для быстрого прототипирования. Подробнее см. Руководство.

Другое

• Стиль документации — правила написания документации при вкладе
• Вклад в ErisPulse — вход в проект
• Поддержка AI — проектные подсказки для помощников по программированию

Получите помощь

• GitHub репозиторий: https://github.com/ErisPulse/ErisPulse
• Сообщить о проблеме: создать Issue
• Техническое обсуждение: посмотреть Discussions

Связанные ссылки

• Стандарт OneBot12
• Официальная документация облака озера
• Telegram Bot API

快速开始

来源: docs/ru/quick-start.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ru/quick-start.md）

Быстрый старт

Это ваш первый шаг. Запустите робота ErisPulse за 5 минут — сначала посмотрите результат, а затем разберитесь с каждым шагом.

Эффект за 60 секунд: заставьте бота сначала ответить вам

1. Установка (Python >= 3.10):

pip install ErisPulse

2. Инициализация проекта. Мастер запросит имя проекта, позволит выбрать платформу адаптера и завершит настройку, после чего сгенерирует main.py и config/config.toml:

epsdk init

3. Откройте main.py и замените его на эти 9 строк кода:

from ErisPulse import sdk
from ErisPulse.Core.Event import command

@command("ping", help="Проверка, работает ли бот")
async def ping_handler(event):
    await event.reply("Pong! Бот работает нормально.")

if __name__ == "__main__":
    import asyncio
    asyncio.run(sdk.run(keep_running=True))

4. Запуск:

epsdk run main.py

В терминале появятся логи запуска фреймворка и информация о подключении адаптера. Теперь, если вы отправите боту /ping на подключенной платформе, он ответит:

Pong! Бот работает нормально.

Если вы увидели этот ответ, значит, цепочка фреймворка, адаптера и обработки событий заработала. Если не получилось, просто перейдите в конец документа по ссылке Возникли проблемы; если всё прошло успешно, читайте дальше, чтобы узнать, какие ещё есть варианты на каждом этапе.

Установка ErisPulse: три способа

Выше использовался самый прямой способ — pip. Выбирайте в зависимости от вашей среды:

Установка с помощью pip

Убедитесь, что у вас установлен Python версии >= 3.10, затем установите с помощью pip:

pip install ErisPulse

Если у вас установлен uv, можно использовать uv pip install ErisPulse — установка будет быстрее.

Если вы хотите установить команду epsdk глобально, чтобы не загрязнять окружение проекта, рекомендуется использовать uv tool install:

uv tool install ErisPulse

После установки команда epsdk будет доступна глобально: при запуске в директории проекта автоматически определяется .venv проекта (команда epsdk install устанавливает в окружение проекта, epsdk run запускает в окружении проекта), а сам фреймворк предоставляется из окружения инструмента. Подробнее см. руководство по установке.

Сценарий установки за один клик (рекомендуется для новых сред)

Сценарий автоматически определяет вашу среду (Docker, Python, uv) и предлагает наиболее подходящий способ установки, поддерживает несколько языков (китайский/English/日本語/Русский/繁體中文).

Windows (PowerShell):

irm https://get.erisdev.com/install.ps1 -OutFile install.ps1; powershell -ExecutionPolicy Bypass -File install.ps1

macOS / Linux:

curl -fsSL https://get.erisdev.com/install.sh -o install.sh && chmod +x install.sh && ./install.sh

Сценарий поможет вам выполнить следующие действия:

• Установка через Docker (рекомендуется, если Docker обнаружен): выбор источника образа (Docker Hub / GHCR), канал версий (стабильный / предварительный), настройка панели управления Dashboard, настройка портов
• Установка глобального CLI (uv tool, рекомендуется): установка epsdk как глобальной команды, без виртуального окружения, uv автоматически управляет Python (даже если версия системы слишком низкая); при запуске в директории проекта автоматически определяется .venv
• Традиционная установка: автоматическое создание виртуального окружения, выбор версии ErisPulse, необязательная установка модуля панели управления Dashboard

Использование Docker

В Docker-образе уже включён фреймворк ErisPulse и панель управления Dashboard, поддерживается архитектура linux/amd64 и linux/arm64.

# Скачайте docker-compose.yml
curl -O https://raw.githubusercontent.com/ErisPulse/ErisPulse/main/docker-compose.yml

# Установите токен панели управления и запустите
ERISPULSE_DASHBOARD_TOKEN=your-token docker compose up -d

image: ghcr.io/erispulse/erispulse:latest

После запуска перейдите по адресу http://<host>:8000/Dashboard и войдите, используя установленный токен.

Инициализация проекта

Интерактивная инициализация (рекомендуется)

epsdk init

Это запустит интерактивное руководство, которое поможет вам выполнить следующие шаги:

• Настройка имени проекта
• Настройка уровня логирования
• Настройка сервера (хост и порт)
• Выбор и настройка адаптера
• Создание структуры проекта

Быстрая инициализация

# Быстрый режим с указанием имени проекта
epsdk init -q -n my_bot

# Или просто указание имени проекта
epsdk init -n my_bot

Создание проекта вручную

Если вы предпочитаете создавать проект вручную:

mkdir my_bot && cd my_bot
epsdk init

Установка модуля

Установка через CLI

epsdk install Yunhu AIChat

Просмотр доступных модулей

epsdk list-remote

Интерактивная установка

При отсутствии имени пакета открывается интерактивный интерфейс установки:

epsdk install

Запуск проекта

# Обычный запуск
epsdk run main.py

# Режим горячей перезагрузки (рекомендуется при разработке)
epsdk run main.py --reload

Включение автодополнения в IDE (по желанию)

Динамически обнаруживаемые модули/адаптеры ErisPulse не поддерживаются автодополнением в IDE по умолчанию. Запустите следующую команду для генерации типовых заглушек:

epsdk types

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

from _ep_types import Yunhu
from ErisPulse import sdk

adapter: Yunhu = sdk.adapter.get("yunhu")
await adapter.Send.To("group", "123").Board(...)  # Автодополнение методов, специфичных для платформы

Структура проекта

Структура проекта после инициализации:

my_bot/
├── config/
│   └── config.toml          # Файл конфигурации
└── main.py                  # Точка входа

Конфигурационный файл

Базовый конфигурационный файл config.toml:

[ErisPulse.server]
host = "0.0.0.0"
port = 8000

[ErisPulse.logger]
level = "INFO"

[Yunhu_Adapter]
# Конфигурация адаптера

Далее

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

Хотите создать групповой бот (обработка сообщений, управление группой, уведомления)

• Введение в обработку событий — Способы обработки событий сообщений, уведомлений и запросов
• Примеры распространенных задач — Хранение данных, планирование задач, управление правами доступа

Хотите написать свой модуль / адаптер

• Основные понятия — Сначала поймите архитектуру адаптеров / модулей / событий
• Введение в разработку модулей
• Введение в разработку адаптеров
• Обзор архитектуры — Визуальная схема архитектуры

Хотите запустить бота на постоянной основе

• Руководство по развертыванию
• Описание конфигурационного файла

Возникли проблемы

• Создание первого бота — Подробное пошаговое руководство по созданию первого бота
• Процесс запуска и ручное управление — Диагностика сбоев при загрузке модулей / адаптеров
• Справочник команд CLI — Полный список подкоманд epsdk

第一个机器人

来源: docs/ru/getting-started/first-bot.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ru/getting-started/first-bot.md）

Создание первого бота

Успешно пройденный быстрый старт — это лишь первый шаг. Эта статья поможет вам написать первый обработчик команды и объяснит, что происходит с каждой строкой кода внутри фреймворка.

Шаг первый: написание первой команды

Откройте main.py и напишите простой обработчик команды:

from ErisPulse import sdk
from ErisPulse.Core.Event import command

@command("hello", help="Отправить приветственное сообщение")
async def hello_handler(event):
    """Обработка команды hello"""
    user_name = event.get_user_nickname() or "друг"
    await event.reply(f"Привет, {user_name}! Я бот ErisPulse.")

@command("ping", help="Проверить, работает ли бот")
async def ping_handler(event):
    """Обработка команды ping"""
    await event.reply("Pong! Бот работает нормально.")

async def main():
    """Основная точка входа"""
    print("Запуск ErisPulse...")
    
    # keep_running=True (по умолчанию): фреймворк блокируется и продолжает работу, пока не получит сигнал о завершении (например, Ctrl+C)
    await sdk.run(keep_running=True)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Параметр keep_running

Параметр sdk.run(keep_running) управляет тем, блокируется ли фреймворк для поддержания работы:

• keep_running=True (по умолчанию): run() будет блокироваться до тех пор, пока не получит сигнал о завершении (например, Ctrl+C), подходит для чистых ботов.
• keep_running=False: run() завершит инициализацию и немедленно вернёт управление. Фреймворк не будет остановлен — запущенные адаптеры/модули продолжат обрабатывать события сообщений в фоновом режиме, вы можете продолжить выполнять собственную логику, пока цикл событий не завершится и фреймворк не закроется. Например:

async def main():
    await sdk.run(keep_running=False)   # Инициализация завершена, немедленно возвращаем управление
    # Фреймворк уже работает в фоне, здесь можно делать что-то ещё
    while True:
        await asyncio.sleep(3600)
        print("Проверка каждый час")

Помимо двух режимов run(), существуют более тонкие способы управления жизненным циклом, такие как init()/uninit() для ручного управления, а также отдельный запуск и остановка адаптеров/маршрутизации. Подробнее см. Жизненный цикл и ручное управление.

Шаг второй: Запуск робота

# Обычный запуск
epsdk run main.py

# Режим разработки (поддержка горячей перезагрузки)
epsdk run main.py --reload

Шаг 3: Тестирование бота

Отправьте команду в вашей системе общения:

/hello

Вы должны получить ответ от бота.

Описание кода

Декоратор команды

@command("hello", help="Отправить приветственное сообщение")

• hello: имя команды, вызывается пользователем с помощью /hello
• help: описание команды, отображается в команде /help

Параметры события

async def hello_handler(event):

Параметр event представляет собой объект Event, содержащий:

• Текст сообщения: event.get_text()
• Информация о отправителе: event.get_user_id(), event.get_user_nickname()
• Информация о платформе: event.get_platform()
• Информация о группе: event.get_group_id()
• Исходные данные: event.get_raw()

Полный список методов объекта Event см. в разделе Подробное описание Event-объекта.

Отправка ответа

await event.reply("Содержимое ответа")

Метод event.reply() — это удобный способ отправки сообщения отправителю.

Расширение: добавление дополнительных функций

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

• Прослушивание сообщений: используйте @message.on_message() для прослушивания различных сообщений → Введение в обработку событий
• Прослушивание уведомлений: используйте @notice.on_friend_add() и другие для прослушивания системных уведомлений → Введение в обработку событий
• Хранение данных: используйте sdk.storage.get/set для постоянного хранения данных → Примеры распространенных задач

Часто задаваемые вопросы

Команда не отвечает?

1. Проверьте, правильно ли настроен адаптер, убедитесь, что status адаптера в файле config/config.toml установлен в true
2. Проверьте вывод логов в терминале, убедитесь, что нет ошибок (особенно сообщений уровня ERROR)
3. Убедитесь, что префикс команды верный (по умолчанию это /), можно проверить в разделе [ErisPulse.event.command] файла конфигурации
4. Убедитесь, что имя команды написано правильно, обратите внимание на настройки чувствительности к регистру

Как изменить префикс команды?

Добавьте в config.toml следующее:

[ErisPulse.event.command]
prefix = "!"
case_sensitive = false

Как поддерживать несколько платформ?

ErisPulse использует стандарт OneBot12 для унификации формата событий на разных платформах. Обработчики, зарегистрированные с помощью @command и @message, автоматически получают события со всех платформ. С помощью event.get_platform() можно определить источник платформы:

@command("hello")
async def hello_handler(event):
    platform = event.get_platform()
    
    if platform == "yunhu":
        await event.reply("Привет! От Юньху")
    elif platform == "telegram":
        await event.reply("Hello! От Telegram")
    else:
        await event.reply("Привет!")

Дополнительные советы по адаптации под несколько платформ можно найти в разделе Примеры типичных задач.

Следующие шаги

• Продолжайте по учебному пути: прочитайте Основные концепции, чтобы понять, что фреймворк делает «под капотом» обработчиков команд
• Хотите реагировать на больше типов сообщений: Введение в обработку событий — как писать обработчики для сообщений / уведомлений / запросов / мета-событий
• Хотите добавить практичные функции: Примеры распространённых задач — готовый код для персистентности, задач по расписанию и контроля прав доступа

基础概念

来源: docs/ru/getting-started/basic-concepts.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ru/getting-started/basic-concepts.md）

Основные концепции

Какие компоненты проходит сообщение от отправки пользователем до ответа бота? За что отвечают адаптеры и за что модули? Прочитав эту статью, вы сможете самостоятельно отвечать на вопросы по дизайну, возникающие при написании обработчиков, вместо того чтобы просто копировать примеры.

Событийно-ориентированная архитектура

ErisPulse использует событийно-ориентированную архитектуру: все взаимодействия передаются и обрабатываются через события.

Поток событий

Пользователь отправляет сообщение
      │
      ▼
Платформа принимает
      │
      ▼
Адаптер принимает нативное событие платформы
      │
      ▼
Преобразование в стандартное событие OneBot12
      │
      ▼
Передача в систему событий
      │
      ▼
Распределение зарегистрированным обработчикам
      │
      ▼
Модуль обрабатывает событие
      │
      ▼
Отправка ответа через адаптер
      │
      ▼
Платформа отображает пользователю

Стандарт OneBot12

ErisPulse использует OneBot12 в качестве основного стандарта событий. OneBot12 — это универсальный стандарт интерфейса приложений чат-ботов, определяющий единый формат событий.

Все адаптеры преобразуют события, специфичные для платформы, в формат OneBot12, обеспечивая согласованность кода.

Основные компоненты

1. Объект SDK

SDK — это единая точка входа для всей функциональности, предоставляющая доступ к основным компонентам.

from ErisPulse import sdk

# Доступ к основным модулям
sdk.storage    # Система хранения
sdk.config     # Система конфигурации
sdk.logger     # Система логирования
sdk.adapter    # Система адаптеров
sdk.module     # Система модулей
sdk.router     # Система маршрутизации
sdk.client     # HTTP-клиент
sdk.lifecycle  # Система жизненного цикла

2. Объект Event

Объект Event инкапсулирует данные события и предоставляет удобные методы доступа.

@command("info")
async def info_handler(event):
    # Получение информации о событии
    event_id = event.get_id()
    user_id = event.get_user_id()
    platform = event.get_platform()
    text = event.get_text()
    
    # Отправка ответа
    await event.reply(f"Пользователь: {user_id}, платформа: {platform}")

3. Адаптеры

Адаптеры — это мост между ErisPulse и внешними платформами.

Обязанности:

• Приём нативных событий платформы
• Преобразование в стандартный формат OneBot12
• Отправка событий в стандартном формате на платформу

Примеры адаптеров:

• Адаптер Yunhu: взаимодействие с платформой Юньху
• Адаптер Telegram: взаимодействие с Telegram Bot API
• Адаптер OneBot11: взаимодействие с приложениями, совместимыми с OneBot11
• Адаптер Email: обработка отправки и получения электронной почты

4. Модули

Модули — это базовые единицы расширения функциональности, которые могут:

• регистрировать обработчики событий
• реализовывать бизнес-логику
• вызывать адаптеры для отправки сообщений
• использовать сервисы, предоставляемые основными модулями

Механизм обнаружения модулей

ErisPulse обнаруживает установленные модули с помощью Python-механизма importlib.metadata.entry_points. Модуль объявляет точку входа в pyproject.toml:

[project.entry-points."erispulse.module"]
MyModule = "my_package:Main"

При инициализации SDK сканируются все точки входа группы erispulse.module, классы модулей регистрируются в ModuleManager, а затем инициализируются по очереди после топологической сортировки по зависимостям.

Минимальный рабочий модуль

from ErisPulse.Core.Bases import BaseModule
from ErisPulse import sdk

class Main(BaseModule):
    def __init__(self):
        self.sdk = sdk
        self.logger = sdk.logger.get_child("MyModule")

    async def on_load(self, event):
        self.logger.info("Модуль загружен")

    async def on_unload(self, event):
        self.logger.info("Модуль выгружен")

Жизненный цикл модуля

• Регистрация: SDK обнаруживает класс модуля и регистрирует его в менеджере
• Загрузка: создаётся экземпляр модуля, вызывается on_load(event) (event = {"module_name": "MyModule"})
• Выгрузка: вызывается on_unload(event), ресурсы очищаются

Стратегия загрузки

Поведение загрузки модуля объявляется через get_load_strategy():

from ErisPulse.loaders import ModuleLoadStrategy

class Main(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(
            lazy_load=True,   # Ленивая загрузка (по умолчанию True)
            priority=0        # Приоритет загрузки; чем больше число, тем раньше инициализация
        )

• lazy_load=True (по умолчанию): модуль инициализируется только при первом обращении через sdk.MyModule, что сокращает время запуска
• lazy_load=False: модуль инициализируется сразу при запуске SDK; подходит для модулей, которым нужно прослушивать события жизненного цикла или выполнять запланированные задачи
• priority: модули с одинаковым приоритетом загружаются в порядке регистрации; чем больше число, тем раньше инициализация

Подробное описание механизма ленивой загрузки см. в Система ленивой загрузки.

Типы событий

ErisPulse поддерживает 5 типов событий:

Подробное использование и примеры кода для каждого типа событий см. в Введение в обработку событий.

Описание основных модулей

Storage (Хранилище)

Система хранения «ключ-значение» на основе SQLite для сохранения данных.

# Установка значения
sdk.storage.set("key", "value")

# Получение значения
value = sdk.storage.get("key", "default_value")

# Массовые операции
sdk.storage.set_multi({
    "key1": "value1",
    "key2": "value2"
})

# Транзакции
with sdk.storage.transaction():
    sdk.storage.set("key1", "value1")
    sdk.storage.set("key2", "value2")

Config (Конфигурация)

Управление файлами конфигурации в формате TOML.

# Получение конфигурации
config = sdk.config.getConfig("MyModule", {})

# Установка конфигурации
sdk.config.setConfig("MyModule", {"key": "value"})

# Чтение вложенной конфигурации
value = sdk.config.getConfig("MyModule.subkey", "default")

Logger (Логирование)

Модульная система логирования.

# Запись в лог
sdk.logger.info("Это информационное сообщение")
sdk.logger.warning("Это предупреждение")
sdk.logger.error("Это ошибка")

# Получение дочернего логгера
child_logger = sdk.logger.get_child("submodule")
child_logger.info("Лог субмодуля")

Синтаксический сахар доступа через атрибуты

Помимо метода get_child(), вы можете создавать дочерние логгеры через доступ к атрибутам — это более лаконичная запись, своего рода синтаксический сахар:

# Создание дочернего логгера через доступ к атрибутам
sdk.logger.mymodule.info("Сообщение модуля")

# Поддерживается вложенный доступ
sdk.logger.mymodule.database.info("Сообщение базы данных")

Router (Маршрутизация)

Управление HTTP- и WebSocket-маршрутами на основе FastAPI + Uvicorn. Поддерживает маршрутизацию через декораторы, промежуточное ПО, группировку, ограничение частоты запросов и CORS.

from ErisPulse.Core import HttpRequest

@sdk.router.get("MyModule", "/api")
async def handler(request: HttpRequest):
    data = await request.json()
    return {"status": "ok"}

Полный API маршрутизации (WebSocket, промежуточное ПО, ограничение частоты запросов, CORS и т.д.) см. в Менеджер маршрутов.

Client (Сетевой клиент)

Унифицированный сетевой клиент, объединяющий HTTP-запросы, WebSocket-соединения, управление пулом соединений, автоматические повторные попытки, контроль таймаутов, статистику запросов и интеграцию с событиями жизненного цикла.

from ErisPulse.Core import client

# HTTP-запрос
resp = await client.get("https://api.example.com/users")
data = await resp.json()

# С повторными попытками и таймаутом
resp = await client.get(url, timeout=30, max_retries=3)

# WebSocket-соединение
ws = await client.ws_connect("wss://example.com/ws")
async for text in ws.iter_text():
    await ws.send_text(f"Echo: {text}")

Полный API сетевого клиента см. в Сетевой клиент.

Отправка сообщений SendDSL

Адаптеры предоставляют интерфейс отправки сообщений в виде цепочки вызовов.

Базовая отправка

# Получение экземпляра адаптера
yunhu = sdk.adapter.get("yunhu")

# Отправка сообщения
await yunhu.Send.To("user", "U1001").Text("Hello")

# Указание аккаунта для отправки
await yunhu.Send.Using("bot1").To("group", "G1001").Text("Сообщение в группу")

Цепочка модификаторов

# Упоминание пользователя
await yunhu.Send.To("group", "G1001").At("U2001").Text("@сообщение")

# Ответ на сообщение
await yunhu.Send.To("group", "G1001").Reply("msg123").Text("Ответ")

# Упоминание всех
await yunhu.Send.To("group", "G1001").AtAll().Text("Объявление")

Методы ответа Event

Объект Event предоставляет удобные методы ответа:

@command("test")
async def test_handler(event):
    # Простой текстовый ответ
    await event.reply("Содержимое ответа")
    
    # Отправка изображения
    await event.reply("http://example.com/image.jpg", method="Image")
    
    # Отправка голосового сообщения
    await event.reply("http://example.com/voice.mp3", method="Voice")

Система ленивой загрузки

ErisPulse по умолчанию включает ленивую загрузку модулей: модуль инициализируется только при первом обращении (например, sdk.MyModule), что значительно ускоряет запуск.

from ErisPulse.loaders import ModuleLoadStrategy

class Main(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(
            lazy_load=True,   # Включение ленивой загрузки (по умолчанию)
            priority=0        # Приоритет загрузки; чем больше число, тем раньше инициализация
        )

Сценарии, требующие отключения ленивой загрузки (lazy_load=False):

• Модули, прослушивающие события жизненного цикла (например, core.init.complete)
• Модули, запускающие запланированные задачи или фоновые сервисы
• Модули, которым необходимо завершить инициализацию до загрузки других модулей

Подробное описание механизма ленивой загрузки и важных замечаний см. в Система ленивой загрузки.

Следующие шаги

• Продолжайте по пути L1: следующая статья — Класс-обёртка Event; освойте все методы чтения и записи объекта события
• Хотите учиться в процессе написания кода: Введение в обработку событий содержит готовые к копированию варианты обработчиков для каждого типа событий; Примеры типовых задач охватывает сохранение данных, запланированные задачи, контроль прав доступа и другие распространённые функции

事件处理

来源: docs/ru/getting-started/event-handling.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ru/getting-started/event-handling.md）

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

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

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

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("Условие выполнено, обрабатываем сообщение")

Далее

• Хотите манипулировать самим объектом события: ознакомьтесь с подробным руководством по Event 包装类详解, где представлены все методы Event.
• Хотите отправлять более сложные сообщения: Platform Features Guide объясняет цепочечную отправку SendDSL, правила отправки и построение пакетов; в Common Tasks Examples есть готовый код для повторной попытки / таймаута / пакетной отправки.
• После написания функции нужно настроить и развернуть: перейдите в User Guide, где находятся файлы конфигурации и CLI-управление.

常见任务

来源: docs/ru/getting-started/common-tasks.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ru/getting-started/common-tasks.md）

Примеры типичных задач

Содержание

1. Персистентность данных
2. Задачи по расписанию
3. Фильтрация сообщений
4. Адаптация к разным платформам
5. Продвинутая отправка сообщений (повторы/таймауты/пакетная отправка)
6. Контроль прав доступа
7. Статистика сообщений
8. Функция поиска
9. Обработка изображений

Персистентность данных

Простой счётчик

from ErisPulse import sdk
from ErisPulse.Core.Event import command

@command("count", help="Просмотр количества вызовов команды")
async def count_handler(event):
    # Получить счётчик
    count = sdk.storage.get("command_count", 0)
    
    # Увеличить счётчик
    count += 1
    sdk.storage.set("command_count", count)
    
    await event.reply(f"Это {count}-й вызов этой команды")

Хранение данных пользователя

@command("profile", help="Просмотр профиля")
async def profile_handler(event):
    user_id = event.get_user_id()
    
    # Получить данные пользователя
    user_data = sdk.storage.get(f"user:{user_id}", {
        "nickname": "",
        "join_date": None,
        "message_count": 0
    })
    
    profile_text = f"""
Ник: {user_data['nickname']}
Дата присоединения: {user_data['join_date']}
Количество сообщений: {user_data['message_count']}
    """
    
    await event.reply(profile_text.strip())

@command("setnick", help="Установить ник")
async def setnick_handler(event):
    user_id = event.get_user_id()
    args = event.get_command_args()
    
    if not args:
        await event.reply("Пожалуйста, введите ник")
        return
    
    # Обновить данные пользователя
    user_data = sdk.storage.get(f"user:{user_id}", {})
    user_data["nickname"] = " ".join(args)
    sdk.storage.set(f"user:{user_id}", user_data)
    
    await event.reply(f"Ник установлен: {' '.join(args)}")

Задачи по расписанию

Простой таймер

from ErisPulse import sdk
from ErisPulse.Core.Event import command
import asyncio

class TimerModule:
    def __init__(self):
        self.sdk = sdk
        self._tasks = []
    
    async def on_load(self, event):
        """Запуск задач по расписанию при загрузке модуля"""
        self._start_timers()
        
        @command("timer", help="Управление таймером")
        async def timer_handler(event):
            await event.reply("Таймер работает...")
    
    def _start_timers(self):
        """Запуск задач по расписанию"""
        # Выполнять каждые 60 секунд
        task = asyncio.create_task(self._every_minute())
        self._tasks.append(task)
        
        # Выполнять каждый день в полночь
        task = asyncio.create_task(self._daily_task())
        self._tasks.append(task)
    
    async def _every_minute(self):
        """Задача, выполняемая каждую минуту"""
        self.sdk.logger.info("Выполнение минутной задачи")
        # Ваша логика...
    
    async def _daily_task(self):
        """Задача, выполняемая каждый день в полночь (примечание: расчёт по времени UTC; при необходимости локального времени скорректируйте самостоятельно)"""
        import time
        
        while True:
            # Вычислить время до полуночи
            now = time.time()
            midnight = now + (86400 - now % 86400)
            
            await asyncio.sleep(midnight - now)
            
            # Выполнить задачу
            self.sdk.logger.info("Выполнение ежедневной задачи")
            # Ваша логика...

Использование событий жизненного цикла

@sdk.lifecycle.on("core.init.complete")
async def init_complete_handler(event_data):
    """Запуск задач по расписанию после инициализации SDK"""
    import asyncio
    
    async def daily_reminder():
        """Ежедневное напоминание"""
        await asyncio.sleep(86400)  # 24 часа
        sdk.logger.info("Выполнение ежедневной задачи")
    
    # Запустить фоновую задачу
    asyncio.create_task(daily_reminder())

Фильтрация сообщений

Фильтрация по ключевым словам

from ErisPulse.Core.Event import message

blocked_words = ["спам", "реклама", "фишинг"]

@message.on_message()
async def filter_handler(event):
    text = event.get_text()
    
    # Проверить наличие запрещённых слов
    for word in blocked_words:
        if word in text:
            sdk.logger.warning(f"Заблокировано сообщение с запрещённым словом: {word}")
            return  # Не обрабатывать это сообщение
    
    # Обычная обработка сообщения
    await event.reply(f"Получено: {text}")

Фильтрация по чёрному списку

# Загрузить чёрный список из конфигурации или хранилища
blacklist = sdk.storage.get("user_blacklist", [])

@message.on_message()
async def blacklist_handler(event):
    user_id = event.get_user_id()
    
    if user_id in blacklist:
        sdk.logger.info(f"Пользователь из чёрного списка: {user_id}")
        return  # Не обрабатывать
    
    # Обычная обработка
    await event.reply(f"Привет, {user_id}")

Адаптация к разным платформам

Ответы для конкретных платформ

@command("help", help="Показать справку")
async def help_handler(event):
    platform = event.get_platform()
    
    if platform == "yunhu":
        await event.reply("Справка для платформы Yunhu...")
    elif platform == "telegram":
        await event.reply("Telegram platform help...")
    elif platform == "onebot11":
        await event.reply("OneBot11 help...")
    else:
        await event.reply("Общая справка")

Определение возможностей платформы

@command("rich", help="Отправить форматированное сообщение")
async def rich_handler(event):
    platform = event.get_platform()
    
    if platform == "yunhu":
        # Yunhu поддерживает HTML
        yunhu = sdk.adapter.get("yunhu")
        await yunhu.Send.To("user", event.get_user_id()).Html(
            "<b>Жирный текст</b><i>Курсивный текст</i>"
        )
    elif platform == "telegram":
        # Telegram поддерживает Markdown
        telegram = sdk.adapter.get("telegram")
        await telegram.Send.To("user", event.get_user_id()).Markdown(
            "**Жирный текст** *Курсивный текст*"
        )
    else:
        # На других платформах использовать простой текст
        await event.reply("Жирный текст Курсивный текст")

Продвинутая отправка сообщений (повторы/таймауты/пакетная отправка)

Помимо простого event.reply(), с помощью Send DSL адаптера можно реализовать более сложные сценарии отправки: автоматический повтор при неудаче, отмена по таймауту, выполнение логики после успешной отправки, пакетная отправка нескольких сообщений.

В примерах ниже event.get_detail_type() и event.get_target_id() используются для получения типа цели и ID из события (в групповом чате автоматически берётся group_id, в личном — user_id), чтобы избежать жёсткого кодирования.

Выполнение логики после успешной отправки

@command("pay", help="Имитация оплаты")
async def pay_handler(event):
    yunhu = sdk.adapter.get(event.get_platform())
    user_id = event.get_user_id()
    # Списывать баллы только после успешной отправки
    await (yunhu.Send.To(event.get_detail_type(), event.get_target_id())
           .Hook(lambda r: sdk.storage.set(f"points:{user_id}", -10))
           .Text("Оплата прошла успешно, списано 10 баллов"))

Повтор при неудаче + отмена по таймауту

@command("notice", help="Отправить важное уведомление")
async def notice_handler(event):
    adapter_inst = sdk.adapter.get(event.get_platform())
    # Не более 3 повторов, таймаут 10 секунд каждый
    task = (adapter_inst.Send.To(event.get_detail_type(), event.get_target_id())
            .Retry(3)
            .Timeout(10)
            .OnError(lambda ctx: sdk.logger.error(f"Не удалось отправить уведомление: {ctx.error}"))
            .Text("Это важное уведомление"))
    # Не ждать, отправка в фоне

Пакетная отправка нескольких сообщений

Несколько сообщений по одной цепочке, единое выполнение:

@command("announce", help="Отправить объявление")
async def announce_handler(event):
    adapter_inst = sdk.adapter.get(event.get_platform())
    # Собрать несколько сообщений и отправить вместе (по умолчанию параллельно)
    results = await (adapter_inst.Send.To(event.get_detail_type(), event.get_target_id())
                    .Build()
                    .Text("📋 Объявление на сегодня")
                    .Image("https://example.com/banner.jpg")
                    .Text("Подробности смотрите на изображении выше")
                    .Retry(2)            # Каждый неудавшийся элемент повторяется отдельно
                    .send_all())
    sdk.logger.info(f"Пакетная отправка завершена, всего {len(results)} сообщений")

Более полные правила и описание пакетной отправки смотрите в Руководстве по возможностям платформ.

Контроль прав доступа

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

# Список владельцев в конфигурации
MASTERS = ["user123", "user456"]

def is_master(user_id):
    """Проверить, является ли пользователь владельцем фреймворка"""
    return user_id in MASTERS

@command("master", help="Команда владельца фреймворка")
async def master_handler(event):
    user_id = event.get_user_id()
    
    if not is_master(user_id):
        await event.reply("Недостаточно прав, эта команда доступна только владельцу фреймворка")
        return
    
    await event.reply("Команда владельца фреймворка выполнена успешно")

@command("addmaster", help="Добавить владельца фреймворка")
async def addmaster_handler(event):
    if not is_master(event.get_user_id()):
        return
    
    args = event.get("text", "").split()
    if len(args) < 2:
        await event.reply("Использование: /addmaster <ID пользователя>")
        return
    
    new_master = args[0]
    MASTERS.append(new_master)
    await event.reply(f"Владелец фреймворка добавлен: {new_master}")

Права в группах

@command("groupinfo", help="Просмотр информации о группе")
async def groupinfo_handler(event):
    if not event.is_group_message():
        await event.reply("Эта команда доступна только в групповых чатах")
        return
    
    group_id = event.get_group_id()
    user_id = event.get_user_id()
    
    await event.reply(f"ID группы: {group_id}, ваш ID: {user_id}")

Статистика сообщений

Подсчёт сообщений

Примечание: в следующих примерах для простого подсчёта используются sdk.storage.get/set. В сценариях с высокой конкурентностью рекомендуется использовать sdk.storage.transaction() для обеспечения атомарности.

@message.on_message()
async def count_handler(event):
    # Получить статистику
    stats = sdk.storage.get("message_stats", {
        "total": 0,
        "by_user": {},
        "by_day": {}
    })
    
    # Обновить статистику
    stats["total"] += 1
    
    user_id = event.get_user_id()
    stats["by_user"][user_id] = stats["by_user"].get(user_id, 0) + 1
    
    # Сохранить
    sdk.storage.set("message_stats", stats)

@command("stats", help="Просмотр статистики сообщений")
async def stats_handler(event):
    stats = sdk.storage.get("message_stats", {
        "total": 0,
        "by_user": {},
        "by_day": {}
    })
    
    top_users = sorted(
        stats["by_user"].items(),
        key=lambda x: x[1],
        reverse=True
    )[:5]
    
    top_text = "\n".join(
        f"{uid}: {count} сообщений" for uid, count in top_users
    )
    
    await event.reply(f"Всего сообщений: {stats['total']}\n\nАктивные пользователи:\n{top_text}")

Функция поиска

Простой поиск

Примечание: в следующем примере история сообщений хранится в списке в памяти, после перезапуска программы данные будут потеряны. В продакшене рекомендуется использовать sdk.storage или таблицу SQLite для персистентного хранения.

from ErisPulse.Core.Event import command, message

# Хранилище истории сообщений
message_history = []

@message.on_message()
async def store_handler(event):
    """Сохранение сообщений для поиска"""
    user_id = event.get_user_id()
    text = event.get_text()
    
    message_history.append({
        "user_id": user_id,
        "text": text,
        "time": event.get_time()
    })
    
    # Ограничить количество записей истории
    if len(message_history) > 1000:
        message_history.pop(0)

@command("search", help="Поиск сообщений")
async def search_handler(event):
    args = event.get_command_args()
    
    if not args:
        await event.reply("Пожалуйста, введите ключевое слово для поиска")
        return
    
    keyword = " ".join(args)
    results = []
    
    # Поиск по истории
    for msg in message_history:
        if keyword in msg["text"]:
            results.append(msg)
    
    if not results:
        await event.reply("Совпадающих сообщений не найдено")
        return
    
    # Показать результаты
    result_text = f"Найдено {len(results)} совпадающих сообщений:\n\n"
    for i, msg in enumerate(results[:10], 1):  # Показывать максимум 10 сообщений
        result_text += f"{i}. {msg['text']}\n"
    
    await event.reply(result_text)

Обработка изображений

Скачивание и сохранение изображений

from ErisPulse.Core import client

@message.on_message()
async def image_handler(event):
    """Обработка сообщений с изображениями"""
    message_segments = event.get_message()
    
    for segment in message_segments:
        if segment.get("type") == "image":
            file_url = segment.get("data", {}).get("file")
            
            if file_url:
                # Рекомендуется использовать встроенный клиент SDK для скачивания изображений
                resp = await client.get(file_url)
                if resp.status == 200:
                    image_data = await resp.read()
                    
                    # Сохранить в файл
                    filename = f"images/{event.get_time()}.jpg"
                    with open(filename, "wb") as f:
                        f.write(image_data)
                    
                    sdk.logger.info(f"Изображение сохранено: {filename}")
                    await event.reply("Изображение сохранено")

Пример распознавания изображений

Примечание: в следующем примере используется адрес-заглушка API; при реальном использовании замените его на свой сервис распознавания изображений.

from ErisPulse.Core import client

@command("identify", help="Распознать изображение")
async def identify_handler(event):
    """Распознавание изображений в сообщении"""
    message_segments = event.get_message()
    
    for segment in message_segments:
        if segment.get("type") == "image":
            file_url = segment.get("data", {}).get("file")
            
            # Вызов API распознавания изображений
            result = await _identify_image(file_url)
            
            await event.reply(f"Результат распознавания: {result}")
            return
    
    await event.reply("Изображение не найдено")

async def _identify_image(url):
    """Вызов API распознавания изображений (пример) — с использованием встроенного клиента SDK"""
    resp = await client.post(
        "https://api.example.com/identify",
        json={"url": url}
    )
    data = await resp.json()
    return data.get("description", "Не удалось распознать")

Дальнейшие шаги

• Функция написана и должна работать постоянно: смотрите Руководство по развёртыванию — демон Docker / systemd, резервное копирование и обновление
• Хотите распространять написанные функции среди других: начните с Введения в разработку модулей и превратите скрипт в устанавливаемый модуль
• Нужно настроить конфигурацию или разобраться, почему «модуль не реагирует»: смотрите Описание файла конфигурации и Области видимости (scope)

IDE 补全

来源: docs/ru/getting-started/ide-completion.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ru/getting-started/ide-completion.md）

Генерация типовых заглушек (дополнение IDE)

ErisPulse использует динамическое обнаружение модулей/адаптеров через entry-points, и точки входа не могут быть известны на статическом уровне конкретные типы пользовательских классов.
Команда epsdk types сканирует установленные модули/адаптеры и генерирует файл типовых заглушек, позволяя использовать эти типы в аннотациях переменных и получать дополнение в IDE.

Основные принципы проектирования

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

• Все импорты находятся внутри TYPE_CHECKING, без накладных расходов во время выполнения, без изменения поведения
• Имена типов используют PascalCase, соответствующий названию entry-point (например, yunhu → Yunhu), что соответствует названию, передаваемому в sdk.adapter.get() / sdk.module.get()
• Пользователи в коде по-прежнему используют sdk.module.get(...) / sdk.adapter.get(...) для получения экземпляров, просто используя импортированные типы для аннотации переменных

Основное использование

Запустите в корневом каталоге проекта:

epsdk types

В текущем каталоге будет создан файл _ep_types.py, содержащий типы всех установленных модулей/адаптеров.

Использование в коде

from _ep_types import MyModule, Yunhu
from ErisPulse import sdk

# Используя импортированные типы для аннотации переменных, IDE будет предлагать методы этого класса
my_mod: MyModule = sdk.module.get("MyModule")
my_mod.hello()                  # ← IDE предлагает hello

my_adapter: Yunhu = sdk.adapter.get("yunhu")
await my_adapter.Send.To("group", "123").Board(...)   # ← Предложения специфичных методов платформы

Работа

1. Сканирование entry-points erispulse.adapter / erispulse.module
2. Инспекция в целевой среде Python через дочерний процесс для сбора информации о фактических классах каждого адаптера/модуля (включая путь модуля и полное имя)
3. Генерация .py файла, в котором:
   • Все from xxx import Yyy as Zzz находятся в TYPE_CHECKING
   • Zzz представляет собой имя entry-point в формате PascalCase
4. IDE читает часть TYPE_CHECKING для предоставления автодополнения; во время выполнения никакой код не выполняется

Пример сгенерированных заглушек:

# _ep_types.py (автоматически сгенерировано)
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    # Адаптеры
    from MyAdapter.Core import MyAdapter as MyAdapter
    from YunhuAdapter.Core import YunhuAdapter as Yunhu

    # Модули
    from MyModule.Core import Main as MyModule

    __all__ = ['MyAdapter', 'Yunhu', 'MyModule']

Параметры командной строки

Когда перегенерировать

• После установки или удаления новых модулей или адаптеров
• После обновления публичного API модуля/адаптера
• Когда автодополнение IDE перестаёт работать или типы устаревают

Отношение к стандартным методам SendDSL

Базовый класс SendDSL уже содержит стандартные методы отправки (Text/Image/Voice/Video/File), и любой полученный экземпляр SendDSL может дополнить эти методы. Команда types в основном используется для дополнения методов, специфичных для платформы (например, Board для Yunhu, Dice для Sandbox) и методов, специфичных для модуля.

Далее

• Хотите узнать, какие методы отправки универсальны для всех платформ?: ознакомьтесь с подробным руководством по SendDSL, где содержится официальное описание стандартных методов, таких как Text / Image / Voice
• Разрабатываете собственный адаптер?: начните с введения в разработку адаптеров, чтобы создать проект адаптера