""" 事件模型模块 定义了 Event 类和 MessageSegment 类,用于封装 OneBot 11 的上报事件和消息段。 """ from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, TYPE_CHECKING from .sender import Sender if TYPE_CHECKING: from core.bot import Bot @dataclass class MessageSegment: """ 消息段,对应 OneBot 11 标准中的消息段对象 """ type: str """消息段类型,如 text, image, at 等""" data: Dict[str, Any] """消息段数据""" @property def text(self) -> str: """ 获取文本内容(仅当 type 为 text 时有效) :return: 文本内容 """ return self.data.get("text", "") if self.type == "text" else "" @property def image_url(self) -> str: """ 获取图片 URL(仅当 type 为 image 时有效) :return: 图片 URL """ return self.data.get("url", "") if self.type == "image" else "" def is_at(self, user_id: int = None) -> bool: """ 判断是否为 @某人 :param user_id: 指定的 QQ 号,如果为 None 则只判断是否为 at 类型 :return: 是否匹配 """ if self.type != "at": return False if user_id is None: return True return str(self.data.get("qq")) == str(user_id) def __repr__(self): return f"[MS:{self.type}:{self.data}]" @dataclass class Event: """ 事件类,封装了 OneBot 11 的上报事件 """ post_type: str """上报类型: message, notice, request, meta_event""" self_id: int """收到消息的机器人 QQ 号""" time: int """事件发生的时间戳""" # --- 消息事件字段 --- message_type: Optional[str] = None """消息类型: group, private""" sub_type: Optional[str] = None """消息子类型""" message_id: Optional[int] = None """消息 ID""" user_id: Optional[int] = None """发送者 QQ 号""" raw_message: Optional[str] = None """原始消息内容""" message: List[MessageSegment] = field(default_factory=list) """消息内容列表""" sender: Optional[Sender] = None """发送者信息""" group_id: Optional[int] = None """群号""" target_id: Optional[int] = None """目标 QQ 号""" # --- 通知事件字段 --- notice_type: Optional[str] = None """通知类型""" operator_id: Optional[int] = None """操作者 QQ 号""" duration: Optional[int] = None """时长""" honor_type: Optional[str] = None """荣誉类型""" # --- 请求事件字段 --- request_type: Optional[str] = None """请求类型""" flag: Optional[str] = None """请求 flag""" comment: Optional[str] = None """验证信息""" # 注入的 Bot 实例 bot: Optional["Bot"] = field(default=None, init=False) """关联的 Bot 实例""" async def reply(self, message: str) -> dict: """ 快捷回复消息 :param message: 回复内容 :return: API 响应结果 """ if not self.bot: return {"status": "failed", "msg": "Bot instance not attached to event"} return await self.bot.send(self, message) @classmethod def from_dict(cls, data: dict): """ 从字典创建 Event 对象 :param data: 原始事件数据字典 :return: Event 对象 """ raw_msg_array = data.get("message") segments = [] if isinstance(raw_msg_array, list): segments = [ MessageSegment(type=seg["type"], data=seg["data"]) for seg in raw_msg_array ] sender_data = data.get("sender") sender_obj = None if isinstance(sender_data, dict): sender_obj = Sender( **{k: v for k, v in sender_data.items() if k in Sender.__annotations__} ) # 数据整合 processed_data = data.copy() processed_data["message"] = segments processed_data["sender"] = sender_obj # 字段过滤:只提取 dataclass 中定义的字段 valid_data = { k: v for k, v in processed_data.items() if k in cls.__annotations__ } return cls(**valid_data) # --- 快捷判断工具 --- @property def is_message(self) -> bool: """是否为消息事件""" return self.post_type == "message" @property def is_notice(self) -> bool: """是否为通知事件""" return self.post_type == "notice" @property def is_request(self) -> bool: """是否为请求事件""" return self.post_type == "request"