適配器開發最佳實踐

本文檔提供了 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: 不支援的動作
10003: 錯誤的參數

# 2xxxx - 動作處理器錯誤
20001: 錯誤的處理器
20002: 內部處理器錯誤

# 3xxxx - 動作執行錯誤
31000: 資料庫錯誤
32000: 檔案系統錯誤
33000: 網路錯誤
34000: 平台錯誤
35000: 邏輯錯誤

多帳戶支援

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-TW/platform-guide/ 下建立 {platform}.md 文件（其他語言版本會自動產生）：

# 平台名稱適配器文件

## 基本資訊
- 對應模組版本: 1.0.0
- 維護者: Your Name

## 支援的消息傳送類型
...

## 特有事件類型
...

## 配置選項
...

2. 更新版本資訊

發布新版本時，更新文件中的版本資訊：

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

相關文件

• 適配器開發入門 - 建立第一個適配器
• 適配器核心概念 - 瞭解適配器架構
• SendDSL 詳解 - 學習訊息傳送