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

name: erispulse-module-dev description: "适用于：要开发 ErisPulse 模块（业务逻辑载体）：命令、事件、多轮对话、ORM。覆盖场景：Следуя этой статье, вы получите запускаемый модуль, который будет отвечать на команду /hello, и узнаете структуру проекта модуля, способ регистрации и обработчики загрузки/выгрузки — все последующие документы по модулям строятся на этой основе.。В этой статье подробно объясняется полный жизненный цикл модуля от загрузки до выгрузки, методы объявления конфигурации и хранения данных — что делает за вас фреймворк, а что вам нужно написать самостоятельно. Прочитав статью, вы получите полное представление.。Параметр event в обработчике может напрямую вызывать .get_text() / .reply() / .wait_reply() именно благодаря классу-обёртке Event. В этой статье приведён полный список его методов — обращайтесь к ней по мере необходимости при написании обработчиков событий; сигнатуры методов и их поведение определяются именно здесь.。Когда читать эту статью: рекомендуется прочитать после написания первого модуля (инженерные лучшие практики). Если не нужно, пропустит..."

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 按需读取：只读需要的章节即可，不必全文加载。

模块开发入门

来源: docs/ru/developer-guide/modules/getting-started.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ru/developer-guide/modules/getting-started.md）

Введение в разработку модулей

Note

Обязательно к прочтению на уровне L1 — одна из пяти ключевых статей в разделе "Путь чтения" на главной странице. Необходима для разработки модулей.

Следуя этой статье, вы получите запускаемый модуль, который будет отвечать на команду /hello, и узнаете структуру проекта модуля, способ регистрации и обработчики загрузки/выгрузки — все последующие документы по модулям строятся на этой основе.

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

Стандартная структура модуля:

MyModule/
├── pyproject.toml
├── README.md
├── LICENSE
└── MyModule/
    ├── __init__.py
    └── Core.py

Конфигурация pyproject.toml

[project]
name = "ErisPulse-MyModule"
version = "1.0.0"
description = "Описание функционала модуля"
readme = "README.md"
requires-python = ">=3.10"
license = { file = "LICENSE" }
authors = [ { name = "yourname", email = "[email protected]" } ]
dependencies = []

[project.urls]
"homepage" = "https://github.com/yourname/MyModule"

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

__init__.py

from .Core import Main

Core.py - Базовый модуль

from ErisPulse import sdk
# Рекомендуется корневой импорт (Python 3.10+): часто используемые символы импортируются напрямую из корневого пакета, но и глубокие пути по-прежнему поддерживаются
from ErisPulse import BaseModule, command

class Main(BaseModule):
    def __init__(self, sdk):
        self.sdk = sdk
        self.logger = sdk.logger.get_child("MyModule")
        self.storage = sdk.storage
    
    @staticmethod
    def get_load_strategy():
        """Возвращает стратегию загрузки модуля"""
        from ErisPulse.loaders import ModuleLoadStrategy
        return ModuleLoadStrategy(
            lazy_load=True,
            priority=0,
            depends=[],  # Необязательно: список модулей, от которых зависит данный модуль
            # Необязательно: ленивая активация по событиям — объявите триггеры, модуль автоматически загрузится при первом совпадении события/команды
            # activate_on=[{"command": {"name": "hello", "help": "Отправить приветствие"}}],
        )
    
    async def on_load(self, event):
        """Вызывается при загрузке модуля"""
        @command("hello", help="Отправить приветствие")
        async def hello_command(event):
            name = event.get_user_nickname() or "друг"
            await event.reply(f"Привет, {name}!")
        
        self.logger.info("Модуль загружен")
    
    async def on_unload(self, event):
        """Вызывается при выгрузке модуля"""
        self.logger.info("Модуль выгружен")

Чтение конфигурации: в приведённом базовом примере конфигурация не используется. При необходимости чтения конфигурации рекомендуется объявить вложенный класс ConfigClass и читать данные в реальном времени через self.cfg (см. Основные концепции модуля). Старый способ — ручной вызов _load_config() — устарел.

Тестирование модуля

Локальное тестирование

# Установите модуль в директории проекта
epsdk install ./MyModule

# Запустите проект
epsdk run main.py --reload

Тестовая команда

Отправьте команду для тестирования:

/hello

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

Базовый класс BaseModule

Все модули должны наследоваться от BaseModule, предоставляя следующие методы:

Метод Описание Обязательно
__init__(self, sdk) Конструктор (фреймворк передаёт экземпляр sdk) Нет
get_load_strategy() Возвращает стратегию загрузки Нет
get_meta() Возвращает метаинформацию о модуле (необязательно) Нет
on_load(self, event) Вызывается при загрузке модуля Да
on_unload(self, event) Вызывается при выгрузке модуля Да

Метаинформация модуля

Note

Эта функция доступна начиная с ErisPulse 2.8.0+.

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

Как и в случае с get_load_strategy(), возвращайте экземпляр класса ModuleMeta (с поддержкой типизации и автодополнения в IDE), но также допускается возвращать словарь:

class MyModule(BaseModule):
    @staticmethod
    def get_meta() -> ModuleMeta:
        return ModuleMeta(
            name="Погода",               # Отображаемое имя (по умолчанию имя регистрации)
            description="Получение погоды в городе",  # Описание модуля
            version="1.0.0",
            author="ErisDev",
            group="Инструменты",               # Категория функционала
            tags=["Погода", "Поиск"],
        )

Совместимый способ (через словарь):

class MyModule(BaseModule):
    @staticmethod
    def get_meta() -> dict:
        return {
            "name": "Погода",
            "description": "Получение погоды в городе",
            "version": "1.0.0",
            "author": "ErisDev",
            "group": "Инструменты",
            "tags": ["Погода", "Поиск"],
        }

Поддержка i18n для полей метаинформации

Значения полей метаинформации могут быть простыми строками или словарями i18n {"i18n": "key.path", "default": "текст по умолчанию"} (согласно соглашению для description в конфигурации). Ключи перевода объявляются через I18nClass, а при чтении module.get_meta() автоматически разрешаются в текст текущего языка:

class MyModule(BaseModule):
    class I18nClass(BaseI18n):
        meta_description: I18nKey = I18nKey(
            default="Weather lookup",
            zh_CN="查询城市天气",
            en="Weather lookup",
        )

    @staticmethod
    def get_meta() -> ModuleMeta:
        return ModuleMeta(
            name="Погода",
            description={"i18n": "MyModule.meta_description", "default": "Weather lookup"},
        )

Объект SDK

Доступ к основным функциям осуществляется через объект sdk:

from ErisPulse import sdk

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

Далее

Продолжение пути обучения

После написания

模块核心概念

来源: docs/ru/developer-guide/modules/core-concepts.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ru/developer-guide/modules/core-concepts.md）

Основные концепции модуля

Note

Обязательный к прочтению для L1 — одна из пяти основных документов в разделе "Путь чтения" на главной странице, необходима для разработки модуля.

В этой статье подробно объясняется полный жизненный цикл модуля от загрузки до выгрузки, методы объявления конфигурации и хранения данных — что делает за вас фреймворк, а что вам нужно написать самостоятельно. Прочитав статью, вы получите полное представление.

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

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

from ErisPulse.Core.Bases import BaseModule
from ErisPulse.loaders import ModuleLoadStrategy

class MyModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        """Возвращает стратегию загрузки модуля"""
        return ModuleLoadStrategy(
            lazy_load=True,   # Загрузка по требованию или немедленная загрузка
            priority=0,       # Приоритет загрузки (чем больше, тем выше приоритет)
            depends=["OtherModule"]  # Необязательно: объявить зависимости от других модулей
        )

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

Note

Каскадное выгрузка / перезагрузка (ErisPulse 2.8.0+): при выгрузке модуля, от которого зависят другие модули, модули, зависящие от него, будут предварительно каскадно выгружены (в логах будет указан каскадный цепной путь); при горячей перезагрузке любого модуля (локальный плагин / модуль, зарегистрированный во время выполнения / пакет из PyPI), модули, от которых зависит этот модуль, также каскадно перезагружаются, чтобы избежать использования устаревших ссылок на экземпляры. Циклические зависимости будут отклонены с ошибкой RuntimeError при загрузке.

Перезагрузка — полная перезагрузка (ErisPulse 2.11.0+): reload() по умолчанию очищает кэш импорта по имени пакета и перезагружает зависимости, удаляя старые экземпляры до выгрузки (при ошибке отклоняется, старые экземпляры продолжают обслуживать), после успешной проверки подлинности кода (предупреждение о ложной перезагрузке / stale метка); при неудачной загрузке можно получить информацию о последней ошибке с помощью sdk.module.last_error(имя) (с traceback).

Метод on_load

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

async def on_load(self, event):
    # Регистрация обработчика события
    @command("hello", help="Команда приветствия")
    async def hello_handler(event):
        await event.reply("Привет!")

    # Использование встроенного HTTP-клиента SDK (автоматически управляет пулом соединений, не нужно создавать session вручную)
    # Отправлять запросы можно через sdk.client

Метод on_unload

Вызывается при выгрузке модуля, используется для очистки ресурсов:

async def on_unload(self, event):
    # Очистка пользовательских ресурсов
    # sdk.client управляется фреймворком, закрывать вручную не нужно
    
    # Отмена обработчиков событий (фреймворк автоматически обрабатывает)
    self.logger.info("Модуль выгружен")

Детали создания и очистки фоновых задач (используя self.spawn() / фреймворк отменяет по умолчанию) см. в управлении жизненным циклом.

Метод on_before_unload

Note

Этот хук доступен начиная с ErisPulse 2.11.0+.

Опциональный хук, вызываемый на первом этапе цепочки выгрузки (до on_unload и освобождения ресурсов) — в этот момент все ресурсы модуля еще доступны, подходит для сохранения состояния в sdk.storage перед выгрузкой / сном / перезагрузкой, чтобы восстановить его при следующем on_load (продолжение сна при ленивой загрузке основано на этом):

async def on_before_unload(self, event):
    # event["reason"] указывает причину: unload / reload / disable / cascade / dormant
    if event["reason"] in ("reload", "dormant"):
        await self.sdk.storage.aset(f"{self._module_name}:state", self.collect_state())

Если хук завершается с ошибкой, это только предупреждение, не блокирует процесс выгрузки; модули без объявленного хука ведут себя как обычно. Семантика сна / пробуждения описана в Ленивой загрузке сна.

Выгрузка и полная выгрузка (purge)

Note

Эта функция доступна начиная с ErisPulse 2.8.0+.

