事件轉換器實現指南

事件轉換器 (Converter) 是適配器的核心組件之一，負責將平台原生事件轉換為 ErisPulse 統一的 OneBot12 標準事件格式。

Converter 職責

平台原生事件 ──→ Converter.convert() ──→ OneBot12 標準事件

Converter 只負責正向轉換（接收方向），即將平台的原生事件數據轉換為 OneBot12 標準格式。反向轉換（發送方向）由 Send.Raw_ob12() 方法處理。

核心原則

無損轉換：原始數據必須完整保留在 {platform}_raw 字段中
標準兼容：轉換後的事件必須符合 OneBot12 標準格式
平台擴展：平台特有數據使用 {platform}_ 前綴字段存儲

BaseConverter 基類（推薦）

從 2.7.0 起，框架提供 BaseConverter 基類（ErisPulse.Core.Bases），封裝 OneBot12 事件的公共字段建構與常用消息段輔助，讓轉換器只需聚焦類型映射：

from ErisPulse.Core.Bases import BaseConverter


class MyConverter(BaseConverter):
    def __init__(self):
        super().__init__(platform="myplatform")

    def convert(self, raw_event: dict) -> dict | None:
        if not isinstance(raw_event, dict):
            return None
        event_type = raw_event.get("type", "")
        base = self.build_base_event(raw_event, event_type)  # id/time/platform/self/raw
        if event_type == "message":
            base["type"] = "message"
            base["detail_type"] = "group" if raw_event.get("group_id") else "private"
            base["user_id"] = str(raw_event.get("sender_id", ""))
            base["message"] = [self.text(raw_event.get("content", ""))]
            base["alt_message"] = raw_event.get("content", "")
            return base
        return None

build_base_event() 已填充的公共字段：

字段	來源
`id`	`raw_event["event_id"]`，缺省自生成 UUID
`time`	`raw_event["timestamp"]`，缺省當前時間
`platform`	建構時傳入的 `platform`
`self`	`{"platform": ..., "user_id": raw_event["bot_id"]}`
`{platform}_raw`	原始事件（滿足"無損轉換"原則）
`{platform}_raw_type`	原始事件類型

常用消息段輔助方法（均為靜態方法，直接複用）：

converter.text("hi")          # {"type": "text", "data": {"text": "hi"}}
converter.at("123456")        # {"type": "at", "data": {"user_id": "123456"}}
converter.image("file.png")   # {"type": "image", "data": {"file": "file.png"}}

手動實現時 build_base_event 的公共字段建構是必須重複寫的樣板代碼，使用 BaseConverter 可省去這部分，且天然滿足"無損轉換"（原始事件始終進 {platform}_raw）。

convert() 方法

方法簽名

def convert(self, raw_event: dict) -> dict:
    """
    將平台原生事件轉換為 OneBot12 標準格式

    :param raw_event: 平台原生事件數據
    :return: OneBot12 標準格式事件字典
    """
    pass

返回值結構

轉換後的事件字典應包含以下標準字段：

{
    "id": "事件唯一ID",
    "time": 1234567890,           # Unix 時間戳（秒）
    "type": "message",             # 事件類型
    "detail_type": "private",      # 詳細類型
    "platform": "myplatform",      # 平台名稱
    "self": {
        "platform": "myplatform",
        "user_id": "bot_user_id"
    },

    # 消息事件字段
    "user_id": "sender_id",
    "message": [...],              # OneBot12 消息段列表
    "alt_message": "純文本內容",

    # 必須保留原始數據
    "myplatform_raw": { ... },     # 平台原生事件完整數據
    "myplatform_raw_type": "原生事件類型名",
}

必填字段映射

通用字段（所有事件類型）

OB12 字段	類型	說明
`id`	str	事件唯一標識符
`time`	int	Unix 時間戳（秒）
`type`	str	事件類型：`message` / `notice` / `request` / `meta`
`detail_type`	str	詳細類型：`private` / `group` / `friend` 等
`platform`	str	平台名稱，與適配器註冊名一致
`self`	dict	机器人信息：`{"platform": "...", "user_id": "..."}`

消息事件額外字段

OB12 字段	類型	说明
`user_id`	str	發送者 ID
`message`	list[dict]	OneBot12 消息段列表
`alt_message`	str	純文本備用內容

通知事件額外字段

OB12 字段	類型	说明
`user_id`	str	相關用戶 ID
`operator_id`	str	操作者 ID（如群成員變動）

消息段轉換

OneBot12 標準定義了以下消息段類型：

# 文本
{"type": "text", "data": {"text": "Hello"}}

# 圖片
{"type": "image", "data": {"file": "https://example.com/img.jpg"}}

# 音頻
{"type": "audio", "data": {"file": "https://example.com/audio.mp3"}}

# 視頻
{"type": "video", "data": {"file": "https://example.com/video.mp4"}}

# 文件
{"type": "file", "data": {"file": "https://example.com/doc.pdf"}}

# @提及
{"type": "mention", "data": {"user_id": "123"}}

# @全體
{"type": "mention_all", "data": {}}

# 回覆
{"type": "reply", "data": {"message_id": "msg_123"}}

如果平台有不支援的消息段類型，可以省略該段或轉換為最接近的標準類型。

平台擴展字段

平台特有的數據應使用 {platform}_ 前綴存儲，避免與標準字段衝突：

{
    # 標準字段
    "type": "message",
    "detail_type": "group",
    # ...

    # 平台擴展字段
    "myplatform_raw": { ... },          # 原始事件數據（必須）
    "myplatform_raw_type": "chat",      # 原始事件類型（必須）

    # 其他平台特有字段
    "myplatform_group_name": "群名稱",
    "myplatform_sender_role": "admin",
}

