適配器核心概念

了解 ErisPulse 適配器的核心概念是開發適配器的基礎。

適配器架構

組件關係

正向轉換（接收方向）                           反向轉換（發送方向）
─────────────────                           ─────────────────
                                             
┌──────────────────┐                        ┌──────────────────┐
│ 平台原生事件     │                        │ 模組構建訊息     │
└────────┬─────────┘                        └────────┬─────────┘
         │                                           │
         ↓                                           ↓
┌──────────────────┐   ┌──────────────────┐   ┌──────────────────┐
│                  │   │ 適配器 (MyAdapter) │   │ Send.Raw_ob12()  │
│  Converter       │   │ ┌──────────────┐ │   │ (反向轉換入口)   │
│  (事件轉換器)    │──→│ │              │ │   │                  │
│                  │   │ │              │ │   │                  │
└──────────────────┘   │ └──────────────┘ │   └────────┬─────────┘
                       └──────────────────┘            │
                                │                      ↓
                                ↓              ┌──────────────────┐
                       ┌──────────────────┐    │ 平台 API 調用    │
                       │ OneBot12 標準事件 │    └────────┬─────────┘
                       └────────┬─────────┘             │
                                │                      ↓
                                ↓              ┌──────────────────┐
                       ┌──────────────────┐    │ 標準回應格式     │
                       │ 事件系統         │    └──────────────────┘
                       └────────┬─────────┘
                                │
                                ↓
                       ┌──────────────────┐
                       │ 模組 (處理事件)  │
                       └──────────────────┘

核心對稱性：

正向轉換（Converter）：平台原生事件 → OneBot12 標準事件，原始資料保留在 {platform}_raw
反向轉換（Raw_ob12）：OneBot12 消息段 → 平台 API 調用，回傳標準回應格式

AdapterManager 適配器管理器

AdapterManager 是 ErisPulse 適配器系統的核心元件，負責管理所有平台適配器的註冊、啟動、關閉和事件分發。

核心功能

適配器註冊：註冊和管理多個平台適配器
生命週期管理：控制適配器的啟動和關閉
事件分發：分發 OneBot12 標準事件和平台原生事件
配置管理：管理適配器的啟用/禁用狀態
中介層支援：支援 OneBot12 事件中介層

基本使用

from ErisPulse import sdk

# 註冊適配器（通常由 Loader 自動完成）
sdk.adapter.register("myplatform", MyPlatformAdapter)

# 啟動所有適配器
await sdk.adapter.startup()

# 啟動指定適配器
await sdk.adapter.startup(["myplatform"])
# 啟動全部適配器
await sdk.adapter.startup()

# 獲取適配器實例
my_adapter = sdk.adapter.get("myplatform")
# 或透過屬性存取
my_adapter = sdk.adapter.myplatform

# 關閉所有適配器
await sdk.adapter.shutdown()

啟動和關閉

啟動適配器

# 啟動所有已註冊的適配器
await sdk.adapter.startup()

# 啟動指定平台
await sdk.adapter.startup(["platform1", "platform2"])

啟動流程：

提交 adapter.start 生命週期事件
提交 adapter.status.change 事件（starting）
並行啟動各個適配器
如果啟動失敗，自動重試（指數退避策略）
啟動成功後提交 adapter.status.change 事件（started）

重試機制：

前 4 次重試：60秒、10分鐘、30分鐘、60分鐘
第 5 次及以後：3 小時固定間隔

關閉適配器

# 關閉所有適配器
await sdk.adapter.shutdown()

關閉流程：

提交 adapter.stop 生命週期事件
呼叫所有適配器的 shutdown() 方法
關閉路由伺服器
清空事件處理器
提交 adapter.stopped 生命週期事件

配置管理

檢查平台狀態

# 檢查平台是否已註冊
exists = sdk.adapter.exists("myplatform")

# 檢查平台是否啟用
enabled = sdk.adapter.is_enabled("myplatform")

# 使用 in 操作符
if "myplatform" in sdk.adapter:
    print("平台存在且已啟用")

列出平台

# 列出所有已註冊的平台
platforms = sdk.adapter.list_registered()

# 列出所有平台及其狀態
status_dict = sdk.adapter.list_items()
# 回傳: {"platform1": true, "platform2": false, ...}