unload() по умолчанию только отменяет загрузку (выгружает экземпляр и ресурсы), но сохраняет метаданные регистрации (класс модуля и метаинформация) — модуль всё ещё может быть обнаружен и перезагружен с помощью load(), без необходимости повторной register().

Для полной выгрузки (освобождение ссылки на класс модуля, очистка sys.modules, чтобы плагин и его уникальные зависимости могли быть собраны сборщиком мусора) передайте purge=True:

# Только отмена загрузки: сохраняются метаданные регистрации, можно перезагрузить в любое время
await sdk.module.unload("MyModule")

# Полная выгрузка: удаление метаданных регистрации + очистка sys.modules (только для плагинов из папки)
await sdk.module.unload("MyModule", purge=True)
Семантика unload() по умолчанию unload(purge=True)
Выгрузка экземпляра и ресурсов (события/задачи/маршруты/lifecycle/i18n) ✅ ✅
Сохранение метаданных регистрации (класс модуля и метаинформация) ✅ ❌ Удалено
Очистка sys.modules (только для плагинов из папки) ❌ ✅
Класс модуля может быть собран сборщиком мусора ❌ ✅
Перезагрузка load() доступен напрямую Необходимо сначала register() + load()

При purge=True зависимые модули, которые выгружаются каскадно, также будут удалены; после выгрузки фреймворк выполнит gc.collect() и проверит, можно ли собрать класс модуля/экземпляр, наличие оставшихся ссылок будет предупреждено в логах (с указанием источника, уровень DEBUG).

Жизненный цикл в целом

Вот что делает фреймворк за вас при загрузке и выгрузке модуля:

flowchart TD
    subgraph Load["Загрузка (register → load）"]
        L1["register: Зарегистрировать класс модуля и метаинформацию"] --> L2["Проверка зависимостей<br/>Если отсутствует, пропустить"]
        L2 --> L3["Топологическая сортировка (Kahn + priority）"]
        L3 --> L4["Внедрение owner (current_owner）"]
        L4 --> L5["Создание шаблона конфигурации + регистрация ключей перевода i18n"]
        L5 --> L6["Инициализация модуля (внедрение sdk）"]
        L6 --> L7["Вызов on_load()"]
        L7 --> L8["Присвоение свойства sdk + emit module.load"]
    end

    subgraph Unload["Выгрузка (unload）"]
        U1["Вызов on_unload()"] --> U2["Обеспечение отмены фоновых задач (self.spawn принадлежность）"]
        U2 --> U3["Очистка ключей перевода i18n"]
        U3 --> U4["Удаление маршрутов / команд / обработчиков событий (по owner）"]
        U4 --> U5["Очистка хуков lifecycle (по owner）"]
        U5 --> U6["Удаление свойства sdk + удаление прокси ленивой загрузки"]
        U6 --> U7["emit module.unload"]
    end

    Load --> Unload

Что делает фреймворк при загрузке (вам нужно только написать on_load, остальное делается автоматически):

Этап Что делает фреймворк автоматически
Внедрение owner Во время инициализации модуля с помощью owner_scope — все зарегистрированные вами команды/события/хуки/фоновые задачи автоматически присваиваются к этому модулю, при выгрузке они удаляются по owner
Шаблон конфигурации Модули, объявившие ConfigClass, получают автоматически сгенерированный/заполненный конфигурационный раздел ErisPulse.<ModuleName>
Ключи перевода i18n Модули, объявившие I18nClass, автоматически регистрируют ключи перевода (автоматически удаляются при выгрузке)
Зависимости Сортировка по depends, гарантирует, что зависимые модули загружаются первыми; циклические зависимости отклоняются с ошибкой RuntimeError
SDK присвоение После инициализации присваивается к sdk.<ModuleName>, чтобы вы могли обращаться к sdk.MyModule.xxx

Что делает фреймворк при выгрузке (соответствует U1→U7): После выполнения on_unload происходит дополнительная очистка — фоновые задачи отменяются (созданные через self.spawn, для корректного завершения нужно сделать в on_unload), ключи перевода, маршруты, команды/обработчики событий, хуки lifecycle, затем удаляется свойство SDK. purge=True дополнительно удаляет метаданные регистрации и очищает sys.modules.

Именно эта автоматическая очистка — основа того, что «вам нужно только писать on_load/on_unload, не нужно вручную unregister» — фреймворк реализует "кто зарегистрировал, тот и очищает" как единый процесс.

Объект SDK

Доступ к основным модулям

from ErisPulse import sdk

# Доступ к всем основным модулям через объект sdk
sdk.logger.info("Лог")
sdk.storage.set("key", "value")
config = sdk.config.getConfig("MyModule")

Коммуникация между модулями

# Доступ к другим модулям
other_module = sdk.OtherModule
result = await other_module.some_method()

Методы отправки адаптера

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

Список поддерживаемых методов отправки

# Вывести список всех методов отправки, поддерживаемых платформой
methods = sdk.adapter.list_sends("onebot11")
# Возвращает: ["Text", "Image", "Voice", "Markdown", ...]

Получение подробной информации о методе

# Получить подробную информацию о методе
info = sdk.adapter.send_info("onebot11", "Text")
# Возвращает:
# {
#     "name": "Text",
#     "parameters": [
#         {"name": "text", "type": "str", "default": null, "annotation": "str"}
#     ],
#     "return_type": "Awaitable[Any]",
#     "docstring": "Отправка текстового сообщения..."
# }

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

Объявленная конфигурация (рекомендуется)

Начиная с версии v2.5.2, модуль может объявить класс конфигурации с помощью ConfigClass, используя ту же систему схемы конфигурации, что и адаптер. Конфигурация читается в реальном времени через self.cfg, изменения применяются немедленно:

from dataclasses import dataclass, field
from ErisPulse.Core.Bases import BaseModule
from ErisPulse.Core.Bases import BaseConfig

@dataclass
class MyModuleConfig(BaseConfig):
    api_key: str = field(
        default="",
        metadata={
            "description": {"i18n": "my_module.api_key", "default": "Ключ API"},
            "required": True,
            "secret": True,
            "ui": {"widget": "password", "group": "basic", "order": 1},
        },
    )
    timeout: int = field(
        default=30,
        metadata={
            "description": {"i18n": "my_module.timeout", "default": "Время ожидания (секунды)"},
            "ui": {"widget": "number", "group": "advanced", "order": 2},
        },
    )

class MyModule(BaseModule):
    ConfigClass = MyModuleConfig

    def __init__(self, sdk):
        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):
        pass

    async def do_something(self):
        cfg = self.cfg  # Чтение в реальном времени, тип безопасно
        api_key = cfg.api_key
        timeout = cfg.timeout

BaseConfig — универсальный базовый класс конфигурации, применим для адаптеров, модулей, внешних проектов и любых сценариев. Поля конфигурации поддерживают многоязычные описания i18n (см. документацию по i18n).

Note

Жизненный цикл модуля (включая адаптеры) — загрузка, выгрузка, запуск, остановка и т.д. — управляется фреймворком (модульный менеджер / scope предоставляет API для запроса и управления). ConfigClass должен содержать только бизнес-параметры, модули не должны вкладывать логику "нужно ли запускать себя" внутри.

Система схемы конфигурации также поддерживает (начиная с v2.8.0, см. основные понятия адаптера):

Объявленные ключи перевода (v2.7.0+)

Начиная с v2.7.0, модуль может объявить ключи перевода, как и ConfigClass, с помощью вложенного класса I18nClass. Фреймворк автоматически регистрирует все объявленные ключи перевода при загрузке, без необходимости вызова i18n.register(), и регистрация происходит до генерации шаблона конфигурации, гарантируя, что ключи перевода, используемые в описании конфигурации, уже доступны.

from ErisPulse.Core.Bases import BaseConfig, BaseI18n, I18nKey

class MyModule(BaseModule):
    # Класс конфигурации (необязательно)
    @dataclass
    class ConfigClass(BaseConfig):
        welcome_msg: str = field(
            default="Добро пожаловать",
            metadata={
                "description": {"i18n": "mymodule.welcome_msg", "default": "Сообщение приветствия"},
            },
        )

    # Класс набора ключей перевода (необязательно)
    class I18nClass(BaseI18n):
        # Имя атрибута автоматически объединяется в полный ключ пути: <имя модуля>.<имя атрибута>
        welcome_msg: I18nKey = I18nKey(
            default="Welcome Message",   # Безопасный запасной текст без привязки к языку
            zh_CN="欢迎消息",
            zh_TW="歡迎訊息",
            en="Welcome Message",
            ja="ウェルカムメッセージ",
            ru="Приветственное сообщение",
        )
        hello: I18nKey = I18nKey(
            default="Hello, {name}!",
            zh_CN="你好，{name}！",
            zh_TW="你好，{name}！",
            en="Hello, {name}!",
            ja="こんにちは、{name}！",
            ru="Привет, {name}!",
        )

Подробнее см. Рекомендуемый способ написания i18n.

Ручное чтение конфигурации (устарело)

Устарело: используйте объявленную конфигурацию + чтение self.cfg в реальном времени.

class MyModule(BaseModule):
    def __init__(self, sdk):
        self.sdk = sdk

    def _load_config(self):
        config = self.sdk.config.getConfig("MyModule")
        if not config:
            self.sdk.config.setConfig("MyModule", {"api_key": "", "timeout": 30})
            return {"api_key": "", "timeout": 30}
        return config

Система хранения

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

# Сохранить данные
sdk.storage.set("user:123", {"name": "张三"})

# Получить данные
user = sdk.storage.get("user:123", {})

# Удалить данные
sdk.storage.delete("user:123")

Использование транзакций

# Использовать транзакцию для обеспечения согласованности данных
with sdk.storage.transaction():
    sdk.storage.set("key1", "value1")
    sdk.storage.set("key2", "value2")
    # Если любая операция завершится с ошибкой, все изменения будут отменены

Обработка событий

Регистрация обработчиков событий

from ErisPulse.Core.Event import command, message

# Регистрация команды
@command("info", help="Получить информацию")
async def info_handler(event):
    await event.reply("Это информация")

# Регистрация обработчика сообщений
@message.on_group_message()
async def group_handler(event):
    sdk.logger.info(f"Получено групповое сообщение: {event.get_text()}")

Жизненный цикл обработчиков событий

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

Механизм ленивой загрузки

Принцип работы

# Модуль инициализируется только при первом обращении
result = await sdk.my_module.some_method()
# ↑ Здесь будет инициирована инициализация модуля

