96 lines
2.6 KiB
Python
96 lines
2.6 KiB
Python
from dataclasses import dataclass, field
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from .sender import Sender
|
|
|
|
|
|
@dataclass
|
|
class MessageSegment:
|
|
type: str
|
|
data: Dict[str, Any]
|
|
|
|
@property
|
|
def text(self) -> str:
|
|
return self.data.get("text", "") if self.type == "text" else ""
|
|
|
|
@property
|
|
def image_url(self) -> str:
|
|
return self.data.get("url", "") if self.type == "image" else ""
|
|
|
|
def is_at(self, user_id: int = None) -> bool:
|
|
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:
|
|
post_type: str
|
|
self_id: int
|
|
time: int
|
|
|
|
message_type: Optional[str] = None
|
|
sub_type: Optional[str] = None
|
|
message_id: Optional[int] = None
|
|
user_id: Optional[int] = None
|
|
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
|
|
|
|
notice_type: Optional[str] = None
|
|
operator_id: Optional[int] = None
|
|
duration: Optional[int] = None
|
|
honor_type: Optional[str] = None
|
|
|
|
request_type: Optional[str] = None
|
|
flag: Optional[str] = None
|
|
comment: Optional[str] = None
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict):
|
|
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"
|