From b80b5afe2a190267a9e3e4bf71fcb90bad298bab Mon Sep 17 00:00:00 2001 From: UnCLAS-Prommer Date: Fri, 13 Feb 2026 15:45:38 +0800 Subject: [PATCH] =?UTF-8?q?=E5=85=88=E6=8E=A8=E4=B8=8A=E5=8E=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../data_models/chat_session_data_model.py | 27 ++++ .../data_models/mai_message_data_model.py | 133 ++++++++++++++++++ 2 files changed, 160 insertions(+) create mode 100644 src/common/data_models/chat_session_data_model.py create mode 100644 src/common/data_models/mai_message_data_model.py diff --git a/src/common/data_models/chat_session_data_model.py b/src/common/data_models/chat_session_data_model.py new file mode 100644 index 00000000..9041c08f --- /dev/null +++ b/src/common/data_models/chat_session_data_model.py @@ -0,0 +1,27 @@ +from typing import Optional + +from src.common.database.database_model import ChatSession + + +class MaiChatSession: + def __init__(self, session_id: str, platform: str, user_id: Optional[str] = None, group_id: Optional[str] = None): + self.session_id = session_id + self.platform = platform + self.user_id = user_id + self.group_id = group_id + + # 验证字段 + assert self.platform, "Platform must be provided" + assert self.user_id or self.group_id, "UserID 或 GroupID 必须提供其一" + + # 其他字段初始化 + self.is_group_session = bool(self.group_id) + + @classmethod + def from_db_instance(cls, session: ChatSession): + return cls( + session_id=session.session_id, + platform=session.platform, + user_id=session.user_id, + group_id=session.group_id, + ) diff --git a/src/common/data_models/mai_message_data_model.py b/src/common/data_models/mai_message_data_model.py new file mode 100644 index 00000000..d11d7a4f --- /dev/null +++ b/src/common/data_models/mai_message_data_model.py @@ -0,0 +1,133 @@ +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from maim_message import MessageBase +from typing import Optional + +import json +from datetime import datetime + +from src.common.database.database_model import Messages +from src.common.data_models.message_component_model import MessageSequence +from src.common.utils.utils_message import MessageUtils + + +class BaseMAIMessageModel(ABC): + @classmethod + @abstractmethod + def from_db_instance(cls, message: "Messages"): + raise NotImplementedError + + @abstractmethod + def to_db_instance(self) -> "Messages": + raise NotImplementedError + + @abstractmethod + def from_maim_message(cls, message: MessageBase): + raise NotImplementedError + + @abstractmethod + def to_maim_message(self) -> MessageBase: + raise NotImplementedError + + @abstractmethod + def parse_message_segments(self): + raise NotImplementedError + + +@dataclass +class UserInfo: + user_id: str + user_nickname: str + user_cardname: Optional[str] = None + + +@dataclass +class GroupInfo: + group_id: str + group_name: str + + +@dataclass +class MessageInfo: + user_info: UserInfo + group_info: Optional[GroupInfo] = None + additional_config: dict = field(default_factory=dict) + + +class MaiMessage(BaseMAIMessageModel): + def __init__(self, message_id: str, timestamp: datetime): + self.message_id: str = message_id + self.timestamp: datetime = timestamp # 时间戳 + self.initialized = False # 用于标记是否已初始化其他属性 + self.platform: str # 初始化后赋值 + + # 定义其他属性 + self.message_info: MessageInfo # 初始化后赋值 + self.is_mentioned: bool = False + self.is_at: bool = False + self.is_emoji: bool = False + self.is_picture: bool = False + self.is_command: bool = False + self.is_notify: bool = False + + self.session_id: str + self.reply_to: Optional[str] = None + + self.processed_plain_text: Optional[str] = None + self.display_message: Optional[str] = None + self.raw_message: MessageSequence + + @classmethod + def from_db_instance(cls, message: "Messages") -> "MaiMessage": + obj = cls(message_id=message.message_id, timestamp=message.timestamp) + + user_info = UserInfo(message.user_id, message.user_nickname, message.user_cardname) + if message.group_id and message.group_name: + group_info = GroupInfo(message.group_id, message.group_name) + else: + group_info = None + obj.message_info = MessageInfo( + user_info=user_info, + group_info=group_info, + additional_config=json.loads(message.additional_config) if message.additional_config else {}, + ) + + obj.is_mentioned = message.is_mentioned + obj.is_at = message.is_at + obj.is_emoji = message.is_emoji + obj.is_picture = message.is_picture + obj.is_command = message.is_command + obj.is_notify = message.is_notify + obj.reply_to = message.reply_to + obj.session_id = message.session_id + obj.processed_plain_text = message.processed_plain_text + obj.display_message = message.display_message + obj.raw_message = MessageUtils.from_db_record_msg_to_MaiSeq(message.raw_content) + return obj + + def to_db_instance(self) -> Messages: + additional_config = ( + json.dumps(self.message_info.additional_config) if self.message_info.additional_config else None + ) + return Messages( + message_id=self.message_id, + timestamp=self.timestamp, + platform=self.platform, + user_id=self.message_info.user_info.user_id, + user_nickname=self.message_info.user_info.user_nickname, + user_cardname=self.message_info.user_info.user_cardname, + group_id=self.message_info.group_info.group_id if self.message_info.group_info else None, + group_name=self.message_info.group_info.group_name if self.message_info.group_info else None, + is_mentioned=self.is_mentioned, + is_at=self.is_at, + session_id=self.session_id, + reply_to=self.reply_to, + is_emoji=self.is_emoji, + is_picture=self.is_picture, + is_command=self.is_command, + is_notify=self.is_notify, + raw_content=MessageUtils.from_MaiSeq_to_db_record_msg(self.raw_message), + processed_plain_text=self.processed_plain_text, + display_message=self.display_message, + additional_config=additional_config, + )