Немедленная загрузка

Для модулей, требующих немедленной инициализации (например, триггеры, таймеры):

@staticmethod
def get_load_strategy():
    return ModuleLoadStrategy(
        lazy_load=False,  # Немедленная загрузка
        priority=100
    )

Обработка ошибок

Перехват исключений

async def handle_event(self, event):
    try:
        # Бизнес-логика
        await self.process_event(event)
    except ValueError as e:
        self.logger.warning(f"Ошибка параметра: {e}")
        await event.reply(f"Ошибка параметра: {e}")
    except Exception as e:
        self.logger.error(f"Обработка не удалась: {e}")
        raise

Запись в лог

# Использование разных уровней логирования
self.logger.debug("Отладочная информация")    # Подробная отладочная информация
self.logger.info("Состояние работы")      # Информация о нормальной работе
self.logger.warning("Предупреждение")  # Предупреждение
self.logger.error("Ошибка")    # Ошибка
self.logger.critical("Критическая ошибка") # Критическая ошибка

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

Продолжение пути обучения

Глубокое изучение и отладка

Event 包装类

来源: docs/ru/developer-guide/modules/event-wrapper.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ru/developer-guide/modules/event-wrapper.md）

Подробное руководство по классу-обёртке Event

Note

Обязательно к прочтению, ядро L1 — один из пяти ключевых документов в «пути чтения» на главной странице, необходим для разработки модулей.

Параметр event в обработчике может напрямую вызывать .get_text() / .reply() / .wait_reply() именно благодаря классу-обёртке Event. В этой статье приведён полный список его методов — обращайтесь к ней по мере необходимости при написании обработчиков событий; сигнатуры методов и их поведение определяются именно здесь.

Добавление аннотации типа к параметру event

Параметр event обработчика событий — это класс-обёртка Event (подкласс dict). Настоятельно рекомендуется добавить для него аннотацию типа:

from ErisPulse.Core.Event import Event

@message.on_private_message()
async def handler(event: Event):
    text = event.get_text()   # IDE автоматически дополнит все удобные методы
    await event.reply(text)   # Опечатки будут обнаружены уже при статической проверке

Без аннотации IDE не сможет распознать методы Event (подсказки для get_text() / reply() / wait_reply() / методов расширений платформы не появятся), и их придётся писать по памяти.

Обратите внимание на различие: event в колбэке обработчика событий — это класс-обёртка Event (аннотируется как Event); а event в методах жизненного цикла модуля on_load / on_unload — это обычный dict (аннотируется как dict). Не путайте их.

Ключевые особенности

Методы основных полей

from ErisPulse.Core.Event import command

@command("info")
async def info_command(event: Event):
    event_id = event.get_id()
    platform = event.get_platform()
    time = event.get_time()
    print(f"ID: {event_id}, платформа: {platform}, время: {time}")

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

from ErisPulse.Core.Event import message

@message.on_private_message()
async def private_handler(event: Event):
    text = event.get_text()
    user_id = event.get_user_id()
    nickname = event.get_user_nickname()
    await event.reply(f"Привет, {nickname}!")

Определение типа сообщения

from ErisPulse.Core.Event import message

@message.on_group_message()
async def group_handler(event: Event):
    is_private = event.is_private_message()
    is_group = event.is_group_message()
    is_at = event.is_at_message()
    await event.reply(f"Тип: {'личный чат' if is_private else 'групповой чат'}")

Функции ответа

from ErisPulse.Core.Event import command

@command("ask")
async def ask_command(event: Event):
    await event.reply("Введите своё имя:")
    reply = await event.wait_reply(timeout=30)
    if reply:
        name = reply.get_text()
        await event.reply(f"Привет, {name}!")

@command("price")
async def price_command(event: Event):
    await event.reply("Введите сумму (например: 5 руб.):")
    # Ответ должен соответствовать регулярному выражению, иначе ожидание продолжится до таймаута
    reply = await event.wait_reply(timeout=30, regex=r"\d+\s*руб")
    if reply:
        await event.reply(f"Получена сумма: {reply.get_text()}")

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

Note

Возможности этого раздела требуют ErisPulse 2.8.0+.

# Напоминание по таймеру в сессии: напоминание через 5 минут без ответа, автоматически отменяется после ответа пользователя
reminder = event.remind(300, "Вы ещё здесь? Если не хотите общаться, ответьте «выход»")
reminder.cancel()  # также можно отменить вручную

# Эскалация по таймауту: гарантированное срабатывание по истечении срока (не отменяется ответом), например уведомление владельца при долгой необработке
event.escalate(1800, lambda e: notify_master("Время обработки заявки истекло"))

# Многоканальное ожидание: одновременное ожидание «согласен» и «отказ»; побеждает первый пришедший ответ
which, reply = await event.select(
    event.expect(pattern="согласен*", user="10001"),
    event.expect(pattern="отказ*", user="10002"),
    timeout=60,
)
if which is None:
    await event.reply("Время ожидания подтверждения истекло")

# Ожидание на уровне сессии: ответ любого участника группы может сработать (групповое взаимодействие)
reply = await event.wait_reply(session=True, prompt="Кто-нибудь может помочь с ответом?")

# Входящие сессии: последние 20 сообщений текущей сессии (включая бота; основа для контекста ИИ / защиты от повторов)
messages = await event.history(20)

# Транзакция сообщений: при исключении автоматически отзываются сообщения, отправленные в рамках транзакции
async with event.message_tx():
    await event.reply("Обработка, пожалуйста, подождите...")
    result = await do_something()
    await event.reply(f"Готово: {result}")

Получение информации о команде

from ErisPulse.Core.Event import command

@command("cmdinfo")
async def cmdinfo_command(event: Event):
    cmd_name = event.get_command_name()
    cmd_args = event.get_command_args()
    await event.reply(f"Команда: {cmd_name}, аргументы: {cmd_args}")

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

from ErisPulse.Core.Event import notice

@notice.on_friend_add()
async def friend_add_handler(event: Event):
    await event.reply("Спасибо, что добавили меня в друзья!")

Краткий справочник методов

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

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

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

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

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

Содержимое сообщения

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

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

Связанное с @-сообщениями

Определение типа сообщения

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

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

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

Определение типа уведомления

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

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

Определение типа запроса

Функции ответа

Базовый ответ

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

Функция пересылки

Примечание: функция пересылки должна быть реализована через Send DSL адаптера; сам класс-обёртка Event не предоставляет прямых методов пересылки.

# Пересылка сообщения в группу
adapter = sdk.adapter.get(event.get_platform())
target_id = event.get_group_id()  # или укажите другой ID группы
await adapter.Send.To("group", target_id).Text(event.get_text())

Функции ожидания ответа

Интерактивные методы

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

confirm() - диалог подтверждения:

@command("delete", help="удаление данных")
async def delete_handler(event: Event):
    if await event.confirm("Вы уверены, что хотите удалить все данные?"):
        sdk.storage.delete("all_data")
        await event.reply("Данные удалены")
    else:
        await event.reply("Отменено")

confirm() - с подсказкой:

# hint=True добавит "(да/нет)" в конец подсказки
if await event.confirm("Продолжить?", hint=True):
    await event.reply("Продолжено")
# Пользователь видит: Продолжить? (да/нет)

choose() - меню выбора:

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

choose() - форматирование вариантов и объединение сообщений:

# формат inline: варианты отображаются в одной строке
choice = await event.choose("Выберите:", ["A", "B", "C"], options_format="inline")
# вывод: 1.A | 2.B | 3.C

# пользовательский формат
choice = await event.choose("Выберите:", ["кот", "пёс"],
    options_format=lambda opts: " / ".join(opts))
# вывод: кот / пёс

# options_format="auto" (по умолчанию): встроенный стиль выбирается автоматически в зависимости от method
# Markdown → маркированный список
choice = await event.choose(
    "## Выберите", ["кот", "пёс"],
    method="Markdown",  # auto автоматически распознаётся как md-список
)
# вывод:
# ## Выберите
# - 1. кот
# - 2. пёс

# Html → нумерованный список
choice = await event.choose(
    "<h2>Выберите</h2>", ["кот", "пёс"],
    method="Html", merge_prompt=True,  # auto автоматически распознаётся как html-список
)
# вывод:
# <h2>Выберите</h2>
# <ol><li>1. кот</li><li>2. пёс</li></ol>

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

# пользовательский заполнитель
choice = await event.choose(
    "Выберите: [choices]",
    ["кот", "пёс"],
    placeholder="[choices]",
)

collect() - сбор данных формы:

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

reply с методами, отличными от Text:

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

from ErisPulse.Core.Event import MessageBuilder
segments = MessageBuilder.text("Взгляните на это изображение:").image("http://example.com/img.jpg").build()
await event.reply_ob12(segments)

Полную информацию об использовании многоходового диалога Conversation см. в разделе Многоходовый диалог Conversation.

Информация о команде

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

Необработанные данные

Методы расширений платформы

Адаптеры могут регистрировать специфичные для платформы методы у класса-обёртки Event. Методы доступны только на экземплярах Event соответствующей платформы; при обращении с других платформ выбрасывается AttributeError.

Методы платформы через Event.__getattribute__ имеют приоритет над встроенными методами, поэтому можно переопределять встроенные интерактивные методы, такие как confirm, choose, collect, wait_reply, предоставляя специфичные для платформы реализации (например, кнопки, карточки и т.д.). Встроенные реализации экспортируются в виде функций _builtin_* и доступны для вызова из переопределяющих методов.

# Событие электронной почты - только почтовые методы
event = Event({"platform": "email", "email_raw": {"subject": "Hello"}})
event.get_subject()      # ✅ возвращает "Hello"
event.get_chat_type()    # ❌ AttributeError

# Событие Telegram - только методы Telegram
event = Event({"platform": "telegram", "telegram_raw": {"chat": {"type": "private"}}})
event.get_chat_type()    # ✅ возвращает "private"
event.get_subject()      # ❌ AttributeError

# Встроенные методы доступны всегда
event.get_text()         # ✅ на любой платформе
event.reply("hi")        # ✅ на любой платформе

Запрос зарегистрированных методов

from ErisPulse.Core.Event import get_platform_event_methods

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

Поддержка hasattr и dir

hasattr(event, "get_subject")   # возвращает True, только если platform="email"
"get_subject" in dir(event)     # аналогично

Кроссплатформенные расширения (подстановочный знак)

register_event_method и register_event_mixin поддерживают передачу "*" в качестве имени платформы; зарегистрированные таким образом методы доступны на экземплярах Event всех платформ. Подходит для функций, требующих кроссплатформенного переиспользования, таких как диалоги с ИИ, управление контекстом и т.д.