重要：{platform}_raw 字段是必須的，ErisPulse 的事件系統和模組可能依賴它來訪問平台原始數據。

完整示例

以下是一個完整的 Converter 實現：

class MyConverter:
    def __init__(self, platform: str):
        self.platform = platform

    def convert(self, raw_event: dict) -> dict:
        event_type = raw_event.get("type", "")

        base_event = {
            "id": raw_event.get("id", ""),
            "time": raw_event.get("timestamp", 0),
            "platform": self.platform,
            "self": {
                "platform": self.platform,
                "user_id": raw_event.get("self_id", ""),
            },
            "myplatform_raw": raw_event,
            "myplatform_raw_type": event_type,
        }

        if event_type == "chat":
            return self._convert_message(raw_event, base_event)
        elif event_type == "notification":
            return self._convert_notice(raw_event, base_event)
        elif event_type == "request":
            return self._convert_request(raw_event, base_event)

        return base_event

    def _convert_message(self, raw: dict, base: dict) -> dict:
        base["type"] = "message"
        base["detail_type"] = "group" if raw.get("group_id") else "private"
        base["user_id"] = raw.get("sender_id", "")
        base["message"] = self._convert_message_segments(raw.get("content", ""))
        base["alt_message"] = raw.get("content", "")

        if raw.get("group_id"):
            base["group_id"] = raw["group_id"]

        return base

    def _convert_message_segments(self, content: str) -> list:
        segments = []
        if content:
            segments.append({"type": "text", "data": {"text": content}})
        return segments

    def _convert_notice(self, raw: dict, base: dict) -> dict:
        base["type"] = "notice"
        notification_type = raw.get("notification_type", "")

        if notification_type == "member_join":
            base["detail_type"] = "group_member_increase"
            base["user_id"] = raw.get("user_id", "")
            base["group_id"] = raw.get("group_id", "")
            base["operator_id"] = raw.get("operator_id", "")
        elif notification_type == "friend_add":
            base["detail_type"] = "friend_increase"
            base["user_id"] = raw.get("user_id", "")

        return base

    def _convert_request(self, raw: dict, base: dict) -> dict:
        base["type"] = "request"
        request_type = raw.get("request_type", "")

        if request_type == "friend":
            base["detail_type"] = "friend"
            base["user_id"] = raw.get("user_id", "")
            base["comment"] = raw.get("message", "")
        elif request_type == "group_invite":
            base["detail_type"] = "group"
            base["group_id"] = raw.get("group_id", "")
            base["user_id"] = raw.get("inviter_id", "")

        return base

富媒體消息轉換示例

實際平台的消息通常包含圖片、@提及、回覆等富媒體內容。以下是 _convert_message_segments 處理多種消息類型的示例：

def _convert_message_segments(self, raw_content: list) -> list:
    """將平台原生消息段列表轉換為 OneBot12 標準消息段"""
    segments = []

    for item in raw_content:
        item_type = item.get("type", "")

        if item_type == "text":
            segments.append({
                "type": "text",
                "data": {"text": item.get("content", "")}
            })

        elif item_type == "image":
            file_url = item.get("url") or item.get("file_id", "")
            segments.append({
                "type": "image",
                "data": {"file": file_url}
            })

        elif item_type == "at":
            segments.append({
                "type": "mention",
                "data": {"user_id": item.get("target_id", "")}
            })

        elif item_type == "reply":
            segments.append({
                "type": "reply",
                "data": {"message_id": item.get("reply_to_id", "")}
            })

        elif item_type == "at_all":
            segments.append({"type": "mention_all", "data": {}})

        else:
            segments.append({
                "type": "text",
                "data": {"text": f"[不支援的消息類型: {item_type}]"}
            })

    return segments

常見陷阱

1. 缺少 `{platform}_raw` 字段

這是常見的錯誤。缺少原始數據字段會導致模組無法訪問平台特有的資訊。

base_event["myplatform_raw"] = raw_event        # 必須！
base_event["myplatform_raw_type"] = event_type   # 必須！

2. 時間戳格式錯誤

OneBot12 標準要求 time 字段為 Unix 秒級時間戳（整數）。如果你的平台返回毫秒時間戳或 ISO 格式字串，需要轉換：

import time

# 毫秒 → 秒
"time": raw_event.get("timestamp", 0) // 1000

# ISO 字串 → 秒
"time": int(time.mktime(time.strptime(raw_event["created_at"], "%Y-%m-%dT%H:%M:%S")))

3. 缺少 `self` 字段

self 字段包含機器人自身資訊，user_id 為機器人的帳號 ID。多 Bot 場景下此字段至關重要：

"self": {
    "platform": self.platform,
    "user_id": raw_event.get("bot_id", ""),   # 機器人自身的 ID
}

4. detail_type 使用了非標準值

detail_type 必須使用 OneBot12 標準定義的值，如 private、group、friend_increase、group_member_increase 等。不要使用平台特有的命名。

5. 往返一致性

確保 Converter 生成的消息段類型與 Send 端支援的方法對應。例如，如果 Converter 將平台的圖片消息轉換為 {"type": "image", ...}，那麼 Send 端的 Image() 方法必須能處理圖片發送。

最佳實踐

總是保留原始數據：{platform}_raw 字段不能省略
使用標準消息段：盡量將平台消息轉換為 OneBot12 標準消息段
合理設定 detail_type：使用標準類型（private/group/channel 等），不要自定義
處理邊界情況：原始事件可能缺少某些字段，使用 .get() 並提供合理預設值
效能考量：convert() 在每個事件上呼叫，避免在其中執行耗時操作