适配器开发最佳实践

本文档提供了 ErisPulse 适配器开发的最佳实践建议。

Bot 状态管理与 Meta 事件

适配器应主动通过 adapter.emit() 发送 meta 事件，让框架自动追踪 Bot 的连接状态、上下线和心跳信息。

1. 何时发送 Meta 事件

2. 发送 Meta 事件

框架提供 emit_meta() 方法，一行即可发送 meta 事件：

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

        # Bot 上线：一行发送 connect 事件
        await self.emit_meta("connect", bot_id, user_name="MyBot", nickname="我的机器人")

        try:
            while True:
                data = await websocket.receive_text()
                event = self.convert(data)
                if event:
                    await self.adapter.emit(event)
        except WebSocketDisconnect:
            pass
        finally:
            # Bot 下线
            await self.emit_meta("disconnect", bot_id)

3. 心跳事件

适配器应在连接存活期间定期发送心跳事件，更新 Bot 的活跃时间：

class MyAdapter(BaseAdapter):
    async def _heartbeat_loop(self, bot_id: str):
        while self._connected:
            # 向框架发送 meta heartbeat（一行完成）
            await self.emit_meta("heartbeat", bot_id)
            await asyncio.sleep(30)

4. self 字段自动发现

框架的 adapter.emit() 会自动处理所有事件（不仅是 meta 事件）中的 self 字段：

• 普通事件（message/notice/request）中的 self 字段会自动发现并注册 Bot
• self 字段扩展信息：支持 user_name、nickname、avatar、account_id 可选字段

# 转换器中包含 self 字段即可自动注册 Bot
onebot_event = {
    "type": "message",
    "detail_type": "private",
    "platform": "myplatform",
    "self": {
        "platform": "myplatform",
        "user_id": "bot123",
        "user_name": "MyBot",
        "nickname": "我的机器人",
    },
    # ... 其他字段
}
await self.adapter.emit(onebot_event)
# Bot "bot123" 已自动注册并更新活跃时间

5. Bot 状态查询

框架提供以下查询方法：

from ErisPulse import sdk

# 获取 Bot 详细信息
info = sdk.adapter.get_bot_info("myplatform", "bot123")
# {"status": "online", "last_active": 1712345678.0, "info": {"nickname": "MyBot"}}

# 列出所有 Bot（按平台分组）
all_bots = sdk.adapter.list_bots()

# 列出指定平台的 Bot
platform_bots = sdk.adapter.list_bots("myplatform")

# 检查 Bot 是否在线
is_online = sdk.adapter.is_bot_online("myplatform", "bot123")

# 获取完整状态摘要（适合 WebUI 展示）
summary = sdk.adapter.get_status_summary()
# {"adapters": {"myplatform": {"status": "started", "bots": {...}}}}

连接管理

1. 实现连接重试

import asyncio

class MyAdapter(BaseAdapter):
    async def start(self):
        retry_count = 0
        max_retries = 5
        
        while retry_count < max_retries:
            try:
                await self._connect_to_platform()
                self.logger.info("连接成功")
                break
            except Exception as e:
                retry_count += 1
                if retry_count < max_retries:
                    # 指数退避策略
                    wait_time = min(60 * (2 ** retry_count), 600)
                    self.logger.warning(
                        f"连接失败，{wait_time}秒后重试 ({retry_count}/{max_retries}): {e}"
                    )
                    await asyncio.sleep(wait_time)
                else:
                    self.logger.error("连接失败，已达到最大重试次数")
                    raise

2. 连接状态管理

class MyAdapter(BaseAdapter):
    async def start(self):
        self.connection = None
        self._connected = False
    
    async def _ws_handler(self, websocket: WebSocket):
        self.connection = websocket
        self._connected = True
        self.logger.info("连接已建立")
        
        try:
            while True:
                data = await websocket.receive_text()
                await self._process_event(data)
        except WebSocketDisconnect:
            self.logger.info("连接已断开")
        finally:
            self.connection = None
            self._connected = False

3. 心跳保活与 Meta 心跳

适配器的心跳应同时完成两个任务：向平台发送心跳保活，并向框架发送 meta heartbeat 事件。