from ErisPulse.Core.Event.wrapper import register_event_method

@register_event_method("*")
async def ai_chat(self, prompt: str):
    # self — это экземпляр Event, с доступом к данным события и встроенным методам
    await self.reply(f"AI: {prompt}")

После регистрации обработчики событий любой платформы могут вызывать event.ai_chat(...).

Приоритет разрешения методов (от высшего к низшему): специфичные для платформы методы → методы с подстановочным знаком → встроенные методы → доступ по ключам словаря.

О том, как разработчики адаптеров регистрируют методы расширений, см. API системы событий — Кроссплатформенные расширения.

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

Продолжение пути разработки модулей

Углубление интерактивных возможностей

模块最佳实践

来源: docs/ru/developer-guide/modules/best-practices.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ru/developer-guide/modules/best-practices.md）

Лучшие практики разработки модулей

Tip

Когда читать эту статью: рекомендуется прочитать после написания первого модуля (инженерные лучшие практики). Если не нужно, пропустите — это не повлияет на использование других функций.

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

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

1. Принцип единственной ответственности

Каждый модуль должен отвечать только за одну основную функцию:

# Хорошее проектирование: каждый модуль отвечает за одну функцию
class WeatherModule(BaseModule):
    """Модуль для запроса погоды"""
    pass

class NewsModule(BaseModule):
    """Модуль для запроса новостей"""
    pass

# Плохое проектирование: один модуль отвечает за несколько несвязанных функций
class UtilityModule(BaseModule):
    """Содержит погоду, новости, шутки и другие функции"""
    pass

2. Нормы именования модулей

[project]
name = "ErisPulse-ModuleName"  # Использование префикса ErisPulse-

3. Четкое управление конфигурацией

Рекомендуется использовать декларативную конфигурацию (ConfigClass + BaseConfig), чтобы получить типовую безопасность, автоматическое создание шаблонов, поддержку веб-интерфейса и т.д.:

from dataclasses import dataclass, field
from ErisPulse.Core.Bases import BaseConfig

@dataclass
class MyModuleConfig(BaseConfig):
    api_url: str = field(default="https://api.example.com", metadata={
        "description": {"i18n": "my_module.api_url", "default": "Адрес API"},
    })
    timeout: int = field(default=30, metadata={
        "description": {"i18n": "my_module.timeout", "default": "Время ожидания (секунды)"},
    })
    cache_ttl: int = field(default=3600, metadata={
        "description": {"i18n": "my_module.cache_ttl", "default": "Время жизни кэша (секунды)"},
    })

class MyModule(BaseModule):
    ConfigClass = MyModuleConfig

    async def do_something(self):
        cfg = self.cfg  # Типовая безопасность, получение в реальном времени
        await self._fetch(cfg.api_url, timeout=cfg.timeout)

Также можно продолжать использовать ручной способ чтения и записи конфигурации (см. Основные концепции модуля).

Декларативные ключи перевода (v2.7.0+)

Модуль может объявить ключи перевода с помощью I18nClass, и фреймворк автоматически зарегистрирует их в системе перевода, без необходимости вызывать i18n.register() вручную.

from ErisPulse.Core.Bases import BaseI18n, I18nKey

class MyModule(BaseModule):
    class I18nClass(BaseI18n):
        # Ключ перевода с плейсхолдером
        welcome: I18nKey = I18nKey(
            default="Welcome, {name}!",
            zh_CN="Добро пожаловать, {name}!",
            zh_TW="歡迎你，{name}！",
            en="Welcome, {name}!",
            ja="ようこそ、{name}！",
            ru="Добро пожаловать, {name}!",
        )
        # Описание поля конфигурации
        api_url: I18nKey = I18nKey(
            default="API URL",
            zh_CN="Адрес API",
            zh_TW="API 位址",
            en="API URL",
            ja="API URL",
            ru="API URL",
        )

Подробное использование смотрите в документации по i18n.

Асинхронное программирование

1. Использование асинхронных библиотек

# Рекомендуется использовать встроенный HTTP-клиент SDK (асинхронный, с автоматической логированием и статистикой)
from ErisPulse.Core import client

class MyModule(BaseModule):
    async def fetch_data(self, url):
        resp = await client.get(url)
        return await resp.json()

# Также можно использовать sdk.client (результат тот же)
from ErisPulse import sdk

class MyModule(BaseModule):
    async def fetch_data(self, url):
        resp = await sdk.client.get(url)
        return await resp.json()

# Не используйте прямой импорт aiohttp (трудно управлять в рамках фреймворка)
import aiohttp

class MyModule(BaseModule):
    async def fetch_data(self, url):
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.json()

# Не используйте requests (синхронный, блокирует цикл событий)
import requests

class MyModule(BaseModule):
    def fetch_data(self, url):
        return requests.get(url).json()  # Блокирует цикл событий

2. Правильное выполнение асинхронных операций

from ErisPulse.Core.Event import Event  # 注解 event: Event получает поддержку IDE

async def handle_command(self, event: Event):
    # Операции, требующие ожидания результата: просто await (жизненный цикл ясен)
    result = await self._long_operation()

async def on_load(self, event: dict):
    # Фоновые задачи (опрос/таймер/fire-and-forget): используйте self.spawn(),
    # при отключении модуля фреймворк в on_unload гарантирует отмену, чтобы избежать утечек
    self.spawn(self._poll())

Note

Рекомендуется использовать self.spawn() (ErisPulse 2.8.0+). С версии 2.8.3 даже asyncio.create_task будет автоматически привязан к модулю (Task Factory автоматически регистрирует, отменяет при отключении, не утечка self); self.spawn() по-прежнему рекомендуется — поддерживает переключение потоков, явное указание owner=. До версии 2.8.3 фоновые задачи не привязаны, удерживают ссылку на self, что приводит к утечке при горячей перезагрузке; необходимо использовать self.spawn(). Подробнее смотрите в управление жизненным циклом.

3. Управление ресурсами

async def on_load(self, event):
    # Клиент SDK уже управляет пулы соединений, не нужно создавать сессию вручную
    pass
    
async def on_unload(self, event):
    # Если нужно использовать кастомный клиент, не забудьте освободить ресурсы
    pass

Обработка событий

1. Использование обёртки Event

# Использование удобных методов обёртки Event
@command("info")
async def info_command(event: Event):
    user_id = event.get_user_id()
    nickname = event.get_user_nickname()
    await event.reply(f"Привет, {nickname}!")

# А не прямой доступ к словарю
@command("info")
async def info_command(event: Event):
    user_id = event["user_id"]  # Менее понятно, легко ошибиться

2. Разумное использование ленивой загрузки

# Модуль с редко используемыми командами: объявите триггер activate_on, модуль автоматически активируется при первом совпадении команды (поддерживает ленивую загрузку)
class CommandModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(lazy_load=True, activate_on=[
            {"command": {"name": "dice", "help": "Бросить кубик", "aliases": ["d"]}},
        ])

# Модуль с редкими триггерами: объявите триггер события, модуль активируется при наступлении события
class ListenerModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(lazy_load=True, activate_on=[
            {"notice": "group_member_increase"},
        ])

# Модуль, который должен обрабатывать каждое событие или быть готовым при запуске: загружается немедленно
class HotListenerModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(lazy_load=False)

# Утилитарные модули подходят для ленивой загрузки
class UtilityModule(BaseModule):
    @staticmethod
    def get_load_strategy():
        return ModuleLoadStrategy(lazy_load=True)

Полный синтаксис activate_on (три формы события / сокращённая и dict-декларация команды / цепочка fallback help) смотрите в системе ленивой загрузки модулей.

3. Регистрация обработчиков событий

async def on_load(self, event):
    # Регистрируйте обработчики событий в on_load
    @command("hello")
    async def hello_handler(event: Event):
        await event.reply("Привет!")
    
    @message.on_group_message()
    async def group_handler(event: Event):
        self.logger.info("Получено сообщение в группе")
    
    # Не нужно отменять регистрацию вручную, фреймворк сделает это автоматически

Утилитарные модули: при управлении чужими объектами нужно обрабатывать "уведомления об отключении"

Когда это нужно: ваш модуль управляет объектами других модулей (периодические обратные вызовы, подписчики, соединения, кэш и т.д.). Если эти ссылки не будут удалены после отключения модуля, экземпляр модуля не может быть освобождён — это наиболее распространённая причина утечек памяти в утилитарных модулях.

from ErisPulse.Core.Bases import BaseModule
from ErisPulse.runtime import off_cleanup, on_cleanup

class MyToolModule(BaseModule):
    def __init__(self):
        self._entries = {}  # {имя модуля: управляемый объект}

    def register(self, entry):
        owner = on_cleanup(self._drop)   # ① При регистрации добавить в цепочку очистки, автоматически определить вызывающий модуль
        self._entries.setdefault(owner, []).append(entry)

    def _drop(self, owner: str):
        self._entries.pop(owner, None)   # ② При отключении модуля фреймворк автоматически вызывает: удалить его объекты

    async def on_unload(self, event):
        off_cleanup(self._drop)          # ③ Перед отключением модуля отменить хук

Вот и всё, фреймворк гарантирует:

Последствия отсутствия подключения: при полном отключении модуля экземпляр не может быть освобождён (диагностика утечек сообщает "неосвобождаемый"); если вызывающий модуль сам не отменяет уведомление в on_unload, утечка будет постоянной.

Обычным модулям (не управляющим чужими объектами) не нужно об этом беспокоиться — очистка ресурсов (команды / обработчики / маршрутизация / фоновые задачи и т.д.) автоматически управляется фреймворком.

Подробности о времени срабатывания, правилах определения вызывающей стороны, таймаутах и обработке ошибок смотрите в системе принадлежности · руководство для утилитарных модулей.

Обработка ошибок

1. Классификация обработки исключений

from ErisPulse.Core.Bases.errors import ClientError

async def handle_event(self, event: Event):
    try:
        result = await self._process(event)
    except ValueError as e:
        # Ожидаемые бизнес-ошибки
        self.logger.warning(f"Предупреждение: {e}")
        await event.reply(f"Ошибка параметра: {e}")
    except ClientError as e:
        # Ошибки сети (нижележащие aiohttp-исключения sdk.client уже преобразованы)
        self.logger.error(f"Ошибка сети {e.method} {e.url}: {e}")
        await event.reply("Сетевой запрос не удался, попробуйте позже")
    except Exception as e:
        # Неожиданные ошибки
        self.logger.error(f"Неизвестная ошибка: {e}", exc_info=True)
        await event.reply("Обработка не удалась, свяжитесь с администратором")
        raise

