Введение в разработку адаптеров

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

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

Что такое адаптер

Адаптер — это мост между ErisPulse и различными платформами сообщений, который отвечает за:

Прямое преобразование: получение событий платформы и их преобразование в стандартный формат OneBot12 (Converter)
Обратное преобразование: преобразование OneBot12-сегментов сообщений в вызовы API платформы (Raw_ob12)
Управление подключениями к платформе (WebSocket/WebHook)
Предоставление унифицированного интерфейса отправки сообщений SendDSL

Архитектура адаптера

flowchart LR
    subgraph receive["Прямое преобразование (прием)"]
        direction TB
        P1["События платформы"] --> C1["Converter.convert()"] --> O1["События в стандарте OneBot12"] --> S1["Система событий"] --> M1["Обработка модулями"]
    end
    subgraph send["Обратное преобразование (отправка)"]
        direction TB
        M2["Создание сообщения модулем"] --> R1["Send.Raw_ob12()"] --> N1["Вызов API платформы"] --> R2["Стандартный формат ответа"]
    end

Структура каталогов

Стандартная структура пакета адаптера:

MyAdapter/
├── pyproject.toml          # Конфигурация проекта
├── README.md               # Описание проекта
├── LICENSE                 # Лицензия
└── MyAdapter/
    ├── __init__.py          # Точка входа пакета
    ├── Core.py               # Главный класс адаптера
    └── Converter.py          # Конвертер событий

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

1. Создание проекта

mkdir MyAdapter && cd MyAdapter

2. Создание pyproject.toml

[project]
name = "ErisPulse-MyAdapter"
version = "1.0.0"
description = "Адаптер MyAdapter"
readme = "README.md"
requires-python = ">=3.10"
license = { file = "LICENSE" }
authors = [ { name = "yourname", email = "[email protected]" } ]

dependencies = [
    "ErisPulse>=2.4.0"  # aiohttp уже встроен в ErisPulse, обычно не нужно отдельно зависеть
]

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

[project.entry-points."erispulse.adapter"]
"MyAdapter" = "MyAdapter:MyAdapter"

3. Создание главного класса адаптера

Рамка предоставляет декларативное управление конфигурацией с помощью ConfigClass / AccountConfigClass. Адаптеру нужно только объявить класс конфигурации, чтобы автоматически загружать, проверять и генерировать шаблон конфигурации.

# MyAdapter/Core.py
from dataclasses import dataclass, field
from ErisPulse.Core import BaseAdapter
from ErisPulse.Core.Bases import BaseConfig

@dataclass
class MyAdapterConfig(BaseConfig):
    """Конфигурация MyAdapter"""
    api_endpoint: str = field(
        default="https://api.example.com",
        metadata={
            "description": {"i18n": "my_adapter.api_endpoint", "default": "Адрес API"},
            "required": False,
            "ui": {"widget": "text", "group": "connection", "order": 1},
        },
    )
    token: str = field(
        default="",
        metadata={
            "description": {"i18n": "my_adapter.token", "default": "Токен платформы"},
            "required": True,
            "secret": True,
            "ui": {"widget": "password", "group": "basic", "order": 2},
        },
    )

class MyAdapter(BaseAdapter):
    ConfigClass = MyAdapterConfig  # Объявление класса конфигурации, управление автоматически
    
    # Не нужно переопределять __init__! Рамка автоматически обрабатывает:
    # - self.sdk / self.logger автоматически устанавливаются
    # - self.cfg читается в реальном времени из конфигурации
    # - self.Send / self.Request автоматически инициализируются
    
    def _setup_converter(self):
        from .Converter import MyPlatformConverter
        return MyPlatformConverter()

⚠️ О __init__: В новой версии BaseAdapter.__init__(self, sdk=None) автоматически обрабатывает ссылку на SDK, инициализацию логирования и загрузку конфигурации. Большинству адаптеров не нужно переопределять __init__. Подробнее см. Примечания по init.

⚠️ О super().__init__(): BaseAdapter.__init__() отвечает за создание фабрик Send и Request. Если забыть вызвать, все операции отправки сообщений и запросов будут вызывать AttributeError. Подробнее см. Примечания по init.