class MyAdapter(BaseAdapter):
    async def start(self):
        self.connection = await self._connect_to_platform()
        self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())

    async def _heartbeat_loop(self):
        while self.connection:
            try:
                # 1. 向平台发送心跳保活
                await self.connection.send_json({"type": "ping"})

                # 2. 向框架发送 meta heartbeat（使用 emit_meta 一行完成）
                await self.emit_meta("heartbeat", self._bot_id)

                await asyncio.sleep(30)
            except Exception as e:
                self.logger.error(f"心跳失败: {e}")
                break

4. 连接信息暴露

适配器注册的路由应对用户可见，便于用户配置平台侧的回调地址。推荐在 start() 中主动输出连接信息：

class MyAdapter(BaseAdapter):
    async def start(self):
        router.register_websocket(
            module_name=self.platform,
            path="/ws",
            handler=self._ws_handler
        )

        if self.sdk:
            info = self.sdk.adapter.get_connection_info(self.platform)
            if info:
                self.logger.info(f"WebSocket 地址: "
                    f"{info.get('connection', {}).get('base_url', '')}"
                    f"{info.get('connection', {}).get('websocket_routes', [])}")

用户可以通过以下 API 查看适配器的所有路由和连接地址：

from ErisPulse import sdk

# 适配器级别的连接信息（推荐）
info = sdk.adapter.get_connection_info("myplatform")

# 路由管理器级别的查询
sdk.router.list_namespaces()              # 列出所有命名空间
sdk.router.get_module_routes("myplatform")  # 详细路由信息
sdk.router.get_module_urls("myplatform")    # 完整连接 URL

注意：路由注册时的 module_name 必须与适配器在 ErisPulse 中注册的 platform 名称完全一致，否则 get_connection_info() 将无法关联路由。多账户适配器应为每个账户注册子路径（如 /account1/webhook、/account2/webhook），而非使用不同的 module_name。

事件转换

1. 严格遵循 OneBot12 标准

class MyPlatformConverter:
    def convert(self, raw_event):
        """转换事件"""
        onebot_event = {
            "id": str(raw_event.get("event_id", uuid.uuid4())),
            "time": int(time.time()),
            "type": self._convert_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

2. 时间戳标准化

def _convert_timestamp(self, timestamp):
    """转换为 10 位秒级时间戳"""
    if not timestamp:
        return int(time.time())
    
    # 如果是毫秒级时间戳
    if timestamp > 10**12:
        return int(timestamp / 1000)
    
    # 如果是秒级时间戳
    return int(timestamp)

3. 事件 ID 生成

import uuid

def _generate_event_id(self, raw_event):
    """生成事件 ID"""
    event_id = raw_event.get("event_id")
    if event_id:
        return str(event_id)
    # 如果平台没有提供 ID，生成 UUID
    return str(uuid.uuid4())

SendDSL 实现

At/AtAll/Reply 修饰器已由框架 SendDSL 基类内置，适配器只需实现 Raw_ob12 和具体发送方法。使用 self._apply_modifiers(message) 和 self.send_context 简化开发。

1. 必须返回 Task 对象

class Send(BaseAdapter.Send):
    def Raw_ob12(self, message, **kwargs):
        """推荐实现：使用框架辅助方法"""
        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())

    def Text(self, text: str):
        return self.Raw_ob12([{"type": "text", "data": {"text": text}}])

2. 链式修饰方法返回 self

class Send(BaseAdapter.Send):

    def __init__(self, adapter, target_type=None, target_id=None, account_id=None):
        super().__init__(adapter, target_type, target_id, account_id)
        self.buttons = []

    def Button(self, content: list) -> 'Send':
        self.buttons.append(content)
        return self # 返回 self

3. 支持平台特有方法

class Send(BaseAdapter.Send):
    def Sticker(self, sticker_id: str):
        """发送表情包"""
        return asyncio.create_task(
            self._adapter.call_api(
                endpoint="/send_sticker",
                message=[{"type": "sticker", "data": {"id": sticker_id}}],
                **self.send_context
            )
        )
    
    def Card(self, card_data: dict):
        """发送卡片消息"""
        return asyncio.create_task(
            self._adapter.call_api(
                endpoint="/send_card",
                message=[{"type": "card", "data": card_data}],
                **self.send_context
            )
        )