# 獲取已啟用的平台列表
enabled_platforms = [p for p, enabled in status_dict.items() if enabled]

事件監聽

OneBot12 標準事件

from ErisPulse import sdk

# 監聽所有平台的標準訊息事件
@sdk.adapter.on("message")
async def handle_message(data):
    print(f"收到OneBot12訊息: {data}")

# 監聽特定平台的標準訊息事件
@sdk.adapter.on("message", platform="myplatform")
async def handle_platform_message(data):
    print(f"收到 myplatform 訊息: {data}")

# 監聽所有事件
@sdk.adapter.on("*")
async def handle_any_event(data):
    print(f"收到事件: {data.get('type')}")

平台原生事件

# 監聽特定平台的原生事件
@sdk.adapter.on("raw_event_type", raw=True, platform="myplatform")
async def handle_raw_event(data):
    print(f"收到原生事件: {data}")

# 監聽所有平台的原生事件（通配符）
@sdk.adapter.on("*", raw=True)
async def handle_all_raw_events(data):
    print(f"收到原生事件: {data}")

事件分發機制

當呼叫 adapter.emit(event_data) 時：

中介層處理：先執行所有 OneBot12 中介層
標準事件分發：分發到匹配的 OneBot12 事件處理器
原生事件分發：如果存在原始資料，分發到原生事件處理器

匹配規則：

精確匹配：@sdk.adapter.on("message") 只匹配 message 事件
通配符：@sdk.adapter.on("*") 匹配所有事件
平台過濾：platform="myplatform" 只分發指定平台的事件

中介層

添加中介層

@sdk.adapter.middleware
async def logging_middleware(data):
    """日誌記錄中介層"""
    print(f"處理事件: {data.get('type')}")
    return data  # 必須回傳資料

@sdk.adapter.middleware
async def filter_middleware(data):
    """事件過濾中介層"""
    # 過濾不需要的事件
    if data.get("type") == "notice":
        return None  # 回傳 None 時中介層鏈會忽略該回傳值，保留原資料繼續傳遞
    return data  # 必須回傳資料以繼續傳遞

中介層回傳契約

回傳值	行為
`dict`	改寫事件載入（後續處理器收到改寫後的事件）
`None`	放行，載入不變（輸出 WARNING 提示——建議顯式 `return data`）
`False`	否決：事件被丟棄，不進入任何處理器、無任何出站副作用

否決適用於防火牆、限流、黑名單等"在事件層面直接丟棄"的場景（此前只能用高優先級事件處理器繞行實現）。否決時框架輸出 TRACE 日誌並觸發 adapter.event.blocked 生命週期鉤子（攜帶 middleware 中介層名、完整 event、platform / event_type / detail_type），便於排查"事件為什麼沒回應"：

@sdk.adapter.middleware
async def rate_limit_middleware(data):
    """限流中介層"""
    if _is_rate_limited(data):
        return False  # 否決：事件被丟棄
    data["rate_marked"] = True
    return data

@sdk.lifecycle.on("adapter.event.blocked")
async def on_event_blocked(data):
    print(f"事件被 {data['middleware']} 否決: {data['event_type']}")

中介層執行順序

中介層按照註冊順序執行，後註冊的中介層先執行。

注意：如果中介層回傳 None（例如忘記 return data），框架會忽略該回傳值並保留原資料繼續傳遞，同時輸出 warning 級別日誌。這確保了單個中介層的失誤不會導致整個事件鏈中斷。

# 註冊順序
sdk.adapter.middleware(middleware1)  # 最後執行
sdk.adapter.middleware(middleware2)  # 中間執行
sdk.adapter.middleware(middleware3)  # 最先執行

# 執行順序：middleware3 -> middleware2 -> middleware1

獲取適配器實例

get() 方法

adapter = sdk.adapter.get("myplatform")
if adapter:
    await adapter.Send.To("user", "123").Text("Hello")

屬性存取

# 透過屬性名存取（不區分大小寫）
adapter = sdk.adapter.myplatform
await adapter.Send.To("user", "123").Text("Hello")

BaseAdapter 基類

基本結構

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