2. Обработка таймаутов

# Рекомендуется использовать встроенный клиент SDK (с таймаутом и повторами)
from ErisPulse.Core import client
from ErisPulse.Core.Bases.errors import ClientTimeoutError

async def fetch_with_timeout(self, url, timeout=30):
    try:
        resp = await client.get(url, timeout=timeout)
        return await resp.json()
    except ClientTimeoutError:
        self.logger.warning(f"Таймаут запроса: {url}")
        raise

Система хранения данных

1. Использование транзакций

# Используйте транзакции для обеспечения согласованности данных
async def update_user(self, user_id, data):
    with self.sdk.storage.transaction():
        self.sdk.storage.set(f"user:{user_id}:profile", data["profile"])
        self.sdk.storage.set(f"user:{user_id}:settings", data["settings"])

# ❌ Без транзакций возможна несогласованность данных
async def update_user(self, user_id, data):
    self.sdk.storage.set(f"user:{user_id}:profile", data["profile"])
    # Если здесь произойдёт ошибка, предыдущая запись не откатится
    self.sdk.storage.set(f"user:{user_id}:settings", data["settings"])

2. Массовые операции

# Используйте массовые операции для повышения производительности
def cache_multiple_items(self, items):
    self.sdk.storage.set_multi({
        f"item:{k}": v for k, v in items.items()
    })

# ❌ Многократные вызовы менее эффективны
def cache_multiple_items(self, items):
    for k, v in items.items():
        self.sdk.storage.set(f"item:{k}", v)

Логирование

1. Разумное использование уровней логирования

# DEBUG: Подробная отладочная информация (только для разработки)
self.logger.debug(f"Входные параметры: {params}")

# INFO: Информация о нормальной работе
self.logger.info("Модуль загружен")
self.logger.info(f"Обработка запроса: {request_id}")

# WARNING: Предупреждения, не влияющие на основную функциональность
self.logger.warning(f"Параметр {key} не установлен, используется значение по умолчанию")
self.logger.warning("API отвечает медленно, возможно, требуется оптимизация")

# ERROR: Ошибки
self.logger.error(f"Запрос API не удался: {e}")
self.logger.error(f"Ошибка обработки события: {e}", exc_info=True)

# CRITICAL: Критические ошибки, требующие немедленного вмешательства
self.logger.critical("Не удалось подключиться к базе данных, бот не может нормально работать")

2. Структурированное логирование

# Используйте структурированное логирование для удобства анализа
self.logger.info(f"Обработка запроса: request_id={request_id}, user_id={user_id}, duration={duration}ms")

# ❌ Использование неструктурированного логирования
self.logger.info(f"Обработка запроса, от пользователя {user_id}, затрачено {duration} миллисекунд")

Оптимизация производительности

1. Использование кэша

class MyModule(BaseModule):
    def __init__(self):
        self._cache = {}
        self._cache_lock = asyncio.Lock()
    
    async def get_data(self, key):
        async with self._cache_lock:
            if key in self._cache:
                return self._cache[key]
            
            # Получение данных из базы
            data = await self._fetch_from_db(key)
            
            # Кэширование данных
            self._cache[key] = data
            return data

2. Избегание блокирующих операций

# Используйте асинхронные операции
async def process_message(self, event: Event):
    # Асинхронная обработка
    await self._async_process(event)

# ❌ Блокирующая операция
async def process_message(self, event: Event):
    # Синхронная операция, блокирует цикл событий
    result = self._sync_process(event)

Безопасность

1. Защита конфиденциальных данных

# Конфиденциальные данные хранятся в конфигурации (декларативный ConfigClass, secret-поля не попадают в логи/экспорт)
from dataclasses import dataclass, field
from ErisPulse.Core.Bases import BaseModule, BaseConfig

@dataclass
class MyModuleConfig(BaseConfig):
    api_key: str = field(
        default="",
        metadata={"description": "API-ключ", "secret": True},
    )

class MyModule(BaseModule):
    ConfigClass = MyModuleConfig

    def check_api_key(self):
        if not self.cfg.api_key or self.cfg.api_key == "YOUR_API_KEY_HERE":
            raise ValueError("Пожалуйста, настройте действительный API-ключ в config.toml")

# ❌ Конфиденциальные данные в коде
class MyModule(BaseModule):
    API_KEY = "sk-1234567890"  # Не делайте так!

2. Проверка входных данных

# Проверка пользовательского ввода
async def process_command(self, event: Event):
    user_input = event.get_text()
    
    # Проверка длины ввода
    if len(user_input) > 1000:
        await event.reply("Слишком длинный ввод, пожалуйста, повторите")
        return
    
    # Проверка формата ввода
    if not re.match(r'^[a-zA-Z0-9]+$', user_input):
        await event.reply("Неверный формат ввода")
        return

Тестирование

1. Юнит-тесты

import pytest
from ErisPulse.Core.Bases import BaseModule

class TestMyModule:
    def test_config_defaults(self):
        """Тестирование значений по умолчанию конфигурации"""
        config = MyModule.ConfigClass()
        assert config.timeout == 30

2. Интеграционные тесты

@pytest.mark.asyncio
async def test_command_handling():
    """Тестирование обработки команд"""
    module = MyModule()
    await module.on_load({})
    
    # Симуляция командного события
    event = create_test_command_event("hello")
    await module.handle_command(event)

Развертывание

1. Управление версиями

[project]
name = "ErisPulse-MyModule"
version = "1.0.0"

Следуйте семантическому версионированию:

2. Заголовок README

README, созданный с помощью epsdk create, уже содержит логотип и шапку ErisPulse. Два рекомендуемых подхода:

Способ A — Только логотип ErisPulse (по умолчанию):

<div align="center">

<img src="https://raw.githubusercontent.com/ErisPulse/ErisPulse/main/.github/assets/ErisPulseLogo.png" width="180" alt="MyModule" />

# MyModule

**Одно предложение описания**

<p>
  <a href="https://pypi.org/project/ErisPulse-MyModule/"><img src="https://img.shields.io/pypi/v/ErisPulse-MyModule?style=for-the-badge&logo=pypi&logoColor=white" alt="PyPI"></a>
  <a href="https://pypi.org/project/ErisPulse-MyModule/"><img src="https://img.shields.io/badge/Python-3.10+-FFD43B?style=for-the-badge&logo=python&logoColor=blue" alt="Python"></a>
  <a href="./LICENSE"><img src="https://img.shields.io/badge/License-MIT-blue?style=for-the-badge" alt="License"></a>
  <a href="https://github.com/ErisPulse/ErisPulse"><img src="https://img.shields.io/badge/Powered_by-ErisPulse-FF6B9D?style=for-the-badge&logo=bookstack&logoColor=white" alt="ErisPulse"></a>
</p>

</div>

Способ B — Иконка модуля × Логотип ErisPulse (если есть пользовательская иконка):

<div align="center">

<img src=".github/assets/MyModuleIcon.svg" width="120" alt="MyModule" />
<span style="font-size:44px;color:#c8c8c8;margin:0 18px;vertical-align:middle;">×</span>
<img src="https://raw.githubusercontent.com/ErisPulse/ErisPulse/main/.github/assets/ErisPulseLogo.png" height="120" alt="ErisPulse" />

# MyModule
(Строка с шаблонами та же)
</div>

Можно дополнительно добавить шаблоны GitHub Stars, Downloads и т.д. Логотип также можно загрузить в проект (.github/assets/ErisPulseLogo.png) и использовать относительный путь.

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

Возвращение к учебному пути

Глубокое изучение по сценариям

ORM 数据模型

来源: docs/ru/developer-guide/orm.md（https://github.com/ErisPulse/ErisPulse/blob/main/docs/ru/developer-guide/orm.md）

Слой моделей данных (ORM)

Tip

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

Начиная с версии 2.9.0 во фреймворк встроен декларативный слой моделей данных: унаследуйте Model и объявите поля через Field, чтобы получить автоматическое создание таблиц и возможности CRUD. Модели построены непосредственно поверх встроенного слоя хранения — SQLite / MySQL / PostgreSQL определяется конфигурацией ErisPulse.storage.backend, при смене бэкенда не нужно изменять код моделей.

Объявление моделей

from ErisPulse.Core.Bases import Model, Field

class User(Model):
    id: int = Field(primary_key=True, autoincrement=True)
    name: str = Field(max_length=64)
    age: int = Field(default=0, ge=0, le=150)
    role: str = Field(default="user", choices=["user", "admin"])
    tags: list = Field(default_factory=list)          # JSON-столбец, автоматическая сериализация при чтении и записи
    bio: str = Field(default="", description={"i18n": "user.bio", "default": "Описание"})

Параметры полей

Параметр Описание
default Значение по умолчанию (не задано и не автоинкремент → обязательное NOT NULL)
default_factory Фабрика изменяемого значения по умолчанию (например, list)
primary_key Первичный ключ
autoincrement Автоинкрементный первичный ключ (подразумевает первичный ключ, автоматически заполняется после create)
max_length Максимальная длина строки (генерирует VARCHAR(n), проверка при записи)
nullable Допускается ли NULL (по умолчанию False)
index Создаёт обычный индекс
unique Ограничение уникальности
choices / ge / le Перечисление / числовой диапазон (проверка при записи)
description Описание (i18n-словарь, тот же формат, что у конфигурационных классов)
column_type Переопределяет определение типа SQL-столбца

Создание таблиц и CRUD

await User.create_table()                      # Идемпотентно (IF NOT EXISTS)

user = await User.create(name="Alice", age=20) # Вставка (автоинкрементный первичный ключ заполняется автоматически)
got = await User.get(id=user.id)               # Первая запись по условию равенства

users = await User.where(User.age > 18).order_by("-age").limit(10).all()
first = await User.where(User.name == "Alice").first()
total = await User.count(User.age > 18)

user.age = 21
await user.save()                              # Обновление по первичному ключу (сначала проверка ограничений); при отсутствии первичного ключа выполняется вставка (автоинкрементный первичный ключ также заполняется в экземпляре)
await user.delete()                            # Удаление по первичному ключу

await User.update_all(User.age > 18, role="adult")  # Массовое обновление
await User.delete_all(User.age > 100)               # Массовое удаление

Выражения запросов поддерживают > >= < <= == !=, in_([...]), а также комбинации & (И) / | (ИЛИ):

