qyyfw
78ff261b5a
- 前端 React+TypeScript (源码+构建产物) - 后端 Python FastAPI - 数据库迁移 001-015 - Docker部署配置 - 微信ClawBot API频道绑定 - 项目文档 README.md
126 lines
4.1 KiB
Python
126 lines
4.1 KiB
Python
"""
|
|
平台适配器注册机制 - 参照AstrBot @register_platform_adapter 装饰器模式
|
|
让新平台接入标准化，通过装饰器注册 + 统一接口
|
|
"""
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Callable
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 全局注册表
|
|
platform_registry: dict[str, "PlatformMetadata"] = {}
|
|
platform_cls_map: dict[str, type] = {}
|
|
|
|
@dataclass
|
|
class PlatformMetadata:
|
|
"""平台元数据"""
|
|
name: str # 平台标识 wecom/feishu/dingtalk/telegram/slack/whatsapp
|
|
display_name: str # 显示名称
|
|
description: str # 描述
|
|
icon: str = "" # 图标URL或base64
|
|
version: str = "1.0"
|
|
support_streaming: bool = True # 是否支持流式消息
|
|
support_image: bool = True # 是否支持图片消息
|
|
support_file: bool = True # 是否支持文件消息
|
|
support_voice: bool = False # 是否支持语音消息
|
|
config_schema: dict = field(default_factory=dict) # JSON Schema for config validation
|
|
|
|
class BasePlatformAdapter(ABC):
|
|
"""平台适配器基类 - 所有平台适配器必须继承此类"""
|
|
|
|
def __init__(self, config: dict):
|
|
self.config = config
|
|
self._status = "stopped" # pending/running/error/stopped
|
|
|
|
@abstractmethod
|
|
def get_metadata(self) -> PlatformMetadata:
|
|
"""返回平台元数据"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def parse_webhook(self, payload: dict) -> dict:
|
|
"""解析webhook payload为统一消息格式
|
|
返回: {
|
|
"content_type": "text"|"image"|"file"|...,
|
|
"content": "文本内容或资源标识",
|
|
"sender_id": "发送者ID",
|
|
"sender_name": "发送者名称",
|
|
"message_id": "平台消息ID",
|
|
"timestamp": 1234567890,
|
|
"group_id": "群聊ID(可选)",
|
|
"group_name": "群聊名称(可选)",
|
|
"raw": raw_payload,
|
|
}
|
|
"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def verify_signature(self, payload: dict, signature: str, timestamp: str) -> bool:
|
|
"""验证webhook签名"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def build_reply(self, channel_config: dict, content: str, reply_to: str = None) -> dict:
|
|
"""构建回复消息格式
|
|
返回: {"api_url": "...", "headers": {...}, "body": {...}}
|
|
"""
|
|
...
|
|
|
|
def get_status(self) -> str:
|
|
return self._status
|
|
|
|
def start(self):
|
|
self._status = "running"
|
|
|
|
def stop(self):
|
|
self._status = "stopped"
|
|
|
|
def register_platform_adapter(
|
|
name: str,
|
|
display_name: str,
|
|
description: str,
|
|
icon: str = "",
|
|
support_streaming: bool = True,
|
|
support_image: bool = True,
|
|
support_file: bool = True,
|
|
support_voice: bool = False,
|
|
config_schema: dict = None,
|
|
):
|
|
"""注册平台适配器的装饰器 - 参照AstrBot @register_platform_adapter 模式"""
|
|
def decorator(cls):
|
|
if not issubclass(cls, BasePlatformAdapter):
|
|
raise TypeError(f"{cls.__name__} 必须继承 BasePlatformAdapter")
|
|
|
|
metadata = PlatformMetadata(
|
|
name=name,
|
|
display_name=display_name,
|
|
description=description,
|
|
icon=icon,
|
|
support_streaming=support_streaming,
|
|
support_image=support_image,
|
|
support_file=support_file,
|
|
support_voice=support_voice,
|
|
config_schema=config_schema or {},
|
|
)
|
|
|
|
if name in platform_cls_map:
|
|
logger.warning(f"平台 {name} 已注册，将被覆盖")
|
|
|
|
platform_registry[name] = metadata
|
|
platform_cls_map[name] = cls
|
|
logger.info(f"[Platform] 注册平台适配器: {name} ({display_name})")
|
|
return cls
|
|
return decorator
|
|
|
|
def get_platform_adapter(name: str, config: dict) -> BasePlatformAdapter:
|
|
"""获取平台适配器实例"""
|
|
if name not in platform_cls_map:
|
|
raise ValueError(f"未知平台: {name}. 可用平台: {list(platform_cls_map.keys())}")
|
|
return platform_cls_map[name](config)
|
|
|
|
def list_platforms() -> list[PlatformMetadata]:
|
|
"""列出所有已注册平台"""
|
|
return list(platform_registry.values())
|