@dataclass
class MyConfig(BaseConfig):
    """適配器配置（聲明後框架自動管理）"""
    token: str = field(
        default="",
        metadata={
            "description": {"i18n": "my_adapter.token", "default": "Bot Token"},
            "required": True,
            "secret": True,
            "ui": {"widget": "password", "group": "basic", "order": 1},
        },
    )

class MyAdapter(BaseAdapter):
    ConfigClass = MyConfig  # 聲明配置類
    
    # 無需覆寫 __init__，框架自動處理：
    # - self.sdk, self.logger
    # - self.cfg（類型安全的配置實例，實時讀取）
    # - self.Send, self.Request
    
    async def start(self):
        """啟動適配器（必須實現）"""
        cfg = self.cfg  # 自動加載的類型安全配置
        pass
    
    async def shutdown(self):
        """關閉適配器（必須實現）"""
        pass
    
    async def call_api(self, endpoint: str, **params):
        """呼叫平台 API（必須實現）"""
        pass

配置管理

框架提供了宣告式配置管理，透過 dataclass 定義配置結構，框架自動處理加載、驗證和範本生成。

單帳戶配置

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

@dataclass
class TelegramConfig(BaseConfig):
    token: str = field(default="", metadata={
        "description": {"i18n": "telegram.token", "default": "Bot Token"},
        "required": True,
        "secret": True,
        "ui": {"widget": "password", "group": "basic", "order": 1},
    })
    proxy: str = field(default="", metadata={
        "description": {"i18n": "telegram.proxy", "default": "代理地址"},
        "ui": {"widget": "text", "group": "advanced", "order": 10},
    })

class TelegramAdapter(BaseAdapter):
    ConfigClass = TelegramConfig
    
    async def start(self):
        cfg = self.cfg  # 類型安全，實時讀取
        if not cfg.token:
            raise ValueError("未配置 Token")
        await self._connect(cfg.token, proxy=cfg.proxy)

多帳戶配置

BotAccountConfig 基類提供 enabled 和 name 欄位。絕大多數適配器能從平台協議或登入回應中自動取得 bot_id，在事件轉換時注入到帳戶配置中。：

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

# 大多數適配器：bot_id 運行時自動取得，無需配置
@dataclass
class MyBotConfig(BotAccountConfig):
    token: str = field(default="", metadata={
        "description": {"i18n": "my_adapter.bot_token", "default": "Token"},
        "required": True,
    })

# 如果登入時無法取得 bot_id，可以讓使用者在配置中填寫
@dataclass
class YunhuBotConfig(BotAccountConfig):
    bot_id: str = field(default="", metadata={
        "description": {"i18n": "yunhu.bot_id", "default": "機器人ID"},
        "required": True,
    })
    token: str = field(default="", metadata={
        "description": {"i18n": "yunhu.token", "default": "Token"},
        "required": True,
    })

class MyAdapter(BaseAdapter):
    AccountConfigClass = MyBotConfig
    
    async def start(self):
        for name, account in self.enabled_accounts.items():
            user_id = await self._login(name, account)
            await self.emit_meta("connect", user_id)

metadata 約定

欄位 metadata 同時服務於 TOML 注釋生成和 WebUI 表單渲染：

metadata = {
    "description": str | dict,  # 欄位描述（支援 i18n）
    "required": bool,         # 是否必填（驗證 + WebUI 必填標記）
    "secret": bool,           # 是否敏感（WebUI 顯示為 ***，日誌中脫敏）
    "example": bool,          # 不落盤標誌：不寫入 config.toml（預設值/範本均排除），
                              # 僅渲染進 config.full.example；schema 帶 "example": true 標記，
                              # CLI 配置向導預設跳過；使用者手動設定後正常持久化
    "min": number, "max": number,  # 數值範圍驗證
    "ui": {                   # WebUI 控件配置（舊名 "webui" 仍兼容）
        "widget": str,        # 控件類型: "text" | "switch" | "select" | "number" | "password"
        "group": str,         # 分組: "basic" | "advanced" | "connection" 等
        "order": int,         # 排序權重（越小越靠前）
        "options": list,      # select 控件的可選項 [{label, value}]，label 支援 i18n
        "placeholder": str | dict,  # 輸入框佔位符（支援 i18n）
    },
    "extra": dict,            # 額外擴展欄位（透傳到 schema）
}