4. Реализация обязательных методов

class MyAdapter(BaseAdapter):
    # ... код __init__ ...
    
    async def start(self):
        """Запуск адаптера (обязательно реализовать)"""
        # Регистрация маршрутов WebSocket или WebHook
        router.register_websocket(
            module_name="myplatform",
            path="/ws",
            handler=self._ws_handler
        )
        self.logger.info("Адаптер запущен")
    
    async def shutdown(self):
        """Остановка адаптера (обязательно реализовать)"""
        router.unregister_websocket(
            module_name="myplatform",
            path="/ws"
        )
        # Очистка подключений и ресурсов
        self.logger.info("Адаптер остановлен")
    
    async def call_api(self, endpoint: str, **params):
        """Вызов API платформы (обязательно реализовать)"""
        raise NotImplementedError("Необходимо реализовать call_api")

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

Адаптер должен активно отправлять метасобытия, чтобы система отслеживала состояние онлайн-бота. Это можно сделать одной строкой с помощью emit_meta():

class MyAdapter(BaseAdapter):
    async def _ws_handler(self, websocket):
        bot_id = self._get_bot_id()

        # Бот вошёл в сеть
        await self.emit_meta("connect", bot_id, user_name="MyBot")

        try:
            while True:
                data = await websocket.receive_text()
                event = self.convert(data)
                if event:
                    await self.adapter.emit(event)
        except WebSocketDisconnect:
            pass
        finally:
            # Бот вышел из сети
            await self.emit_meta("disconnect", bot_id)

Подробнее о управлении состоянием бота и метасобытиях см. Рекомендации по адаптерам - Управление состоянием бота и метасобытия.

5. Реализация класса Send

Декораторы At/AtAll/Reply уже реализованы в базовом классе SendDSL, адаптеру нужно только реализовать Raw_ob12 и конкретные методы отправки.

Рамка предоставляет два ключевых вспомогательных метода:

self._apply_modifiers(message) — автоматически объединяет At/AtAll/Reply декораторы в сегменты сообщения
self.send_context — получает словарь контекста отправки (target_type, target_id, account_id)

import asyncio

class MyAdapter(BaseAdapter):
    # ... остальной код ...
    
    class Send(BaseAdapter.Send):

        def Raw_ob12(self, message, **kwargs):
            """
            Отправка сообщения в формате OneBot12 (обязательно реализовать)

            Использование _apply_modifiers для автоматического объединения состояния декораторов,
            использование send_context для получения контекста отправки.
            """
            async def _do_send():
                segments = self._apply_modifiers(message)
                return await self._adapter.call_api(
                    endpoint="/send_message",
                    message=segments,
                    **self.send_context,
                    **kwargs
                )
            return asyncio.create_task(_do_send())

        # Методы Text/Image/Voice/Video/File уже унаследованы от базового класса SendDSL,
        # по умолчанию делегируют Raw_ob12, повторная реализация не требуется.
        # Если нужны платформенно-специфические логики, можно переопределить отдельные методы:
        # def Text(self, text: str):
        #     return self.Raw_ob12([{"type": "text", "data": {"text": text}}])

Особенности реализации методов отправки медиа-файлов (Image/Video/File):

Базовая реализация по умолчанию упаковывает параметр file в сегмент OneBot12 и передаёт в Raw_ob12, адаптер должен обрабатывать загрузку/передачу в Raw_ob12
Параметр file должен поддерживать как bytes (бинарные данные), так и str (URL)
При передаче URL необходимо сначала загрузить файл, а затем загрузить на платформу
Платформа обычно требует сначала вызвать загрузочный API для получения идентификатора файла, а затем вызвать API отправки

Метод __getattr__:

Реализация методов нечувствительна к регистру (Text, text, TEXT вызывают один и тот же метод)
Неопределённые методы должны возвращать информационное сообщение, а не вызывать ошибку

Метод Raw_ob12:

Преобразует стандартный формат OneBot12 в формат платформы для отправки
Использует self._apply_modifiers(message) для автоматической обработки декораторов At/AtAll/Reply
Использует **self.send_context для передачи информации о цели и аккаунте

6. Реализация конвертера