API 响应

1. 标准化响应格式

框架提供 make_response() 和 make_error() 方法构造标准化响应：

async def call_api(self, endpoint: str, **params):
    try:
        raw_response = await self._platform_api_call(endpoint, **params)
        
        if raw_response.get("success"):
            return self.make_response(
                data=raw_response.get("data"),
                message_id=raw_response.get("data", {}).get("message_id", ""),
                raw=raw_response,
            )
        else:
            return self.make_error(
                retcode=raw_response.get("code", 10001),
                message=raw_response.get("message", ""),
                raw=raw_response,
            )
    except Exception as e:
        return self.make_error(message=str(e))

make_response() 会自动生成包含 {platform}_raw 键的响应字典。make_error() 默认使用 retcode=34000（Platform Error）。

2. 错误码规范

遵循 OneBot12 标准错误码：

# 1xxxx - 动作请求错误
10001: Bad Request
10002: Unsupported Action
10003: Bad Param

# 2xxxx - 动作处理器错误
20001: Bad Handler
20002: Internal Handler Error

# 3xxxx - 动作执行错误
31000: Database Error
32000: Filesystem Error
33000: Network Error
34000: Platform Error
35000: Logic Error

多账户支持

1. 声明式配置（推荐）

使用 AccountConfigClass 声明配置类后，框架自动管理多账户加载、校验和模板生成。BotAccountConfig 基类提供 enabled 和 name 字段，适配器无需声明：

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

@dataclass
class MyBotConfig(BotAccountConfig):
    token: str = field(default="", metadata={
        "description": {"i18n": "my_adapter.bot_token", "default": "Bot Token"},
        "required": True,
        "secret": True,
    })

class MyAdapter(BaseAdapter):
    AccountConfigClass = MyBotConfig
    
    async def start(self):
        for name, account in self.enabled_accounts.items():
            self.logger.info(f"启动账户 {name}")
            await self._connect(name, account.token)
            # bot_id 由框架自动从平台协议/登录响应中获取并回填
    
    async def call_api(self, endpoint: str, **params):
        account_id = params.pop("account_id", None)
        name, account = self._resolve_account(account_id)
        # name: 账户名, account: MyBotConfig 实例

配置文件自动生成为：

[MyAdapter.accounts.default]
token = ""
enabled = true
name = ""

2. 账户选择机制

框架内置 _resolve_account() 方法，匹配优先级：

1. 账户名 — 配置键名精确匹配
2. bot_id 字段 — 自动获取的 bot_id（即 event["self"]["user_id"]）
3. 任意 str 字段 — 配置中其他字符串字段
4. 兜底 — 第一个启用的账户

# 按账户名匹配
name, account = self._resolve_account("account1")

# 按 bot_id 匹配（最常用的方式，来自事件）
name, account = self._resolve_account("bot_123")

# 获取第一个启用的账户（传入 None）
name, account = self._resolve_account(None)

错误处理

1. 分类异常处理

使用 make_error() 构造标准化错误响应。通过 sdk.client 请求时捕获 ErisPulse 异常：

from ErisPulse.Core.Bases.errors import ClientError, ClientTimeoutError

async def call_api(self, endpoint: str, **params):
    try:
        from ErisPulse.Core import client
        resp = await client.post(
            f"https://api.platform.com/{endpoint}",
            json=params,
            max_retries=2,
        )
        response = await resp.json()
        return self.make_response(data=response, raw=response)
    except ClientTimeoutError:
        self.logger.error(f"请求超时: {endpoint}")
        return self.make_error(retcode=32000, message="请求超时")
    except ClientError as e:
        self.logger.error(f"网络错误: {e}")
        return self.make_error(retcode=33000, message="网络请求失败")
    except json.JSONDecodeError:
        self.logger.error("JSON 解析失败")
        return self.make_error(retcode=10006, message="响应格式错误")
    except Exception as e:
        self.logger.error(f"未知错误: {e}", exc_info=True)
        return self.make_error(message=str(e))

向后兼容：直接使用 aiohttp 的旧适配器代码不受影响，仍可捕获 aiohttp.ClientError。异常转换仅在通过 sdk.client 发起请求时生效。

2. 日志记录

