適配器系統 API

本文檔詳細介紹了 ErisPulse 適配器系統的 API。

Adapter 管理器

獲取適配器

from ErisPulse import sdk

# 透過名稱獲取適配器
adapter = sdk.adapter.get("platform_name")

# 或者也可以直接透過屬性存取
adapter = sdk.adapter.platform_name

使用適配器事件監聽

一般情況下，更建議使用Event模組進行事件的監聽/處理;

同時Event模組提供了強大的包裝器，可以為您的模組開發帶來更多便利

# 監聽 OneBot12 標準事件
@sdk.adapter.on("message")
async def handle_message(event):
    pass

# 監聽特定平台的標準事件
@sdk.adapter.on("message", platform="yunhu")
async def handle_yunhu_message(event):
    pass

# 監聽平台原生事件
@sdk.adapter.on("raw_event", raw=True, platform="yunhu")
async def handle_raw_event(data):
    pass

適配器管理

# 獲取所有平台
platforms = sdk.adapter.platforms

# 檢查適配器是否存在
exists = sdk.adapter.exists("platform_name")

# 啟用/禁用適配器
sdk.adapter.enable("platform_name")
sdk.adapter.disable("platform_name")

# 啟動/關閉適配器
# 以下方法都只展示了傳入參數的情況，無參數時代表啟動/停止全部已註冊適配器
await sdk.adapter.startup(["platform1", "platform2"])
await sdk.adapter.shutdown(["platform1", "platform2"])

# 檢查適配器是否正在運行
is_running = sdk.adapter.is_running("platform_name")

# 列出所有正在運行的適配器
running = sdk.adapter.list_running()

中間件

中間件在事件分發到處理器之前執行，可以對事件資料進行修改、過濾或記錄。

註冊中間件

@sdk.adapter.middleware
async def my_middleware(event):
    sdk.logger.info(f"中間件處理: {event}")
    return event

中間件執行模型

• 執行順序：中間件按註冊順序執行（先註冊先執行）
• 資料傳遞：每個中間件接收上一個中間件返回的 event 資料；如果某個中間件返回 None，則忽略該返回值並保留原資料繼續傳遞（同時輸出 warning 級別日誌）
• 修改資料：中間件可以修改事件資料並返回修改後的字典
• 事件否決：中間件顯式返回 False 時否決事件——事件被丟棄，不進入任何處理器、無任何出站副作用；否決時輸出 TRACE 日誌並觸發 adapter.event.blocked 生命周期鉤子（攜帶中間件名與完整事件）

@sdk.adapter.middleware
async def add_timestamp(event):
    event["processed_at"] = time.time()
    return event

@sdk.adapter.middleware
async def filter_spam(event):
    if event.get("detail_type") == "private":
        text = event.get("alt_message", "")
        if "垃圾廣告" in text:
            return False  # 否決：事件被丟棄，不進入任何處理器
    return event

注意：只有顯式返回 False 才否決事件（返回空字典 / 0 / "" 等 falsy 值不否決）； 返回 None 仍然是放行且載荷不變。否決後的事件可透過監聽 adapter.event.blocked 鉤子進行審計與排查"事件為什麼沒響應"。

Send 消息發送

基本發送

# 獲取適配器
adapter = sdk.adapter.get("platform")

# 發送文字訊息
await adapter.Send.To("user", "123").Text("Hello")

# 發送圖片訊息
await adapter.Send.To("group", "456").Image("https://example.com/image.jpg")

指定發送帳號

# 使用帳號名
await adapter.Send.Using("account1").To("user", "123").Text("Hello")

# 使用帳號 ID
await adapter.Send.Using("bot_id").To("user", "123").Text("Hello")

查詢支援的發送方法

# 列出平台支援的所有發送方法
methods = sdk.adapter.list_sends("onebot11")
# 返回: ["Text", "Image", "Voice", "Markdown", ...]

# 獲取某個方法的詳細資訊
info = sdk.adapter.send_info("onebot11", "Text")
# 返回:
# {
#     "name": "Text",
#     "parameters": [
#         {"name": "text", "type": "str", "default": null, "annotation": "str"}
#     ],
#     "return_type": "Awaitable[Any]",
#     "docstring": "發送文字訊息..."
# }

鏈式修飾

# @用戶
await adapter.Send.To("group", "456").At("789").Text("你好")

# @全體成員
await adapter.Send.To("group", "456").AtAll().Text("大家好")

# 回覆訊息
await adapter.Send.To("group", "456").Reply("msg_id").Text("回覆內容")

# 組合使用
await adapter.Send.To("group", "456").At("789").Reply("msg_id").Text("回覆@的訊息")

API 調用

call_api 方法

注意：call_api 是直接調用平台原生 API 的底層方法，各平台的參數和返回值可能不同，請參考對應平台適配器文件。推薦使用 Send DSL 發送訊息，僅在 Send DSL 不支援的場景（如獲取平台特有的資料、調用平台管理介面等）中使用 call_api。

# 調用平台 API
result = await adapter.call_api(
    endpoint="/send",
    content="Hello",
    recvId="123",
    recvType="user"
)

# 標準化回應
{
    "status": "ok",
    "retcode": 0,
    "data": {...},
    "message_id": "msg_id",
    "message": "",
    "{platform}_raw": raw_response
}