# MyAdapter/Converter.py
import time
import uuid

class MyPlatformConverter:
    def convert(self, raw_event):
        """Преобразование событий платформы в стандартный формат OneBot12"""
        if not isinstance(raw_event, dict):
            return None
        
        onebot_event = {
            "id": str(raw_event.get("event_id", uuid.uuid4())),
            "time": int(time.time()),
            "type": self._convert_event_type(raw_event.get("type")),
            "detail_type": self._convert_detail_type(raw_event),
            "platform": "myplatform",
            "self": {
                "platform": "myplatform",
                "user_id": str(raw_event.get("bot_id", ""))
            },
            "myplatform_raw": raw_event,
            "myplatform_raw_type": raw_event.get("type", "")
        }
        
        return onebot_event
    
    def _convert_event_type(self, event_type):
        """Преобразование типа события"""
        type_map = {
            "message": "message",
            "notice": "notice"
        }
        return type_map.get(event_type, "unknown")
    
    def _convert_detail_type(self, raw_event):
        """Преобразование детального типа"""
        return "private"  # Упрощённый пример

7. Реализация класса Request (операции запросов)

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

from ErisPulse.Core import BaseAdapter, RequestDSL

class MyAdapter(BaseAdapter):
    # ... Send и другие коды ...

    class Request(RequestDSL):
        """Реализация операций запросов (запросы на добавление, приглашения и т.д.)"""

        def accept(self, **kwargs):
            """Принять запрос"""
            async def _do():
                result = await self._adapter.call_api(
                    endpoint="/set_request",
                    request_id=self._request_id,
                    approve=True,
                    **kwargs,
                )
                return {
                    "status": "ok" if result.get("code") == 0 else "failed",
                    "retcode": result.get("code", 0),
                    "data": None,
                    "message_id": "",
                    "message": result.get("message", ""),
                }
            return self._create_task(_do())

        def reject(self, **kwargs):
            """Отклонить запрос"""
            async def _do():
                result = await self._adapter.call_api(
                    endpoint="/set_request",
                    request_id=self._request_id,
                    approve=False,
                    **kwargs,
                )
                return {
                    "status": "ok" if result.get("code") == 0 else "failed",
                    "retcode": result.get("code", 0),
                    "data": None,
                    "message_id": "",
                    "message": result.get("message", ""),
                }
            return self._create_task(_do())

Способ использования модулями:

from ErisPulse.Core.Event import request

@request.on_friend_request()
async def handle_friend_request(event):
    # Использование удобного метода Event
    await event.approve()
    # Или напрямую через адаптер
    await adapter.myplatform.Request("req_id").accept()

Если платформа не поддерживает операции запросов, можно не реализовывать внутренний класс Request. Базовый класс по умолчанию возвращает retcode=10002 (операция не поддерживается). Подробнее см. Спецификация операций запросов.

8. Создание точки входа пакета

# MyAdapter/__init__.py
from .Core import MyAdapter

Зависимости (опционально, 2.8.0+)

Адаптер может объявлять зависимости от других адаптеров или модулей для реализации взаимодействия между адаптерами и опциональных функций:

from typing import ClassVar

class MyAdapter(BaseAdapter):
    # Жёсткая зависимость: отсутствие приводит к пропуску запуска (предупреждение + событие status=skipped-dependency)
    depends: ClassVar[dict] = {
        "adapters": ["onebot11"],   # Зависимые адаптеры (по названию платформы)
        "modules": ["TranslateEngine"],  # Зависимые модули (по зарегистрированному имени)
    }
    # Опциональная зависимость: отсутствие не влияет на запуск; модули получают обратный вызов при загрузке/выгрузке (режим опциональной функции)
    optional_modules: ClassVar[list] = ["TranslateEngine"]

Порядок запуска: адаптеры, объявившие жёсткую зависимость от модуля, будут запускаться после инициализации модуля
Уведомления по опциональной зависимости: при загрузке модуля из optional_modules (или жёсткой зависимости) вызывается on_dependency_ready(module_name); при выгрузке вызывается on_dependency_lost(module_name) (по умолчанию пустой метод, можно переопределить) — для сценариев поздней загрузки и горячей перезагрузки:

async def on_dependency_ready(self, module_name):
    """Опциональный модуль готов: включить соответствующую опциональную функцию"""
    if module_name == "TranslateEngine":
        self._translate = self.sdk.TranslateEngine

async def on_dependency_lost(self, module_name):
    """Опциональный модуль утрачен: понизить функциональность"""
    if module_name == "TranslateEngine":
        self._translate = None

Note

Эта функция требует ErisPulse 2.8.0+.

Примечания по `init`

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

1. Уровень BaseAdapter (в большинстве случаев не нужно переопределять)

BaseAdapter.__init__(self, sdk=None) отвечает за создание фабрик Send / Request и автоматически выполняет следующее:

Принятие параметра sdk и установку self.sdk, self.logger
Если объявлен ConfigClass, можно читать глобальную конфигурацию через self.cfg в реальном времени
Если объявлен AccountConfigClass, можно читать конфигурацию нескольких аккаунтов через self.accounts в реальном времени

В большинстве случаев не нужно переопределять __init__, достаточно объявить ConfigClass:

class MyAdapter(BaseAdapter):
    ConfigClass = MyAdapterConfig  # После объявления конфигурация автоматически управляется
    
    async def start(self):
        cfg = self.cfg  # Статическая типизация, чтение в реальном времени
        ...

Если действительно нужно пользовательскую инициализацию, вызовите super().__init__(sdk):

class MyAdapter(BaseAdapter):
    ConfigClass = MyAdapterConfig
    
    def __init__(self, sdk=None):
        super().__init__(sdk)  # Передать sdk
        self.converter = self._setup_converter()
        self.convert = self.converter.convert

2. Внутренний класс Send (в большинстве случаев не нужно переопределять)

SendDSL.__init__ отвечает за передачу состояния при цепочечных вызовах (тип цели, ID цели, аккаунт и т.д.). В большинстве случаев нужно переопределять только методы (Raw_ob12, Text и т.д.), а не __init__.

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

class MyAdapter(BaseAdapter):
    class Send(BaseAdapter.Send):
        # Параметры: adapter, target_type, target_id, account_id
        def __init__(self, adapter, target_type=None, target_id=None, account_id=None):
            super().__init__(adapter, target_type, target_id, account_id)  # ← Обязательно передать
            self._my_state = None  # Инициализация платформенно-специфического состояния

Почему необходимо передавать? Каждый шаг цепочечного вызова создаёт новый экземпляр через self.__class__(...):

adapter.Send.To("user", "123")               # → Send(adapter, "user", "123", None)
adapter.Send.To("user", "123").Using("bot1")  # → Send(adapter, "user", "123", "bot1")

Если подпись __init__ не совпадает или не вызван super(), цепочечный вызов прервётся.

3. Внутренний класс Request (в большинстве случаев не нужно переопределять)

Аналогично Send. Параметры: adapter, request_id, account_id:

class MyAdapter(BaseAdapter):
    class Request(RequestDSL):
        # Параметры: adapter, request_id, account_id
        def __init__(self, adapter, request_id=None, account_id=None):
            super().__init__(adapter, request_id, account_id)  # ← Обязательно передать
            self._my_state = None  # Инициализация платформенно-специфического состояния

Сводка

Уровень	Когда переопределять	Обязательные действия
BaseAdapter	При необходимости пользовательской логики инициализации	`super().__init__(sdk)` (передать параметр sdk)
Send внутренний класс	При необходимости инициализации состояний отправки	`super().__init__(adapter, target_type, target_id, account_id)`
Request внутренний класс	При необходимости инициализации состояний запросов	`super().__init__(adapter, request_id, account_id)`
Все уровни	В большинстве случаев	Объявить ConfigClass, не трогать `__init__`

9. Информация о подключении и обнаружение маршрутов

После регистрации маршрутов адаптером, система сохраняет всю информацию о маршрутах. Пользователи могут получить доступ к информации о подключении адаптера с помощью следующего API:

from ErisPulse import sdk