所有使用者可見的字串欄位均支援 i18n，統一採用 {"i18n": "key", "default": "文字"} 格式，純字串則原樣透傳（向後相容）。支援的 i18n 欄位：

欄位	位置	說明
`description`	field metadata	欄位描述
`options[].label`	`ui.options`	select 控件選項標籤
`placeholder`	`ui.placeholder`	輸入框佔位符
`group_labels`	`_schema_meta`	分組顯示名（Dashboard 分區標題）

使用 i18n 時，需提前將翻譯鍵註冊到 i18n 系統（詳見 i18n 文檔）。

description / placeholder / options label 示例：

token: str = field(
    default="",
    metadata={
        "description": {"i18n": "my_adapter.token", "default": "Bot Token"},
        "ui": {
            "widget": "text",
            "placeholder": {"i18n": "my_adapter.token.ph", "default": "請輸入 Token"},
        },
    },
)
mode: str = field(
    default="a",
    metadata={
        "description": {"i18n": "my_adapter.mode", "default": "模式"},
        "ui": {
            "widget": "select",
            "options": [
                {"label": {"i18n": "my_adapter.mode.a", "default": "選項A"}, "value": "a"},
                {"label": "純字串標籤", "value": "b"},  # 純字串原樣透傳
            ],
        },
    },
)

group_labels 示例（在配置類定義後聲明）：

MyConfig._schema_meta = {
    "group_labels": {
        "basic": {"i18n": "my_adapter.group.basic", "default": "基本設定"},
        "advanced": {"i18n": "my_adapter.group.advanced", "default": "高級設定"},
    }
}

框架的 resolve_config_schema() 會根據目前語言自動解析上述所有欄位的 i18n 鍵； get_config_schema() 則原樣透傳 i18n 字典，由前端自行解析。

docstring 自动生成欄位描述（v2.8.0+）

未在 metadata 中聲明 description 的欄位，框架會自動從配置類 docstring 中提取欄位說明作為兜底，支援兩種常見風格（可混用）：

@dataclass
class MyConfig(BaseConfig):
    """
    MyAdapter 配置

    :ivar endpoint: 平台 API 地址        # reST 風格
    :ivar timeout: 請求超時秒數
    """

    endpoint: str = "https://api.example.com"   # 無 metadata description → 注釋/描述取 docstring
    timeout: int = 30

    # Google 風格同樣支援（Attributes: 段）：
    # Attributes:
    #     endpoint: 平台 API 地址

優先級：metadata description > docstring 欄位說明 > 空。 i18n 字典形式的 description 不受影響（始終優先）。

嵌套配置（v2.8.0+）

欄位類型為嵌套 dataclass 時，框架遞迴處理：schema 以 "type": "table" + "fields" 子樹承載（WebUI 渲染為可摺疊嵌套分組），TOML 範本渲染為 [子表] 節，預設值 / 填充 / 驗證 / i18n 解析均遞迴生效。

@dataclass
class RetryConfig(BaseConfig):
    """重試策略

    :ivar max_retries: 最大重試次數
    """
    max_retries: int = 3
    backoff: float = 0.5

@dataclass
class MyConfig(BaseConfig):
    """MyAdapter 配置"""
    endpoint: str = "https://api.example.com"
    retry: RetryConfig = field(default_factory=RetryConfig)   # 嵌套配置段

生成的 TOML 範本：

endpoint = "https://api.example.com"

[retry]
# 最大重試次數
max_retries = 3
backoff = 0.5

嵌套類型建議使用直接類型註解；字串註解（如延遲求值場景）需保證類型可從配置類所在模組全域、__qualname__ 外層類命名空間或類屬性中按名解析。

不落盤的 example 欄位（v2.8.0+）

gc_interval: int = field(default=300, metadata={"example": True})

帶 example: True 的欄位：

不寫入 config.toml（適配器/模組配置範本與預設值均排除，運行時走代碼預設值）
僅渲染進專案內 config.full.example（供使用者參考，按需手動複製到 config.toml）
schema 中帶 "example": true 標記（面板可自行決定展示策略），CLI 配置向導預設跳過
使用者手動設定該鍵後正常持久化、正常熱更新（使用者顯式意圖優先）

適合"繁雜又很少觸碰"的高級配置項目，保持使用者的 config.toml 最小化。