await User.where((User.age > 18) & User.name.in_(["Alice", "Bob"])).all()

Транзакции

Чтение и запись ORM используют тот же маршрутизатор транзакций, что и слой хранения фреймворка: когда вы находитесь внутри контекстной транзакции storage.atransaction(), операции create / save / delete и запросы автоматически переиспользуют соединение транзакции; фиксация или откат выполняются единым образом вместе с транзакцией, отдельной фиксации не происходит.

async with storage.atransaction():
    await User.create(name="Alice")
    ...  # при исключении внутри блока INSERT выше также откатывается

Связь с декларативными конфигурационными классами

Объявления моделей разделяют одну и ту же основу с ConfigClass (@dataclass + field(metadata=...)) — набор ограничений, движок валидации (validate_field_constraints), реестр категорий типов (python_type_category); но базовые классы намеренно разделены: поля конфигурации — обычные значения (TOML round-trip, горячее обновление после однократной загрузки), поля моделей — дескрипторы столбцов (обращение к классу = выражение запроса, экземпляр на каждую строку). Один синтаксис объявления, две основы, каждая выполняющая свою роль.

Когда какой способ объявления использовать

Аспект Конфигурационный класс field(metadata=...) Поле модели Field()
Сценарий применения Параметры поведения модуля (немного, читаемо человеком, требуется горячее обновление) Записи бизнес-данных (много строк, программная запись/чтение, требуются запросы)
Форма хранения config.toml (с сохранением комментариев, TOML round-trip) Таблица базы данных (автоматическое создание таблиц, SQL-диалект)
Форма значения Обычное значение (прямой доступ к атрибуту dataclass) Дескриптор (обращение к классу = выражение столбца, обращение к экземпляру = значение строки)
Объявление ограничений metadata={"choices": ..., "min": ..., "max": ...} Field(choices=..., ge=..., le=..., max_length=...)
Общая основа Движок валидации + набор ограничений + реестр категорий типов (одни и те же) То же самое

Эмпирическое правило: «как работает модуль» — конфигурационный класс, «какие данные создал пользователь» — модель.

Ограничения и важные замечания

Автоматическая миграция (этап 2)

create_table() при уже существующей таблице автоматически сравнивает существующие столбцы с полями модели: для новых полей автоматически выполняется ALTER TABLE ADD COLUMN (мигрируемые столбцы освобождаются от ограничения NOT NULL, существующие строки заполняются NULL), для новых полей с index=True синхронно создаётся индекс. Писать миграционные скрипты вручную не нужно.

# Эволюция модели после релиза v1: добавление полей email / bio
class User(Model):
    __tablename__ = "orm_users"

    id: int = Field(primary_key=True, autoincrement=True)
    name: str = Field(max_length=64)
    age: int = Field(default=0)
    email: str = Field(default="")     # Новое: при следующем create_table() автоматически выполнится ADD COLUMN
    bio: str = Field(default="")

await User.create_table()              # Идемпотентно: мигрируются только новые столбцы

Ограничения: поддерживается только добавление столбцов; изменение первичного ключа, изменение типа столбца и удаление столбцов требуют ручной обработки (во избежание случайных разрушительных ALTER-операций).

Внешние ключи (основа маппинга связей)

foreign_key="таблица.столбец" объявляет ограничение внешнего ключа на уровне столбца; в DDL генерируется клауза REFERENCES:

class Post(Model):
    __tablename__ = "posts"

    id: int = Field(primary_key=True, autoincrement=True)
    author: int = Field(foreign_key="orm_users.id")

    content: str = Field(max_length=255)

Маппинг связей (relationship)

Объявите relationship() как атрибут класса в теле класса модели; направление определяется автоматически в зависимости от того, в какой таблице объявлен столбец внешнего ключа:

from ErisPulse.Core.Bases import Model, Field, relationship

class User(Model):
    __tablename__ = "orm_users"

    id: int = Field(primary_key=True, autoincrement=True)
    name: str = Field(max_length=64)

    posts = relationship("Post", foreign_key="author")   # внешний ключ в таблице другой модели → has-many

class Post(Model):
    __tablename__ = "posts"

    id: int = Field(primary_key=True, autoincrement=True)
    author: int = Field(foreign_key="orm_users.id")
    content: str = Field(max_length=255)

    writer = relationship("User", foreign_key="author")  # внешний ключ в этой таблице → belongs-to

has-many: атрибут экземпляра возвращает набор запросов, доступны все цепочечные возможности QuerySet, create автоматически подставляет первичный ключ текущей таблицы в столбец внешнего ключа связанной модели:

alice = await User.get(name="Alice")

posts = await alice.posts.all()                          # все статьи этого пользователя
latest = await alice.posts.order_by("-id").first()
total = await alice.posts.count()
hot = await alice.posts.where(Post.content != "").all()  # добавление условия по полю связанной таблицы
await alice.posts.delete()                               # удаляются только записи этого пользователя

new_post = await alice.posts.create(content="hi")        # author автоматически = alice.id

belongs-to: прямое await возвращает экземпляр связанной модели (возвращает None, если совпадений нет или внешний ключ равен NULL):

post = await Post.get(id=1)
writer = await post.writer          # экземпляр User или None
await writer.posts.count()          # работает в обе стороны

Ключевые моменты:

Что дальше

Нужно сменить базу данных или развернуться на облачной БД

Не нужна структура таблиц, нужно лишь хранить несколько пар ключ–значение

多轮对话

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

Conversation: многоходовой диалог

Tip

Когда читать этот документ: когда нужны конечный автомат многоходового диалога и восстановление из контрольных точек. Если это не нужно — пропустите, на использование остальных функций это не влияет.

Когда команда должна последовательно запрашивать у пользователя информацию в несколько шагов (сбор данных при регистрации, анкеты, мастер настройки), используйте Conversation, чтобы выстроить в одной сессии цепочку «вопрос → ожидание ответа → проверка → следующий шаг». После тайм-аута или перезапуска диалог можно восстановить из контрольной точки.

Создание диалога

Диалог создаётся методом conversation() объекта Event:

from ErisPulse.Core.Event import command

@command("quiz")
async def quiz_handler(event):
    conv = event.conversation(timeout=30)

    await conv.say("🎮 Добро пожаловать в викторину!")

    answer = await conv.choose("Вопрос 1: кто создатель Python?", [
        "Guido van Rossum",
        "James Gosling",
        "Dennis Ritchie",
    ])

    if answer is None:
        await conv.say("Время вышло, приходите в следующий раз!")
        return

    if answer == 0:
        await conv.say("Правильно!")
    else:
        await conv.say("Неправильно, правильный ответ — Guido van Rossum")

    conv.stop()

Основной API

say(content, **kwargs)

Отправляет сообщение, возвращает self, поддерживая цепочки вызовов:

await conv.say("Первая строка").say("Вторая строка").say("Третья строка")

Можно также указать метод отправки:

await conv.say("https://example.com/image.jpg", method="Image")

wait(prompt=None, timeout=None)

Ожидает ответ пользователя, возвращает объект Event или None (тайм-аут):

# Простое ожидание
resp = await conv.wait()
if resp:
    text = resp.get_text()

# Отправка подсказки и ожидание
resp = await conv.wait(prompt="Пожалуйста, введите ваше имя:")

# Своё время ожидания (переопределяет тайм-аут диалога по умолчанию)
resp = await conv.wait(prompt="Ответьте в течение 10 секунд:", timeout=10)

confirm(prompt=None, **kwargs)

Ожидает подтверждения пользователя (да/нет), возвращает True / False / None (тайм-аут):

result = await conv.confirm("Вы уверены, что хотите удалить все данные?")
if result is True:
    await conv.say("Удалено")
elif result is False:
    await conv.say("Отменено")
else:
    await conv.say("Время ожидания истекло, ответа нет")

Встроенные распознаваемые слова подтверждения: да/yes/y/подтверждаю/точно/хорошо/ok/true/верно/ага/ладно/согласен/без проблем/можно/конечно...

Встроенные распознаваемые слова отрицания: нет/no/n/отмена/не/не надо/нельзя/cancel/false/неверно/неправильно/не/отказ...

choose(prompt, options, **kwargs)

Ожидает, пока пользователь выберет один из вариантов, возвращает индекс варианта (начиная с 0) или None:

choice = await conv.choose("Выберите цвет:", ["Красный", "Зелёный", "Синий"])
if choice is not None:
    colors = ["Красный", "Зелёный", "Синий"]
    await conv.say(f"Вы выбрали {colors[choice]}")

Пользователь может выбрать, введя номер (1/2/3) или текст варианта (Красный).

options_format="auto" (по умолчанию) автоматически выбирает встроенный стиль в зависимости от method: Markdown→маркированный список, Html→нумерованный список, остальные→простой текстовый список. Также поддерживаются "list", "inline", "md", "html" или пользовательская функция.

Поддерживается merge_prompt=True для объединения подсказки и вариантов в одно сообщение, а также плейсхолдер, управляющий позицией вставки вариантов (по умолчанию {options}, настраивается через placeholder):

choice = await conv.choose(
    "## Выберите\n{options}",
    ["Вариант А", "Вариант Б"],
    method="Markdown",
    merge_prompt=True,
)

# Собственное имя плейсхолдера
choice = await conv.choose(
    "Выберите: [choices]",
    ["Вариант А", "Вариант Б"],
    placeholder="[choices]",
)

collect(fields, **kwargs)

Пошаговый сбор информации, возвращает словарь данных или None:

data = await conv.collect([
    {"key": "name", "prompt": "Введите имя"},
    {"key": "age", "prompt": "Введите возраст",
     "validator": lambda e: e.get("alt_message", "").strip().isdigit(),
     "retry_prompt": "Возраст должен быть числом, введите ещё раз"},
    {"key": "city", "prompt": "Введите город"},
])

if data:
    await conv.say(f"Регистрация успешна!\nИмя: {data['name']}\nВозраст: {data['age']}\nГород: {data['city']}")
else:
    await conv.say("Процесс регистрации прерван")

Конфигурация полей:

Параметр Описание По умолчанию
key Ключ поля (обязателен) -
prompt Подсказка "Введите {key}"
validator Функция проверки, принимает Event, возвращает bool нет
retry_prompt Подсказка при повторе после неудачной проверки "Некорректный ввод, попробуйте ещё раз"
max_retries Максимальное число повторных попыток 3
condition Функция условия, принимает dict уже собранных данных, возвращает bool нет

Условные поля: с помощью condition можно реализовать динамические формы — поле собирается только при выполнении условия:

data = await conv.collect([
    {"key": "has_car", "prompt": "У вас есть машина? (да/нет)"},
    {"key": "car_brand", "prompt": "Введите марку машины",
     "condition": lambda d: d.get("has_car", "").lower() in ("да", "yes", "y")},
])

stop()

Вручную завершает диалог, устанавливая is_active в False:

conv.stop()

is_active

Находится ли диалог в активном состоянии:

if conv.is_active:
    await conv.say("Диалог ещё продолжается")

Управление активным состоянием

stateDiagram-v2
    state "Активен" as active
    state "Неактивен" as inactive
    [*] --> active: event.conversation()
    active --> active: say / wait / confirm / choose / collect
    active --> inactive: stop()
    active --> inactive: тайм-аут wait()
    active --> inactive: тайм-аут или исчерпание повторов в collect()
    inactive --> [*]

Диалог автоматически переходит в неактивное состояние в следующих случаях:

  1. Вызван метод stop()
  2. wait() по тайм-ауту вернул None
  3. collect() вернул None из-за тайм-аута на каком-либо шаге или исчерпания повторных попыток

После перехода в неактивное состояние все интерактивные методы (wait/confirm/choose/collect) немедленно возвращают None и больше не ожидают ввод пользователя.

Ветвления и переходы

Декоратор @conv.branch(name)

Метод branch() регистрирует ветвь диалога, а goto() выполняет переходы между ветвями:

@command("menu")
async def menu_handler(event):
    conv = event.conversation(timeout=60)

    @conv.branch("main")
    async def main_menu():
        await conv.say("=== Главное меню ===\n1. Личные данные\n2. Настройки\n3. Выход")
        resp = await conv.wait()
        if resp is None:
            return
        text = resp.get_text().strip()
        if text == "1":
            await conv.goto("profile")
        elif text == "2":
            await conv.goto("settings")
        elif text == "3":
            await conv.say("До свидания!")
            conv.stop()

    @conv.branch("profile")
    async def profile():
        await conv.say("=== Личные данные ===\nИмя: Alice\n0. Назад")
        resp = await conv.wait()
        if resp and resp.get_text().strip() == "0":
            await conv.goto("main")

    @conv.branch("settings")
    async def settings():
        await conv.say("=== Настройки ===\n1. Переключатель уведомлений\n0. Назад")
        resp = await conv.wait()
        if resp and resp.get_text().strip() == "0":
            await conv.goto("main")

    await conv.start()  # Начало с первой зарегистрированной ветви

conv.start(name=None)

Запускает диалог, по умолчанию — с первой зарегистрированной ветви:

await conv.start()           # С первой ветви
await conv.start("settings") # С указанной ветви

Контекст и персистентность

conv.context

У каждого экземпляра диалога есть встроенный словарь context для обмена состоянием между ветвями:

@conv.branch("step1")
async def step1():
    conv.context["username"] = resp.get_text().strip()
    await conv.goto("step2")

@conv.branch("step2")
async def step2():
    name = conv.context.get("username", "Неизвестно")
    await conv.say(f"Привет, {name}!")

save() / resume() / clear_saved()

Диалог поддерживает персистентность и может быть восстановлен после тайм-аута или прерывания:

# Сохранение состояния диалога (обычно вызывать вручную не нужно, см. «Автоматическое сохранение» ниже)
await conv.save()

# ... позже восстановление в той же сессии ...
conv2 = event.conversation()
if await conv2.resume():
    await conv2.say("С возвращением! Продолжим прерванный диалог")
else:
    await conv2.say("Предыдущий диалог не найден")

# Очистка сохранённого диалога
await conv.clear_saved()

Ключ хранения включает измерение target (conversation:{platform}:{user_id}:{target_id}), поэтому диалоги одного пользователя в разных сессиях не перезаписывают друг друга; архивы старого формата (без target) автоматически мигрируют при resume().

Автоматические контрольные точки и восстановление после перезапуска

Автоматическое сохранение

Фреймворк автоматически поддерживает контрольные точки в следующих случаях; обычно вручную вызывать save() не нужно:

Момент Поведение
Переход между ветвями goto() / start() Автосохранение (текущая ветвь + context)
stop() / тайм-аут wait() / сбой collect() Автоочистка (конечное состояние диалога)

TTL контрольных точек

Архивы содержат метку времени; архивы старше ErisPulse.interaction.checkpoint_ttl (по умолчанию 24 часа) удаляются:

[ErisPulse.interaction]
checkpoint_ttl = 86400  # секунды

Автоматическое восстановление после перезапуска

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

from ErisPulse.Core.Event.wrapper import Conversation

@Conversation.register_resume_handler()  # можно передать platform="onebot11", чтобы ограничить платформу
def make_conversation(event) -> Conversation:
    # Задача фабрики: заново создать диалог и перерегистрировать все ветви
    conv = event.conversation(timeout=60)

    @conv.branch("menu")
    async def menu(conv, event):
        ...

    return conv

После регистрации, когда пользователь, находившийся до перезапуска в ветви menu, пришлёт первое сообщение, фреймворк автоматически: восстановит context → захватит это сообщение → продолжит диалог из сохранённой ветви. Без зарегистрированной фабрики этот механизм не создаёт накладных расходов.

Восстановление означает захват

При успешном resume() фреймворк автоматически делает две вещи:

  1. Захват сессии: автоматически выполняет acquire эксклюзивной аренды сессии — другие модули могут через sdk.interaction.get_owner_of(event) узнать, что «этот пользователь занят диалогом»; если сессия уже занята другим модулем, восстановление отменяется (возвращает False), чтобы два диалога не конфликтовали
  2. Возврат истории: последние 10 сообщений из входящего ящика сессии помещаются в conv.recent_history (контекст LLM у AI-модуля не обрывается после восстановления); можно отключить через resume(with_history=0)
if await conv.resume(with_history=20):
    for m in conv.recent_history:
        print(m["role"], ":", m["text"])

Ручное восстановление (если не использовать автоматический механизм)

@command("continue")
async def continue_handler(event):
    conv = event.conversation()
    # ... регистрация ветвей ...
    if await conv.resume():
        conv.goto(conv.get_current_branch())

Типовые шаблоны потоков

Пошаговая регистрация

@command("register")
async def register_handler(event):
    conv = event.conversation(timeout=60)

    await conv.say("Добро пожаловать! Начинаем регистрацию.")

    data = await conv.collect([
        {"key": "username", "prompt": "Введите имя пользователя (3-20 символов)",
         "validator": lambda e: 3 <= len(e.get_text().strip()) <= 20},
        {"key": "email", "prompt": "Введите адрес электронной почты",
         "validator": lambda e: "@" in e.get_text() and "." in e.get_text(),
         "retry_prompt": "Неверный формат email, введите ещё раз"},
    ])

    if not data:
        await event.reply("Регистрация отменена")
        return

    confirmed = await conv.confirm(
        f"Подтвердить данные регистрации?\nИмя пользователя: {data['username']}\nEmail: {data['email']}"
    )

    if confirmed:
        await conv.say("✅ Регистрация успешно завершена!")
    else:
        await conv.say("❌ Регистрация отменена")

Циклический диалог

@command("chat")
async def chat_handler(event):
    conv = event.conversation(timeout=120)
    await conv.say("Включён режим диалога. Введите «выход», чтобы завершить.")

    while conv.is_active:
        resp = await conv.wait()
        if resp is None:
            await conv.say("Время вышло, диалог завершён")
            break

        text = resp.get_text().strip()

        if text == "выход":
            await conv.say("До свидания!")
            conv.stop()
        elif text == "помощь":
            await conv.say("Доступные команды: выход, помощь, статус")
        elif text == "статус":
            await conv.say("Диалог активен")
        else:
            await conv.say(f"Вы сказали: {text}")

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

交互会话

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

Система интерактивных сессий

Tip

Когда стоит читать этот документ: если нужны wait_reply, многошаговые диалоги, таймеры напоминаний, взаимоисключение сессий или сквозная трассировка. Если нет — можно пропустить, на использование остальных функций это не влияет.

Прочитав, вы узнаете: все параметры wait_reply и причины отмены, декларативные многошаговые цепочки .got(), таймеры сессий (remind / escalate), многоканальное ожидание (expect / select) и аренды взаимоисключения сессий.

В этой статье не рассматриваются: машина состояний с ветвлениями для многошаговых диалогов и восстановление после перезапуска (см. Conversation: многошаговые диалоги); параметры политики хранения почтового ящика и переопределение в рантайме (см. Почтовый ящик сессии).

Note

Материал этой главы требует ErisPulse 2.8.0+.

ErisPulse превращает «непрерывное взаимодействие с пользователем» в инфраструктуру уровня фреймворка: от одиночного wait_reply до таймеров напоминаний, многоканального ожидания, взаимоисключения сессий и восстановления после перезапуска — всем этим управляет единый менеджер интерактивных сессий (Core/Event/interaction.py, sdk.interaction).

{!--< tips >!--} Каждая способность, описанная в этой статье, несёт встроенного владельца (owner): интерактивные ожидания, аренды и таймеры фиксируют имя модуля на момент регистрации; при выгрузке модуля / закрытии адаптера фреймворк автоматически их очищает, а ожидающая сторона немедленно получает уведомление вместо томительного ожидания таймаута — это распространение системы владения на измерение взаимодействия (см. Систему владения). {!--< /tips >!--}

Ожидание ответа: wait_reply

Одной фразой: приостанавливает текущую сопрограмму в ожидании следующего сообщения целевого пользователя; при совпадении возвращает событие ответа, при таймауте или отмене — None; pattern / regex / validator позволяют отсеивать неподходящие ответы.

wait_reply — краеугольный камень интерактивных сессий: он приостанавливает текущую сопрограмму и ждёт, пока целевой пользователь «ответит» следующим сообщением.

from ErisPulse.Core.Event import command

@command("ask")
async def ask_command(event):
    reply = await event.wait_reply(prompt="Введите ваше имя:", timeout=30)
    if reply is None:
        await event.reply("Время ожидания истекло")
        return
    await event.reply(f"Привет, {reply.get_text()}!")

Все параметры