# Получение полной информации о подключении адаптера
info = sdk.adapter.get_connection_info("myplatform")
# {
#   "platform": "myplatform",
#   "status": "started",
#   "connection": {
#     "base_url": "http://localhost:8080",
#     "http_routes": [
#       {"path": "/myplatform/webhook", "method": "POST",
#        "url": "http://localhost:8080/myplatform/webhook"}
#     ],
#     "websocket_routes": [
#       {"path": "/myplatform/ws",
#        "url": "ws://localhost:8080/myplatform/ws"}
#     ]
#   }
# }

# Перечисление всех пространств имён (адаптеры/модули) маршрутов
namespaces = sdk.router.list_namespaces()
# {"myplatform": {"http": ["/myplatform/webhook"], "websocket": ["/myplatform/ws"]}}

# Получение полных URL подключения пространства имён
urls = sdk.router.get_module_urls("myplatform")
# {"base_url": "http://localhost:8080", "http": [...], "websocket": [...]}

# Получение подробной информации о маршрутах пространства имён
routes = sdk.router.get_module_routes("myplatform")
# {"http": [{"path": "/myplatform/webhook", "methods": ["POST"]}],
#  "websocket": [{"path": "/myplatform/ws", "auth": false}]}

Подсказка: Возвращаемая информация из get_connection_info() подходит для отображения пользователю (например, в WebUI), помогая настроить адрес обратного вызова или адрес подключения WebSocket на стороне платформы. Имя модуля, указанное при регистрации маршрута, должно полностью соответствовать названию платформы, зарегистрированному в ErisPulse, иначе обнаружение маршрутов не будет корректным.

10. Поддержка SSE (Server-Sent Events)

ErisPulse включает в себя серверно-независимую поддержку SSE, модули и адаптеры могут регистрировать конечные точки SSE с помощью @sdk.router.sse().

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

import asyncio
from ErisPulse import sdk

@sdk.router.sse("MyModule", "/events")
async def event_stream(sse):
    """Отправка событий SSE"""
    count = 0
    while not sse.closed:
        await sse.send({"count": count}, event="update")
        count += 1
        await asyncio.sleep(1)

Использование параметров запроса

Обработчик может объявить параметр request, чтобы получить информацию о клиентском запросе:

@sdk.router.sse("MyModule", "/events")
async def event_stream(request, sse):
    token = request.query_params.get("token")
    if not validate_token(token):
        await sse.close()
        return

    while not sse.closed:
        data = await fetch_data(token)
        await sse.send(data)
        await asyncio.sleep(5)

API SseEmitter

Метод	Описание
`sse.send(data, event=None, id=None, retry=None)`	Отправка события SSE. `data`, не являющийся строкой, автоматически сериализуется в JSON
`sse.close()`	Безопасное закрытие подключения SSE (можно вызывать несколько раз)
`sse.closed`	Закрыто ли подключение
`sse.request`	Объект базового запроса (можно использовать для чтения query params, headers)

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

api = sdk.router.group("MyModule", "/api", version="1")

@api.sse("/events")
async def events(sse):
    await sse.send({"msg": "hello"})

Обнаружение маршрутов

Маршруты SSE автоматически появляются в API обнаружения маршрутов:

# list_namespaces будет содержать ключ "sse"
sdk.router.list_namespaces()
# {"MyModule": {"http": [...], "websocket": [...], "sse": ["/MyModule/events"]}}

# get_module_routes пометит streaming: true
sdk.router.get_module_routes("MyModule")
# {"http": [...], "websocket": [...], "sse": [{"path": "/MyModule/events", "streaming": true}]}

# get_module_urls сгенерирует полный URL
sdk.router.get_module_urls("MyModule")
# {"sse": [{"path": "/MyModule/events", "url": "http://localhost:8080/MyModule/events"}]}

Дизайн, независимый от сервера: SseEmitter использует обратные вызовы для декомпозиции от базового HTTP-фреймворка. Рамка предоставляет register_sse() и декоратор @sse в качестве единого регистрационного входа, адаптер не должен напрямую зависеть от любого базового HTTP-фреймворка, чтобы реализовать конечную точку SSE.

Основные концепции адаптера - Понимание архитектуры адаптера
Подробности SendDSL - Изучение отправки сообщений
Реализация конвертера - Понимание преобразования событий
Рекомендации по адаптерам - Разработка качественных адаптеров