⚠️ _schema_meta 是類級元資料（非配置欄位）。若在 dataclass 類體內部聲明，必須加 ClassVar 註解（_schema_meta: ClassVar[dict] = {...}），否則會被 dataclass 視為普通欄位。框架對下劃線前綴欄位已做防禦性排除（不進入任何 schema / 範本 / 預設值 / 驗證輸出），但仍建議規範聲明。

宣告式翻譯鍵（v2.7.0+）

適配器可以像宣告 ConfigClass 一樣，透過嵌套類 I18nClass 集中宣告翻譯鍵。框架會在 __init__ 階段（配置範本生成之前）自動註冊所有宣告的翻譯鍵，確保配置描述中引用的 i18n 鍵在生成範本時已可用。

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

class MyAdapter(BaseAdapter):
    class I18nClass(BaseI18n):
        endpoint: I18nKey = I18nKey(
            default="API Endpoint",
            zh_CN="API 地址",
            zh_TW="API 位址",
            en="API Endpoint",
            ja="APIアドレス",
            ru="API адрес",
        )
        token: I18nKey = I18nKey(
            default="Platform Token",
            zh_CN="平台 Token",
            zh_TW="平台權杖",
            en="Platform Token",
            ja="プラットフォームトークン",
            ru="Токен платформы",
        )

I18nKey.default 是語言無關的兜底文本，不會註冊到任何語言。要讓翻譯生效，必須顯式傳入至少一個語言參數。

詳細用法（鍵路徑規則、顯式 key 參數等）見 i18n 文檔。

宣告式事件擴展方法（v2.7.0+）

適配器可以透過 EventMixin 集中宣告平台特有的事件擴展方法，框架自動註冊到當前平台。

from ErisPulse.Core import BaseAdapter

class MyAdapter(BaseAdapter):
    class EventMixin:
        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)

註冊後，事件物件直接呼叫這些方法：

@message.on_group_message()
async def handler(event):
    if event.is_official_message():
        chat_name = event.get_chat_name()
        await event.reply(f"[{chat_name}] 官方訊息已收到")

適配器的事件擴展方法註冊到自身平台（self._platform）。模組如需跨平台事件擴展，請使用原有的 register_event_mixin() API。

帳戶解析

多帳戶適配器可使用 _resolve_account() 自動解析目標帳戶：

async def call_api(self, endpoint: str, **params):
    account_id = params.pop("account_id", None)
    name, account = self._resolve_account(account_id)
    # name: 帳戶名, account: 配置實例

解析策略：帳戶名匹配 → bot_id 欄位匹配 → 其他 str 欄位匹配 → 第一個啟用帳戶。

配置熱更新

子類可覆寫 on_config_update() 回應配置變更：

class MyAdapter(BaseAdapter):
    ConfigClass = MyConfig
    
    def on_config_update(self, old_config, new_config):
        if old_config.token != new_config.token:
            self.logger.info("Token 已更新，將重新連接")

初始化過程

框架在 BaseAdapter.__init__(self, sdk=None) 中自動完成以下工作：

SDK 引用：設定 self.sdk、self.logger
Send/Request 工廠：建立 self.Send 和 self.Request
配置範本：如果宣告了 ConfigClass，自動生成預設配置範本（首次）
帳戶範本：如果宣告了 AccountConfigClass，自動生成預設帳戶範本（首次）
EventMixin 註冊：如果宣告了 EventMixin，在 AdapterManager 注入平台名後自動註冊

配置透過 self.cfg / self.accounts 實時讀取（每次存取都從配置儲存讀取最新值）。self.config 作為 self.cfg 的相容別名仍可使用。

大多數適配器無需覆寫 __init__。如需自訂初始化：

class MyAdapter(BaseAdapter):
    ConfigClass = MyConfig
    
    def __init__(self, sdk=None):
        super().__init__(sdk)  # 傳入 sdk
        self.converter = self._setup_converter()
        self.convert = self.converter.convert

Send 消息發送 DSL

繼承關係

class MyAdapter(BaseAdapter):
    class Send(BaseAdapter.Send):
        """Send 嵌套類，繼承自 BaseAdapter.Send"""
        pass

可用屬性

Send 類在呼叫時會自動設定以下屬性：

