20260101_01

models/event.py

@@ -1,6 +1,6 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import List, Optional, Dict, Any
-from .sender import Sender  # 导入上面的 Sender
+from .sender import Sender
 
 @dataclass
 class MessageSegment:
@@ -9,16 +9,13 @@ class MessageSegment:
 
     @property
     def text(self) -> str:
-        """如果是文本段，返回文本内容，否则返回空字符串"""
         return self.data.get("text", "") if self.type == "text" else ""
 
     @property
     def image_url(self) -> str:
-        """如果是图片段，返回图片 URL"""
         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:
@@ -28,36 +25,72 @@ class MessageSegment:
     def __repr__(self):
         return f"[MS:{self.type}:{self.data}]"
 
+
+
 @dataclass
 class Event:
     post_type: str
-    message_type: str # group 或 private
-    user_id: int
     self_id: int
-    raw_message: str
-    message: List[MessageSegment]
-    sender: Sender
     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 = [
-            MessageSegment(type=seg["type"], data=seg["data"]) 
-            for seg in raw_msg_array
-        ]
+        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
+            ]
 
-        data_copy = data.copy()
-        data_copy["message"] = segments
- 
-        sender_data = data.get("sender", {})
-        sender_obj = Sender(**{k: v for k, v in sender_data.items() if k in Sender.__annotations__})        
+        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__
+            })
 
-        data_copy = data.copy()
-        data_copy["message"] = segments
-        data_copy["sender"] = sender_obj # 关键点：把对象塞进去
+        #  数据整合
+        processed_data = data.copy()
+        processed_data["message"] = segments
+        processed_data["sender"] = sender_obj
         
-        valid_data = {k: v for k, v in data_copy.items() if k in cls.__annotations__}
-        return cls(**valid_data)
+        #  字段过滤：只提取 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"

models/sender.py

@@ -1,8 +0,0 @@
-from dataclasses import dataclass
-#TODO 数据类型
-@dataclass
-class Sender:
-    user_id: int
-    nickname: str
-    card: str = ""
-    role: str = ""  # admin, owner, member