Параметр Описание По умолчанию
prompt Подсказка, отправляемая перед приостановкой None
timeout Таймаут ожидания (секунды) 60
pattern glob-фильтр (* / ? / [seq]), при несовпадении ожидание продолжается None
regex Фильтр по регулярному выражению (при одновременном указании с pattern должны совпасть оба), при несовпадении ожидание продолжается None
validator Функция проверки (принимает Event, возвращает bool), при неудаче ожидание продолжается None
callback Колбэк при получении ответа (альтернативный вариант вместо возврата значения) None
method Метод отправки prompt "Text"
session Ожидание уровня сессии: совпасть может ответ любого участника той же сессии (группы / канала) False 
# Принимаем только числовую сумму, иначе продолжаем ждать
reply = await event.wait_reply("Введите сумму:", regex=r"\d+\s*юаней", timeout=30)

# Ожидание уровня сессии: сценарий групповой работы, ответить может любой участник группы
reply = await event.wait_reply(session=True, prompt="Кто-нибудь может подсказать?")

Граница между ожиданием и командами (сквозной проход команд)

Если во время ожидания поступает сообщение с зарегистрированной командой (например, /cancel), по умолчанию оно пропускается диспетчеру команд и не потребляется как ответ (параметр конфигурации ErisPulse.event.wait_reply.cmdpass; при True поведение инвертируется — все сообщения трактуются как ответы). Критерий разбора текста совпадает с диспетчеризацией команд: приоритет у alt_message, при его отсутствии — откат к конкатенации сегментов сообщения (с 2.10.1 действует и для адаптеров, использующих только message-сегменты).

Когда ожидание может быть отменено

Ожидание больше не сводится к «ожиданию таймаута» — в следующих случаях ожидание немедленно прекращается (wait_reply возвращает None), а не заставляет вызывающего ждать до таймаута:

Триггер Причина отмены (InteractionCancelled.reason) Описание
Модуль-владелец выгружен / отключён owner_unload Очистка по владению: чьё ожидание зарегистрировано, у того и изымается при исчезновении
Адаптер закрыт / перезапущен platform_stop Все приостановленные ожидания этой платформы отменяются
Сессия заменена новым ожиданием / арендой conflict См. «арбитраж сессий» ниже
Отвечающий в чёрном списке / модуль owner отвязан revoked Повторная проверка прав при совпадении ответа: измерение идентичности scope + измерение модуля
Совпадение с ответом пользователя —— Обычный путь, возвращает событие ответа

Базовое исключение — InteractionCancelled (входит в иерархию исключений InteractionError), wait_reply уже преобразует его в возврат None; вызывающим, которым нужна причина, стоит использовать низкоуровневый API sdk.interaction.register().

Полная цепочка определения совпадения ответа

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

Совпадение ключа сессии (точное измерение user → откат к уровню сессии)
  → Текстовая фильтрация pattern / regex (при несовпадении продолжаем ждать)
  → Проверка validator (при неудаче продолжаем ждать)
  → Повторная проверка прав (измерение идентичности scope + измерение модуля owner; при неудаче ожидание прекращается)
  → Пробуждение ожидающего + принятие события в обработку (mark_processed)

Декларативные многошаговые цепочки: .got() (2.11+)

@command возвращает агент потока управления для нескольких шагов; с помощью .got() пошагово декларируются этапы сбора — функции шагов по именам параметров получают уже собранные значения (включая входные параметры команды), типы автоматически определяются по type= / аннотациям. Низкоуровнево всё компилируется в Conversation: каждый шаг — отдельная ветвь, переходы автоматически сохраняют контрольные точки, и после перезапуска ответ пользователя продолжится с сохранённого шага (фабрика восстановления регистрируется автоматически при декорировании и автоматически снимается при выгрузке модуля).

@command("weather")
async def weather(event):
    pass  # Точка входа: prompt первого .got отправляется автоматически, вопрос задавать вручную не нужно

@weather.got("city", prompt="Ответьте названием города")
async def weather_city(event, city: str):
    await event.reply(f"{city}: ясно, 25℃")

@weather.got("days", prompt="Прогноз на сколько дней?", type=int, retry_prompt="Введите число")
async def weather_days(event, city: str, days: int):
    await event.reply(f"{city}, следующие {days} дней……")

Описание параметров

Параметр Описание
name Имя собираемого значения (имя параметра, инжектируемого в функции последующих шагов; уникально в рамках потока, повторное объявление вызывает ошибку при регистрации)
prompt Вопрос, автоматически отправляемый при входе в шаг
timeout Таймаут ожидания шага (число секунд или строка длительности "90s" / "1h30m", по умолчанию 60с)
type Тип значения (int / float / str…; при неудаче разбора автоматически повторно отправляется retry_prompt, максимум max_retries раз)
validator / pattern / regex Фильтрация и проверка ответа (семантика совпадает с wait_reply)
session Ожидание уровня сессии (совпасть может ответ любого в той же сессии)
cmdpass Трёхпозиционный флаг прохода команд (по умолчанию команды вроде /cancel во время ожидания выполняются как обычно)

Семантика и границы

Таймеры сессии: remind / escalate

Превращает «таймаут» из возвращаемого значения в оркестрируемый примитив. Таймеры привязаны к интерактивной сессии, автоматически отменяются при выгрузке модуля / закрытии адаптера; лимит активных remind на одну сессию — 5.

remind: напоминание, если нет ответа

@command("ticket")
async def ticket_command(event):
    await event.reply("Тикет отправлен, результат обработки будет сообщён здесь")
    # Если 5 минут нет ответа — один раз мягко напомним; любой ответ пользователя автоматически его отменяет
    event.remind(300, "Вы ещё здесь? Как только будет результат — сразу сообщим")
    reply = await event.wait_reply(timeout=3600)
    ...

escalate: гарантированная эскалация по истечении срока

event.escalate(1800, lambda e: notify_master(f"Тикет не обрабатывался 30 минут: {event.get_command_args()}"))

Единственное отличие от remind: не отменяется ответом пользователя — эскалационное действие (уведомление владельца, передача живому оператору) является обещанием «гарантированной доставки по истечении таймаута», отменяется только вручную через cancel() / при выгрузке модуля / закрытии адаптера.

remind escalate
Поведение по истечении срока Отправка текста / выполнение callback Выполнение callback
Ответ пользователя Автоотмена Не влияет
Очистка по владению (выгрузка / закрытие платформы) Отмена Отмена
Лимит на одну сессию 5 Без ограничений (страховка — очистка по владению)

Многоканальное ожидание: expect + select

Одной фразой: expect лишь описывает ожидание, не регистрируя его; select разом приостанавливает несколько ожиданий и возвращает (индекс, событие ответа) по первому совпавшему, а несовпавшие автоматически очищаются.

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

which, reply = await event.select(
    event.expect(pattern="Согласен*", user="10001"),
    event.expect(pattern="Отказ*", user="10002"),
    event.expect(validator=lambda e: e.get_text() == "Отложить", session=True),
    timeout=60,
)
if which is None:
    await event.reply("За 60 секунд не получено ни одного результата одобрения")
elif which == 0:
    await event.reply("Одобрено")
elif which == 1:
    await event.reply("Отклонено")

{!--< tips >!--} По сравнению с ручной оркестрацией нескольких задач через asyncio.wait: select автоматически очищает несовпавшие ожидания, автоматически принимает совпавшие события в обработку, а повторная проверка прав и очистка по владению полностью работают — не нужно самостоятельно управлять никакими Future. {!--< /tips >!--}

Note

С версии 2.10.1 select — это агрегированное ожидание: все ожидания объединяются в одну запись + пользовательское правило совпадения, несколько путей с одним ключом (например, два разных pattern на одного user) больше не отменяют друг друга, возвращается первый совпавший путь. Критерий проверки текста ответа: приоритет alt_message + откат к сегментам сообщения (как при диспетчеризации команд).

Замечание о границах применимости: event.wait_for(...) подключается напрямую к шине событий адаптера и не проходит через промежуточное ПО и фильтрацию по условиям detail_type (допуск по идентичности страхуется на входе диспетчеризации); для ожиданий с полным конвейером используйте wait_reply / select.

Взаимоисключение сессий: acquire / hold / get_owner_of

Владение распространяется с «ресурсов» на «сессии» — «кем в данный момент занят этот пользователь» становится запросом первого класса.

# Запрос: кто сейчас взаимодействует в этой сессии? (при простое возвращает None)
owner = sdk.interaction.get_owner_of(event)
if owner and owner != "MyModule":
    return  # С другим модулем уже идёт диалог, не мешаем

# Эксклюзивная аренда: монопольный доступ к сессии (стратегия deny, при занятости возвращает None)
lease = sdk.interaction.acquire(event)          # TTL по умолчанию 1 час, можно передать ttl=
if lease is None:
    return  # Уже занята
try:
    ...  # Эксклюзивное взаимодействие
finally:
    lease.release()

Форма контекстного менеджера (при неудаче захвата бросается SessionOccupiedError):

with sdk.interaction.hold(event) as lease:
    ...  # Автоматическое освобождение при выходе

Аренда поддерживает продление через renew(ttl); TTL истекает лениво — просроченная аренда автоматически очищается при следующем обращении.

При восстановлении диалога через Conversation.resume() фреймворк автоматически захватывает аренду (см. «восстановление = перехват» в Conversation: многошаговые диалоги) — восстановленный диалог естественно владеет сессией, и другие модули не вмешаются.

Почтовый ящик сессии: event.history

Политику хранения (7 дней / лимит количества), API переопределения в рантайме, а также риски и аудит см. в специальной статье Почтовый ящик сессии.

Единая запись недавнего потока сообщений каждой сессии (обе стороны — пользователь + бот), служащая общим фундаментом фактов для модулей ИИ-контекста, защиты от повторов и анализа поведения — модулям больше не нужно хранить историю по отдельности.

messages = await event.history(20)   # Последние 20 сообщений текущей сессии, по возрастанию времени
for m in messages:
    print(m["role"], ":", m["text"])  # role: "user" / "bot"

Транзакции сообщений: message_tx

Все исходящие отправки внутри транзакции автоматически заносятся в журнал; при аварийном выходе отправленные сообщения автоматически отзываются в обратном порядке (шаг пропускается, если адаптер не реализует delete_message, журнал при этом ведётся в обычном режиме).

async with event.message_tx():
    await event.reply("Обрабатываем, пожалуйста, подождите")
    result = await do_something()          # Здесь бросается исключение →
    await event.reply(f"Готово: {result}")   # Предыдущее «Обрабатываем» отзывается автоматически

Отправки вне транзакции не учитываются (нулевые накладные расходы); get_send_receipts() позволяет просмотреть чеки по отправкам текущей транзакции.

Сквозная трассировка: trace-id

Каждое входящее событие автоматически получает идентификатор трассировки (переиспользуется event["id"], при отсутствии генерируется), который сопровождает:

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

Связь с другими системами

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