屬性	說明	設定方式
`_target_id`	目標ID	`To(id)` 或 `To(type, id)`
`_target_type`	目標類型	`To(type, id)`
`_target_to`	簡化目標ID	`To(id)`
`_account_id`	發送帳戶ID	`Using(account_id)`
`_adapter`	適配器實例	自動設定
`_at_user_ids`	@使用者列表	`At(user_id)`
`_reply_message_id`	回覆的訊息ID	`Reply(message_id)`
`_at_all`	是否@全體	`AtAll()`

推薦：使用 self.send_context 屬性一次性取得 target_type、target_id、account_id，比直接存取實例變數更清晰。

框架輔助方法

方法/屬性	說明
`self._apply_modifiers(message)`	將 At/AtAll/Reply 修飾器狀態合併到訊息段列表
`self.send_context`	回傳 `{target_type, target_id, account_id}` 字典

基本方法

適配器只需實現 Raw_ob12，標準方法（Text/Image/Voice/Video/File）已從 SendDSL 基類繼承並預設委派給它：

class Send(BaseAdapter.Send):
    def Raw_ob12(self, message, **kwargs):
        """必須實現：OneBot12 消息段 → 平台 API"""
        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 已從基類繼承，自動委派 Raw_ob12，無需重複實現
    # 如需平台特定邏輯，可覆寫單個方法：
    # def Text(self, text: str):
    #     return self.Raw_ob12([{"type": "text", "data": {"text": text}}])

鏈式修飾方法

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

事件轉換器

轉換流程

平台原始事件
    ↓
Converter.convert()
    ↓
OneBot12 標準事件

必需欄位

所有轉換後的事件必須包含：

{
    "id": "事件唯一標識",
    "time": 1234567890,           # 10位 Unix 時間戳
    "type": "message/notice/request/meta",
    "detail_type": "事件詳細類型",
    "platform": "平台名稱",
    "self": {
        "platform": "平台名稱",
        "user_id": "機器人ID"     # 必須與 bot_id 一致
    },
    "{platform}_raw": {...},       # 原始資料（必須）
    "{platform}_raw_type": "..."    # 原始類型（必須）
}

轉換器示例

