""" 消息组件系统 (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