適配器基類

BaseAdapter 方法

from ErisPulse import sdk
from ErisPulse.Core import BaseAdapter

class MyAdapter(BaseAdapter):
    def __init__(self):
        super().__init__()
        self.sdk = sdk
        # 初始化適配器
        pass
    
    async def start(self):
        """啟動適配器（必須實現）"""
        pass
    
    async def shutdown(self):
        """關閉適配器（必須實現）"""
        pass
    
    async def call_api(self, endpoint: str, **params):
        """調用平台 API（必須實現）"""
        pass

Send 嵌套類

class MyAdapter(BaseAdapter):
    class Send(BaseAdapter.Send):
        def Text(self, text: str):
            """發送文字訊息"""
            import asyncio
            return asyncio.create_task(
                self._adapter.call_api(
                    endpoint="/send",
                    content=text,
                    recvId=self._target_id,
                    recvType=self._target_type
                )
            )

Bot 狀態管理

適配器透過發送 OneBot12 標準的 meta 事件來告知框架 Bot 的連接狀態。系統自動從中提取 Bot 資訊進行狀態追蹤。

meta 事件類型

適配器應發送以下三種 meta 事件：

self 字段擴展

ErisPulse 在 OneBot12 標準的 self 字段上擴展了以下可選字段：

meta 事件格式

connect — 連接上線

await adapter.emit({
    "id": "unique_id",
    "time": 1712345678,
    "type": "meta",
    "detail_type": "connect",
    "platform": "telegram",
    "self": {
        "platform": "telegram",
        "user_id": "123456",
        "user_name": "MyBot",
        "avatar": "https://example.com/avatar.jpg"
    },
    "telegram_raw": {...},
    "telegram_raw_type": "bot_connected"
})

系統處理：註冊 Bot，標記為 online，觸發 adapter.bot.online 生命周期事件。

heartbeat — 心跳

await adapter.emit({
    "id": "unique_id",
    "time": 1712345708,
    "type": "meta",
    "detail_type": "heartbeat",
    "platform": "telegram",
    "self": {
        "platform": "telegram",
        "user_id": "123456"
    }
})

系統處理：更新 last_active 時間（心跳中也支援更新元資訊）。

disconnect — 斷開連接

await adapter.emit({
    "id": "unique_id",
    "time": 1712345738,
    "type": "meta",
    "detail_type": "disconnect",
    "platform": "telegram",
    "self": {
        "platform": "telegram",
        "user_id": "123456"
    }
})

系統處理：標記 Bot 為 offline，觸發 adapter.bot.offline 生命周期事件。

普通事件的自動發現

除了 meta 事件外，普通事件（message/notice/request）中的 self 字段也會自動發現並註冊 Bot、更新活躍時間。這意味著即使適配器不發送 connect 事件，框架也能從第一條普通事件中發現 Bot。

適配器接入示例

class MyAdapter(BaseAdapter):
    async def start(self):
        # 與平台建立連接...
        connection = await self._connect()
        
        # 連接成功，發送 connect 事件
        await adapter.emit({
            "id": str(uuid4()),
            "time": int(time.time()),
            "type": "meta",
            "detail_type": "connect",
            "platform": "myplatform",
            "self": {
                "platform": "myplatform",
                "user_id": self.bot_id,
                "user_name": self.bot_name,
                "avatar": self.bot_avatar
            },
            "myplatform_raw": raw_data,
            "myplatform_raw_type": "connected"
        })
    
    async def on_disconnect(self):
        # 斷開連接，發送 disconnect 事件
        await adapter.emit({
            "id": str(uuid4()),
            "time": int(time.time()),
            "type": "meta",
            "detail_type": "disconnect",
            "platform": "myplatform",
            "self": {
                "platform": "myplatform",
                "user_id": self.bot_id
            }
        })

查詢 Bot 狀態

# 獲取所有適配器與 Bot 的完整狀態（WebUI 友好）
summary = sdk.adapter.get_status_summary()
# {
#     "adapters": {
#         "telegram": {
#             "status": "started",
#             "bots": {
#                 "123456": {
#                     "status": "online",
#                     "last_active": 1712345678.0,
#                     "info": {"nickname": "MyBot"}
#                 }
#             }
#         }
#     }
# }

# 列出所有 Bot
all_bots = sdk.adapter.list_bots()

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

# 獲取單個 Bot 詳情
info = sdk.adapter.get_bot_info("telegram", "123456")

# 檢查 Bot 是否在線
if sdk.adapter.is_bot_online("telegram", "123456"):
    print("Bot 在線")

Bot 狀態值

生命周期事件

# 監聽 Bot 上線事件
@sdk.lifecycle.on("adapter.bot.online")
def on_bot_online(event):
    print(f"Bot 上線: {event['data']['platform']}/{event['data']['bot_id']}")

# 監聽適配器狀態變化
@sdk.lifecycle.on("adapter.status.change")
def on_status_change(event):
    print(f"適配器狀態: {event['data']['platform']} -> {event['data']['status']}")

系統關閉時（shutdown），所有 Bot 會自動被標記為 offline。

相關文件

• 核心模組 API - 核心模組 API
• 事件系統 API - Event 模組 API
• 適配器開發指南 - 開發平台適配器