class MyPlatformConverter:
    def convert(self, raw_event):
        """將平台原生事件轉換為 OneBot12 標準格式"""
        if not isinstance(raw_event, dict):
            return None
        
        # 產生事件 ID
        event_id = raw_event.get("event_id") or str(uuid.uuid4())
        
        # 轉換時間戳
        timestamp = raw_event.get("timestamp")
        if timestamp and timestamp > 10**12:
            timestamp = int(timestamp / 1000)
        else:
            timestamp = int(timestamp) if timestamp else int(time.time())
        
        # 轉換事件類型
        event_type = self._convert_type(raw_event.get("type"))
        detail_type = self._convert_detail_type(raw_event)
        
        # 建構標準事件
        onebot_event = {
            "id": str(event_id),
            "time": timestamp,
            "type": event_type,
            "detail_type": detail_type,
            "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

連接管理

WebSocket 連接

class MyAdapter(BaseAdapter):
    async def start(self):
        """註冊 WebSocket 路由"""
        router.register_websocket(
            module_name="myplatform",
            path="/ws",
            handler=self._ws_handler,
            auth_handler=self._auth_handler
        )
    
    async def _ws_handler(self, websocket):
        """WebSocket 連接處理器"""
        self.connection = websocket
        
        try:
            while True:
                data = await websocket.receive_text()
                onebot_event = self.convert(data)
                if onebot_event:
                    await self.adapter.emit(onebot_event)
        except WebSocketDisconnect:
            self.logger.info("連接已斷開")
        finally:
            self.connection = None
    
    async def _auth_handler(self, websocket) -> bool:
        """WebSocket 認證"""
        token = websocket.query_params.get("token")
        return token == "valid_token"

WebHook 連接

class MyAdapter(BaseAdapter):
    async def start(self):
        """註冊 WebHook 路由"""
        router.register_http_route(
            module_name="myplatform",
            path="/webhook",
            handler=self._webhook_handler,
            methods=["POST"]
        )
    
    async def _webhook_handler(self, request):
        """WebHook 請求處理器"""
        data = await request.json()
        onebot_event = self.convert(data)
        if onebot_event:
            await self.adapter.emit(onebot_event)
        return {"status": "ok"}

路由資訊查詢：適配器註冊的路由（HTTP、WebSocket、SSE）可以透過 sdk.adapter.get_connection_info(platform) 和 sdk.router.get_module_urls(module_name) 查詢完整連接位址（包含 base_url + 路徑）。詳見適配器開發入門 - 連接資訊與路由發現和 SSE 支援。

API 回應標準

框架提供 make_response() 和 make_error() 方法構造標準化回應，無需手動建構回應字典。

成功回應

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

手動建構回應（舊版方式仍然相容）

async def call_api(self, endpoint: str, **params):
    return {
        "status": "ok",
        "retcode": 0,
        "data": {...},
        "message_id": "msg_id",
        "message": "",
        "myplatform_raw": raw_response
    }

多帳戶支援

宣告式配置（推薦）

使用 AccountConfigClass 宣告配置類後，框架自動管理多帳戶加載、驗證和範本生成：

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

@dataclass
class MyBotConfig(BotAccountConfig):
    bot_id: str = field(default="", metadata={"description": "Bot ID", "required": True})
    token: str = field(default="", metadata={"description": "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}: {account.bot_id}")
            await self._connect(name, account)
    
    async def call_api(self, endpoint: str, **params):
        account_id = params.pop("account_id", None)
        name, account = self._resolve_account(account_id)
        # 使用 account.token, account.bot_id 等欄位

帳戶配置檔案

[MyAdapter.accounts.account1]
bot_id = "bot_001"
token = "token1"
enabled = true

[MyAdapter.accounts.account2]
bot_id = "bot_002"
token = "token2"
enabled = true

指定帳戶發送

# 使用 Using 方法指定帳戶
my_adapter = adapter.get("myplatform")

# 透過事件中的 self.user_id（推薦，最通用）
await my_adapter.Send.Using(event["self"]["user_id"]).To("user", "123").Text("Hello")

# 透過帳戶名
await my_adapter.Send.Using("account1").To("user", "123").Text("Hello")

self.user_id 與 Using 的關係

框架的事件回應機制會自動從事件的 self 欄位中提取 account_id（優先）或 user_id，作為 Using 參數傳入。適配器開發者需要確保 Converter 中 self.user_id 的值與 _resolve_account() 能夠正確匹配。

框架內部行為：

# 框架提取 bot_id 的邏輯
bot_id = self.get("self", {}).get("account_id", "") or self.get("self", {}).get("user_id", "")

# 僅在 bot_id 非空時呼叫 Using
if bot_id:
    send_chain = send_chain.Using(bot_id)

關鍵點：即使適配器只使用一個 Bot 配置，只要 Converter 正確設定了 self.user_id，框架就會將其作為 Using 參數傳入。適配器需確保 self.user_id 與 AccountConfigClass 中的標識欄位（如 bot_id）一致，使 _resolve_account() 能匹配到正確帳戶。如果 self.user_id 為空，框架不會呼叫 Using，此時 call_api 收到的 account_id 為 None，_resolve_account(None) 回傳第一個啟用的帳戶。

錯誤處理

連接重試

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()
                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}秒後重試")
                    await asyncio.sleep(wait_time)
                else:
                    raise

API 錯誤處理

async def call_api(self, endpoint: str, **params):
    try:
        # 推薦使用 SDK 內建客戶端
        from ErisPulse.Core import client
        from ErisPulse.Core.Bases.errors import ClientError, ClientTimeoutError
        resp = await client.post(
            f"https://api.platform.com/{endpoint}",
            json=params,
            max_retries=2,
        )
        response = await resp.json()
        return self._standardize_response(response)
    except ClientTimeoutError:
        self.logger.error(f"請求超時: {endpoint}")
        return self._error_response("請求超時", 32000)
    except ClientError as e:
        self.logger.error(f"網路錯誤: {e}")
        return self._error_response("網路請求失敗", 33000)
    except Exception as e:
        self.logger.error(f"未知錯誤: {e}")
        return self._error_response(str(e), 34000)

向後相容：直接使用 aiohttp.ClientSession 的舊適配器程式碼不受影響，仍然可以捕獲 aiohttp.ClientError。兩種方式可以共存。推薦新程式碼使用 sdk.client + ErisPulse 異常體系。

Bot 狀態管理

AdapterManager 內建了 Bot 狀態追蹤系統，自動維護所有已註冊 Bot 的線上狀態、活躍時間和元資訊。

自動發現機制

當適配器透過 adapter.emit() 發送事件時，框架會自動檢查事件中的 self 欄位：

meta 事件：根據 detail_type 執行對應操作（connect 註冊/斷開標記離線/heartbeat 更新活躍時間）
普通事件（message/notice/request）：自動發現 Bot 並更新活躍時間

# 所有包含 self 欄位的事件都會觸發自動發現
await self.adapter.emit({
    "type": "message",
    "platform": "myplatform",
    "self": {"platform": "myplatform", "user_id": "bot123"},
    # ...
})
# Bot "bot123" 已自動註冊（如果首次出現）並更新活躍時間

Meta 事件類型

`detail_type`	說明	框架行為
`connect`	Bot 連接	註冊 Bot 並觸發 `adapter.bot.online` 生命週期事件
`disconnect`	Bot 斷開	標記 Bot 離線並觸發 `adapter.bot.offline` 生命週期事件
`heartbeat`	Bot 心跳	更新 Bot 活躍時間和元資訊

適配器發送 Meta 事件

使用 emit_meta() 一行即可發送 meta 事件：

class MyAdapter(BaseAdapter):
    async def _on_bot_connect(self, bot_id: str):
        # 一行發送 connect 事件
        await self.emit_meta("connect", bot_id, user_name="MyBot", nickname="我的機器人")

    async def _on_bot_disconnect(self, bot_id: str):
        await self.emit_meta("disconnect", bot_id)

也支援手動建構（舊版方式仍然相容）：

await self.adapter.emit({
    "type": "meta",
    "detail_type": "connect",
    "platform": "myplatform",
    "self": {"platform": "myplatform", "user_id": bot_id}
})

`self` 欄位擴展資訊

self 欄位除必需的 platform 和 user_id 外，還支援以下可選欄位：

欄位	說明
`user_name`	Bot 使用者名
`nickname`	Bot 昵稱
`avatar`	Bot 頭像 URL
`account_id`	多帳戶標識

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": {...}}}}

監聽 Bot 生命週期

from ErisPulse import sdk

@sdk.lifecycle.on("adapter.bot.online")
async def on_bot_online(data):
    platform = data.get("platform")
    bot_id = data.get("bot_id")
    sdk.logger.info(f"Bot 上線: {platform}/{bot_id}")

@sdk.lifecycle.on("adapter.bot.offline")
async def on_bot_offline(data):
    platform = data.get("platform")
    bot_id = data.get("bot_id")
    sdk.logger.info(f"Bot 下線: {platform}/{bot_id}")

適配器核心概念

適配器架構

組件關係

AdapterManager 適配器管理器

核心功能

基本使用

啟動和關閉

啟動適配器

關閉適配器

配置管理

檢查平台狀態

列出平台

事件監聽

OneBot12 標準事件

平台原生事件

事件分發機制

中介層

添加中介層

中介層回傳契約

中介層執行順序

獲取適配器實例

get() 方法

屬性存取

BaseAdapter 基類

基本結構

配置管理

單帳戶配置

多帳戶配置

metadata 約定

docstring 自动生成欄位描述（v2.8.0+）

嵌套配置（v2.8.0+）

不落盤的 example 欄位（v2.8.0+）

宣告式翻譯鍵（v2.7.0+）

宣告式事件擴展方法（v2.7.0+）

帳戶解析

配置熱更新

初始化過程

Send 消息發送 DSL

繼承關係

可用屬性

框架輔助方法

基本方法

鏈式修飾方法

事件轉換器

轉換流程

必需欄位

轉換器示例

連接管理

WebSocket 連接

WebHook 連接

API 回應標準

成功回應

手動建構回應（舊版方式仍然相容）

多帳戶支援

宣告式配置（推薦）

帳戶配置檔案

指定帳戶發送

self.user_id 與 Using 的關係

錯誤處理

連接重試

API 錯誤處理

Bot 狀態管理

自動發現機制

Meta 事件類型

適配器發送 Meta 事件

self 欄位擴展資訊

Bot 狀態查詢

監聽 Bot 生命週期

相關文件

`self` 欄位擴展資訊