# -*- coding: utf-8 -*- """ 通用数据模型 (Universal Data Models) 此模块定义了平台无关的数据结构,用于在不同平台(如 OneBot, Discord) 和业务逻辑层(如 Plugins)之间传递数据。 """ from dataclasses import dataclass, field from typing import List, Optional, Union, Dict, Any @dataclass class UniversalMessageSegment: """ 平台无关的通用消息段模型。 业务逻辑层只负责生成这个对象,由底层的 Adapter 负责将其翻译成特定平台的格式。 """ type: str # 消息类型:'text', 'image', 'video', 'audio', 'at', 'reply' 等 data: Dict[str, Any] # 消息数据载荷 @staticmethod def text(text: str) -> "UniversalMessageSegment": return UniversalMessageSegment("text", {"text": text}) @staticmethod def image(url: Optional[str] = None, base64: Optional[str] = None, file_path: Optional[str] = None) -> "UniversalMessageSegment": """ 图片消息。 Discord 支持直接发 URL 或上传本地文件;OneBot 支持 URL、Base64 或本地路径。 """ return UniversalMessageSegment("image", {"url": url, "base64": base64, "file_path": file_path}) @staticmethod def video(url: Optional[str] = None, file_path: Optional[str] = None) -> "UniversalMessageSegment": """ 视频消息。 Discord 通常直接发 URL 或作为附件上传;OneBot 支持 URL 或本地路径。 """ return UniversalMessageSegment("video", {"url": url, "file_path": file_path}) @staticmethod def at(user_id: str) -> "UniversalMessageSegment": """ @某人。 注意:为了兼容 Discord 的雪花 ID (Snowflake),user_id 必须是字符串。 """ return UniversalMessageSegment("at", {"user_id": user_id}) @staticmethod def reply(message_id: str) -> "UniversalMessageSegment": """ 回复某条消息。 """ return UniversalMessageSegment("reply", {"message_id": message_id}) @dataclass class UniversalUser: """通用用户模型""" id: str # 用户唯一ID (QQ号 或 Discord Snowflake ID) name: str # 用户昵称/群名片 avatar_url: str # 头像URL is_bot: bool # 是否是机器人 @dataclass class UniversalChannel: """通用频道/群组模型""" id: str # 频道/群组唯一ID (QQ群号 或 Discord Channel ID) name: str # 频道/群组名称 type: str # 类型:'private' (私聊), 'group' (QQ群), 'guild_text' (Discord文字频道) 等 guild_id: Optional[str] = None # 仅 Discord 有效:服务器(Guild) ID @dataclass class UniversalMessageEvent: """ 平台无关的通用消息事件模型。 这是传递给业务逻辑层(如 bili.py)的最终对象。 """ platform: str # 来源平台标识:'onebot' 或 'discord' message_id: str # 消息唯一ID (QQ消息ID 或 Discord Message ID) user: UniversalUser # 发送者信息 channel: UniversalChannel # 消息来源频道/群组信息 raw_message: str # 纯文本形式的消息内容(用于正则匹配、命令解析) # 解析后的消息段列表(可选,如果你需要处理图文混排) message: List[UniversalMessageSegment] = field(default_factory=list) # 原始的底层事件对象(保留引用,方便高级操作) # 例如:OneBot 的原始 JSON 字典,或 discord.py 的 discord.Message 对象 raw_event: Any = field(repr=False, default=None) async def reply(self, message: Union[str, UniversalMessageSegment, List[UniversalMessageSegment]]): """ 统一的回复接口。 这个方法应该是一个抽象方法或由具体的 Adapter 注入实现。 业务逻辑层调用此方法时,不需要关心底层是调用 OneBot API 还是 Discord API。 """ raise NotImplementedError("此方法应由具体的 Platform Adapter 实现")