框架自动为适配器创建子 logger（sdk.logger.get_child("MyAdapter")），无需手动初始化：

class MyAdapter(BaseAdapter):
    # ConfigClass = ...  # 声明配置类后 self.logger 自动可用
    
    async def start(self):
        self.logger.info("适配器启动中...")
        # ...
        self.logger.info("适配器启动完成")
    
    async def shutdown(self):
        self.logger.info("适配器关闭中...")
        # ...
        self.logger.info("适配器关闭完成")

测试

1. 单元测试

import pytest
from ErisPulse.Core.Bases import BaseAdapter

class TestMyAdapter:
    def test_converter(self):
        """测试转换器"""
        converter = MyPlatformConverter()
        raw_event = {"type": "message", "content": "Hello"}
        result = converter.convert(raw_event)
        assert result is not None
        assert result["platform"] == "myplatform"
        assert "myplatform_raw" in result
    
    def test_api_response(self):
        """测试 API 响应格式"""
        adapter = MyAdapter()
        response = adapter.call_api("/test", param="value")
        assert "status" in response
        assert "retcode" in response

2. 集成测试

@pytest.mark.asyncio
async def test_adapter_start():
    """测试适配器启动"""
    adapter = MyAdapter()
    await adapter.start()
    assert adapter._connected is True

@pytest.mark.asyncio
async def test_send_message():
    """测试发送消息"""
    adapter = MyAdapter()
    await adapter.start()
    
    result = await adapter.Send.To("user", "123").Text("Hello")
    assert result is not None

反向转换与消息构建

Raw_ob12 是适配器必须实现的方法，是反向转换（OneBot12 → 平台）的统一入口。标准方法（Text、Image 等）应委托给 Raw_ob12，修饰器状态（At/Reply/AtAll）需在 Raw_ob12 内合并为消息段。

MessageBuilder 是配合 Raw_ob12 使用的消息段构建工具，支持链式调用和快速构建。

完整的实现规范、代码示例和使用方法请参阅：

• 发送方法规范 §6 反向转换规范
• 发送方法规范 §11 消息构建器

平台事件方法扩展

适配器可以为 Event 包装类注册平台专有方法，让模块开发者能更方便地访问平台特有数据。

1. 使用 Mixin 类批量注册（推荐）

当平台有多个专有方法时，推荐使用 Mixin 类：

# 在适配器的 start() 或模块级别注册
from ErisPulse.Core.Event import register_event_mixin

class MyPlatformEventMixin:
    def get_chat_name(self):
        """获取聊天名称"""
        return self.get("myplatform_raw", {}).get("chat", {}).get("name", "")

    def is_official_message(self):
        """判断是否为官方消息"""
        raw = self.get("myplatform_raw", {})
        return raw.get("sender", {}).get("is_official", False)

    def get_message_type(self):
        """获取平台消息类型"""
        return self.get("myplatform_raw", {}).get("msg_type", "text")

# 批量注册
register_event_mixin("myplatform", MyPlatformEventMixin)

2. 使用装饰器注册单个方法

from ErisPulse.Core.Event import register_event_method

@register_event_method("myplatform")
def get_chat_name(self):
    return self.get("myplatform_raw", {}).get("chat", {}).get("name", "")

3. 适配器关闭时清理

from ErisPulse.Core.Event import unregister_platform_event_methods

class MyAdapter(BaseAdapter):
    async def shutdown(self):
        # 清理平台事件方法注册
        unregister_platform_event_methods("myplatform")
        # ... 其他清理

更详细的注册和注销说明请参阅 事件系统 API - 注册平台扩展方法。

文档维护

1. 维护平台特性文档

在 docs/zh-CN/platform-guide/ 下创建 {platform}.md 文档(其它语言版本会自动生成)：

# 平台名称适配器文档

## 基本信息
- 对应模块版本: 1.0.0
- 维护者: Your Name

## 支持的消息发送类型
...

## 特有事件类型
...

## 配置选项
...

2. 更新版本信息

发布新版本时，更新文档中的版本信息：

[project]
version = "2.0.0"  # 更新版本号

相关文档

• 适配器开发入门 - 创建第一个适配器
• 适配器核心概念 - 了解适配器架构
• SendDSL 详解 - 学习消息发送