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

Основные понятия

В этом руководстве представлены основные концепции ErisPulse, которые помогут вам понять философию и архитектуру фреймворка.

Архитектура на основе событий

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 и внешними платформами.

Задачи:

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

4. Модули

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

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

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

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

При инициализации SDK происходит сканирование всех entry points группы 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("Модуль выгружен")

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

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

С помощью get_load_strategy() объявляется поведение загрузки модуля:

from ErisPulse.loaders import ModuleLoadStrategy

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

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

Типы событий

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

Тип события Декоратор Описание
Событие сообщения @message.on_message() Любое сообщение, отправленное пользователем (личные сообщения, группы)
Событие команды @command("name") Сообщение, начинающееся с префикса команды (например, /hello)
Уведомление @notice.on_friend_add() и др. Системные уведомления (добавление друзей, изменения в группе)
Запрос @request.on_friend_request() и др. Запросы от пользователей (запросы на добавление в друзья, приглашения в группы)
Мета-событие @meta.on_connect() и др. Системные события (подключение, отключение, heartbeat)

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

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

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"Эхо: {text}")

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

DSL для отправки сообщений SendDSL

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

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

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

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

# Указание отправляющего аккаунта
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):

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

Далее