qyyfw
78ff261b5a
- 前端 React+TypeScript (源码+构建产物) - 后端 Python FastAPI - 数据库迁移 001-015 - Docker部署配置 - 微信ClawBot API频道绑定 - 项目文档 README.md
340 lines
8.9 KiB
Python
340 lines
8.9 KiB
Python
"""
|
|
消息组件系统 (Message Component System)
|
|
参考 AstrBot 消息组件设计，提供统一的消息链表示
|
|
|
|
消息链 (Message Chain) 是由多个消息组件（Component）组成的有序列表，
|
|
每个组件代表消息中的一种元素（文本、图片、@提及、回复、文件等）。
|
|
|
|
使用示例:
|
|
from app.message_components import Plain, Image, At, Reply, message_chain_to_dict, message_chain_from_dict
|
|
|
|
# 构建消息链
|
|
chain = [
|
|
Reply(message_id="msg_123", content_preview="你好"),
|
|
Plain(text=" "),
|
|
At(user_id="user_456", display_name="张三"),
|
|
Plain(text=" 看这张图："),
|
|
Image(url="https://example.com/img.png", width=800, height=600),
|
|
]
|
|
|
|
# 序列化为字典列表（用于JSON传输）
|
|
dict_list = message_chain_to_dict(chain)
|
|
|
|
# 从字典列表恢复消息链
|
|
restored_chain = message_chain_from_dict(dict_list)
|
|
|
|
# 提取纯文本
|
|
text = extract_plain_text(chain) # " 看这张图："
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Optional, List, Dict
|
|
|
|
|
|
# ============================================================
|
|
# 组件类型常量
|
|
# ============================================================
|
|
|
|
class ComponentType:
|
|
"""消息组件类型常量"""
|
|
PLAIN = "plain"
|
|
IMAGE = "image"
|
|
AT = "at"
|
|
REPLY = "reply"
|
|
FILE = "file"
|
|
|
|
|
|
# 类型到类的映射表，用于 from_dict 反序列化
|
|
_COMPONENT_REGISTRY: Dict[str, type] = {}
|
|
|
|
|
|
def _register_component(component_type: str, cls: type):
|
|
"""注册组件类型，用于反序列化"""
|
|
_COMPONENT_REGISTRY[component_type] = cls
|
|
|
|
|
|
# ============================================================
|
|
# 基类
|
|
# ============================================================
|
|
|
|
class BaseMessageComponent(ABC):
|
|
"""消息组件基类，所有消息组件必须继承此类"""
|
|
|
|
type: str
|
|
|
|
@abstractmethod
|
|
def to_dict(self) -> dict:
|
|
"""将组件序列化为字典"""
|
|
...
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "BaseMessageComponent":
|
|
"""从字典反序列化组件"""
|
|
component_type = data.get("type", "")
|
|
subclass = _COMPONENT_REGISTRY.get(component_type)
|
|
if subclass and subclass is not cls:
|
|
return subclass.from_dict(data)
|
|
raise ValueError(f"未知的组件类型: {component_type}")
|
|
|
|
|
|
# ============================================================
|
|
# 具体组件实现
|
|
# ============================================================
|
|
|
|
@dataclass
|
|
class Plain(BaseMessageComponent):
|
|
"""纯文本组件"""
|
|
text: str
|
|
type: str = ComponentType.PLAIN
|
|
|
|
def to_dict(self) -> dict:
|
|
return {"type": self.type, "text": self.text}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "Plain":
|
|
return cls(text=data.get("text", ""))
|
|
|
|
|
|
_register_component(ComponentType.PLAIN, Plain)
|
|
|
|
|
|
@dataclass
|
|
class Image(BaseMessageComponent):
|
|
"""图片组件
|
|
|
|
Attributes:
|
|
url: 图片URL（网络地址或本地静态资源路径）
|
|
file_path: 本地文件路径
|
|
width: 图片宽度（像素）
|
|
height: 图片高度（像素）
|
|
"""
|
|
url: str = ""
|
|
file_path: str = ""
|
|
width: int = 0
|
|
height: int = 0
|
|
type: str = ComponentType.IMAGE
|
|
|
|
def to_dict(self) -> dict:
|
|
result = {"type": self.type}
|
|
if self.url:
|
|
result["url"] = self.url
|
|
if self.file_path:
|
|
result["file_path"] = self.file_path
|
|
if self.width:
|
|
result["width"] = self.width
|
|
if self.height:
|
|
result["height"] = self.height
|
|
return result
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "Image":
|
|
return cls(
|
|
url=data.get("url", ""),
|
|
file_path=data.get("file_path", ""),
|
|
width=data.get("width", 0),
|
|
height=data.get("height", 0),
|
|
)
|
|
|
|
|
|
_register_component(ComponentType.IMAGE, Image)
|
|
|
|
|
|
@dataclass
|
|
class At(BaseMessageComponent):
|
|
"""@提及组件
|
|
|
|
Attributes:
|
|
user_id: 被提及用户的ID
|
|
display_name: 被提及用户的显示名称
|
|
"""
|
|
user_id: str
|
|
display_name: str = ""
|
|
type: str = ComponentType.AT
|
|
|
|
def to_dict(self) -> dict:
|
|
result = {"type": self.type, "user_id": self.user_id}
|
|
if self.display_name:
|
|
result["display_name"] = self.display_name
|
|
return result
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "At":
|
|
return cls(
|
|
user_id=data.get("user_id", ""),
|
|
display_name=data.get("display_name", ""),
|
|
)
|
|
|
|
|
|
_register_component(ComponentType.AT, At)
|
|
|
|
|
|
@dataclass
|
|
class Reply(BaseMessageComponent):
|
|
"""回复组件（引用/回复某条消息）
|
|
|
|
Attributes:
|
|
message_id: 被回复消息的ID
|
|
content_preview: 被回复消息的内容预览
|
|
"""
|
|
message_id: str
|
|
content_preview: str = ""
|
|
type: str = ComponentType.REPLY
|
|
|
|
def to_dict(self) -> dict:
|
|
result = {"type": self.type, "message_id": self.message_id}
|
|
if self.content_preview:
|
|
result["content_preview"] = self.content_preview
|
|
return result
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "Reply":
|
|
return cls(
|
|
message_id=data.get("message_id", ""),
|
|
content_preview=data.get("content_preview", ""),
|
|
)
|
|
|
|
|
|
_register_component(ComponentType.REPLY, Reply)
|
|
|
|
|
|
@dataclass
|
|
class File(BaseMessageComponent):
|
|
"""文件组件
|
|
|
|
Attributes:
|
|
filename: 文件名
|
|
url: 文件下载地址
|
|
size: 文件大小（字节）
|
|
"""
|
|
filename: str
|
|
url: str = ""
|
|
size: int = 0
|
|
type: str = ComponentType.FILE
|
|
|
|
def to_dict(self) -> dict:
|
|
result = {"type": self.type, "filename": self.filename}
|
|
if self.url:
|
|
result["url"] = self.url
|
|
if self.size:
|
|
result["size"] = self.size
|
|
return result
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "File":
|
|
return cls(
|
|
filename=data.get("filename", ""),
|
|
url=data.get("url", ""),
|
|
size=data.get("size", 0),
|
|
)
|
|
|
|
|
|
_register_component(ComponentType.FILE, File)
|
|
|
|
|
|
# ============================================================
|
|
# 工厂函数
|
|
# ============================================================
|
|
|
|
def message_chain_to_dict(chain: List[BaseMessageComponent]) -> List[dict]:
|
|
"""将消息链组件列表转换为字典列表（用于JSON序列化）
|
|
|
|
Args:
|
|
chain: 消息组件列表
|
|
|
|
Returns:
|
|
字典列表，每个字典包含 type 和组件特定字段
|
|
"""
|
|
return [comp.to_dict() for comp in chain]
|
|
|
|
|
|
def message_chain_from_dict(data_list: List[dict]) -> List[BaseMessageComponent]:
|
|
"""从字典列表恢复消息链组件列表
|
|
|
|
Args:
|
|
data_list: 字典列表
|
|
|
|
Returns:
|
|
消息组件列表
|
|
"""
|
|
chain = []
|
|
for data in data_list:
|
|
component_type = data.get("type", "")
|
|
component_cls = _COMPONENT_REGISTRY.get(component_type)
|
|
if component_cls:
|
|
chain.append(component_cls.from_dict(data))
|
|
else:
|
|
# 未知类型，降级为 Plain 文本
|
|
chain.append(Plain(text=str(data)))
|
|
return chain
|
|
|
|
|
|
# ============================================================
|
|
# 辅助函数
|
|
# ============================================================
|
|
|
|
def extract_plain_text(chain: List[BaseMessageComponent]) -> str:
|
|
"""从消息链中提取所有纯文本内容，拼接为完整文本
|
|
|
|
遍历消息链，提取所有 Plain 组件的 text 字段并拼接。
|
|
|
|
Args:
|
|
chain: 消息组件列表
|
|
|
|
Returns:
|
|
拼接后的纯文本字符串
|
|
"""
|
|
texts = []
|
|
for comp in chain:
|
|
if isinstance(comp, Plain):
|
|
texts.append(comp.text)
|
|
return "".join(texts)
|
|
|
|
|
|
def extract_image_urls(chain: List[BaseMessageComponent]) -> List[str]:
|
|
"""从消息链中提取所有图片URL
|
|
|
|
Args:
|
|
chain: 消息组件列表
|
|
|
|
Returns:
|
|
图片URL列表
|
|
"""
|
|
urls = []
|
|
for comp in chain:
|
|
if isinstance(comp, Image) and comp.url:
|
|
urls.append(comp.url)
|
|
return urls
|
|
|
|
|
|
def extract_at_user_ids(chain: List[BaseMessageComponent]) -> List[str]:
|
|
"""从消息链中提取所有被@的用户ID
|
|
|
|
Args:
|
|
chain: 消息组件列表
|
|
|
|
Returns:
|
|
用户ID列表
|
|
"""
|
|
ids = []
|
|
for comp in chain:
|
|
if isinstance(comp, At):
|
|
ids.append(comp.user_id)
|
|
return ids
|
|
|
|
|
|
def has_component_type(chain: List[BaseMessageComponent], component_type: str) -> bool:
|
|
"""判断消息链中是否包含指定类型的组件
|
|
|
|
Args:
|
|
chain: 消息组件列表
|
|
component_type: 组件类型字符串（如 "image", "at" 等）
|
|
|
|
Returns:
|
|
True 如果包含该类型组件
|
|
"""
|
|
for comp in chain:
|
|
if comp.type == component_type:
|
|
return True
|
|
return False
|