Losita/mai-bot
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

恢复可用性

Browse Source
This commit is contained in:
UnCLAS-Prommer
2025-08-18 11:16:28 +08:00
parent 3481234d2b
commit 5e26414839
13 changed files with 181 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
src/common/data_models/__init__.py
View File

@@ -1,3 +1,51 @@
from typing import Dict, Any
def temporarily_transform_class_to_dict(class_instance) -> Dict[str, Any]:
return class_instance.__dict__
class AbstractClassFlag:
pass
def temporarily_transform_class_to_dict(obj: Any) -> Any:
"""
将对象或容器中的 AbstractClassFlag 子类（类对象）或 AbstractClassFlag 实例
递归转换为普通 dict，不修改原对象。
- 对于类对象（isinstance(value, type) 且 issubclass(..., AbstractClassFlag)），
读取类的 __dict__ 中非 dunder 项并递归转换。
- 对于实例（isinstance(value, AbstractClassFlag)），读取 vars(instance) 并递归转换。
"""
def _transform(value: Any) -> Any:
# 值是类对象且为 AbstractClassFlag 的子类
if isinstance(value, type) and issubclass(value, AbstractClassFlag):
return {k: _transform(v) for k, v in value.__dict__.items() if not k.startswith("__") and not callable(v)}
# 值是 AbstractClassFlag 的实例
if isinstance(value, AbstractClassFlag):
return {k: _transform(v) for k, v in vars(value).items()}
# 常见容器类型，递归处理
if isinstance(value, dict):
return {k: _transform(v) for k, v in value.items()}
if isinstance(value, list):
return [_transform(v) for v in value]
if isinstance(value, tuple):
return tuple(_transform(v) for v in value)
if isinstance(value, set):
return {_transform(v) for v in value}
# 基本类型，直接返回
return value
result = _transform(obj)
def flatten(target_dict: dict):
flat_dict = {}
for k, v in target_dict.items():
if isinstance(v, dict):
# 递归扁平化子字典
sub_flat = flatten(v)
flat_dict.update(sub_flat)
else:
flat_dict[k] = v
return flat_dict
return flatten(result) if isinstance(result, dict) else result

141
src/common/data_models/database_data_model.py
View File

@@ -1,24 +1,36 @@
from typing import Optional
from dataclasses import dataclass, field
from typing import Optional, Dict, Any
from dataclasses import dataclass, field, fields, MISSING
from . import AbstractClassFlag
@dataclass
class DatabaseUserInfo:
class DatabaseUserInfo(AbstractClassFlag):
platform: str = field(default_factory=str)
user_id: str = field(default_factory=str)
user_nickname: str = field(default_factory=str)
user_cardname: Optional[str] = None
def __post_init__(self):
assert isinstance(self.platform, str), "platform must be a string"
assert isinstance(self.user_id, str), "user_id must be a string"
assert isinstance(self.user_nickname, str), "user_nickname must be a string"
assert isinstance(self.user_cardname, str) or self.user_cardname is None, "user_cardname must be a string or None"
@dataclass
class DatabaseGroupInfo:
class DatabaseGroupInfo(AbstractClassFlag):
group_id: str = field(default_factory=str)
group_name: str = field(default_factory=str)
group_platform: Optional[str] = None
def __post_init__(self):
assert isinstance(self.group_id, str), "group_id must be a string"
assert isinstance(self.group_name, str), "group_name must be a string"
assert isinstance(self.group_platform, str) or self.group_platform is None, "group_platform must be a string or None"
@dataclass
class DatabaseChatInfo:
class DatabaseChatInfo(AbstractClassFlag):
stream_id: str = field(default_factory=str)
platform: str = field(default_factory=str)
create_time: float = field(default_factory=float)
@@ -26,12 +38,20 @@ class DatabaseChatInfo:
user_info: DatabaseUserInfo = field(default_factory=DatabaseUserInfo)
group_info: Optional[DatabaseGroupInfo] = None
def __post_init__(self):
assert isinstance(self.stream_id, str), "stream_id must be a string"
assert isinstance(self.platform, str), "platform must be a string"
assert isinstance(self.create_time, float), "create_time must be a float"
assert isinstance(self.last_active_time, float), "last_active_time must be a float"
assert isinstance(self.user_info, DatabaseUserInfo), "user_info must be a DatabaseUserInfo instance"
assert isinstance(self.group_info, DatabaseGroupInfo) or self.group_info is None, "group_info must be a DatabaseGroupInfo instance or None"
@dataclass
class DatabaseMessages:
chat_info: DatabaseChatInfo
user_info: DatabaseUserInfo
group_info: Optional[DatabaseGroupInfo] = None
@dataclass(init=False)
class DatabaseMessages(AbstractClassFlag):
# chat_info: DatabaseChatInfo
# user_info: DatabaseUserInfo
# group_info: Optional[DatabaseGroupInfo] = None
message_id: str = field(default_factory=str)
time: float = field(default_factory=float)
@@ -44,23 +64,23 @@ class DatabaseMessages:
is_mentioned: Optional[bool] = None
# 从 chat_info 扁平化而来的字段
chat_info_stream_id: str = field(default_factory=str)
chat_info_platform: str = field(default_factory=str)
chat_info_user_platform: str = field(default_factory=str)
chat_info_user_id: str = field(default_factory=str)
chat_info_user_nickname: str = field(default_factory=str)
chat_info_user_cardname: Optional[str] = None
chat_info_group_platform: Optional[str] = None
chat_info_group_id: Optional[str] = None
chat_info_group_name: Optional[str] = None
chat_info_create_time: float = field(default_factory=float)
chat_info_last_active_time: float = field(default_factory=float)
# chat_info_stream_id: str = field(default_factory=str)
# chat_info_platform: str = field(default_factory=str)
# chat_info_user_platform: str = field(default_factory=str)
# chat_info_user_id: str = field(default_factory=str)
# chat_info_user_nickname: str = field(default_factory=str)
# chat_info_user_cardname: Optional[str] = None
# chat_info_group_platform: Optional[str] = None
# chat_info_group_id: Optional[str] = None
# chat_info_group_name: Optional[str] = None
# chat_info_create_time: float = field(default_factory=float)
# chat_info_last_active_time: float = field(default_factory=float)
# 从顶层 user_info 扁平化而来的字段 (消息发送者信息)
user_platform: str = field(default_factory=str)
user_id: str = field(default_factory=str)
user_nickname: str = field(default_factory=str)
user_cardname: Optional[str] = None
# user_platform: str = field(default_factory=str)
# user_id: str = field(default_factory=str)
# user_nickname: str = field(default_factory=str)
# user_cardname: Optional[str] = None
processed_plain_text: Optional[str] = None # 处理后的纯文本消息
display_message: Optional[str] = None # 显示的消息
@@ -76,32 +96,65 @@ class DatabaseMessages:
selected_expressions: Optional[str] = None
def __post_init__(self):
self.user_info = DatabaseUserInfo(
user_id=self.user_id,
user_nickname=self.user_nickname,
user_cardname=self.user_cardname,
platform=self.user_platform,
)
# def __post_init__(self):
if self.chat_info_group_id and self.chat_info_group_name:
# if self.chat_info_group_id and self.chat_info_group_name:
# self.group_info = DatabaseGroupInfo(
# group_id=self.chat_info_group_id,
# group_name=self.chat_info_group_name,
# group_platform=self.chat_info_group_platform,
# )
# chat_user_info = DatabaseUserInfo(
# user_id=self.chat_info_user_id,
# user_nickname=self.chat_info_user_nickname,
# user_cardname=self.chat_info_user_cardname,
# platform=self.chat_info_user_platform,
# )
# self.chat_info = DatabaseChatInfo(
# stream_id=self.chat_info_stream_id,
# platform=self.chat_info_platform,
# create_time=self.chat_info_create_time,
# last_active_time=self.chat_info_last_active_time,
# user_info=chat_user_info,
# group_info=self.group_info,
# )
def __init__(self, **kwargs: Any):
defined = {f.name: f for f in fields(self.__class__)}
for name, f in defined.items():
if name in kwargs:
setattr(self, name, kwargs.pop(name))
elif f.default is not MISSING:
setattr(self, name, f.default)
else:
raise TypeError(f"缺失必需字段: {name}")
self.user_info = DatabaseUserInfo(
user_id=kwargs.get("user_id"), # type: ignore
user_nickname=kwargs.get("user_nickname"), # type: ignore
user_cardname=kwargs.get("user_cardname"), # type: ignore
platform=kwargs.get("user_platform"), # type: ignore
)
if kwargs.get("chat_info_group_id") and kwargs.get("chat_info_group_name"):
self.group_info = DatabaseGroupInfo(
group_id=self.chat_info_group_id,
group_name=self.chat_info_group_name,
group_platform=self.chat_info_group_platform,
group_id=kwargs.get("chat_info_group_id"), # type: ignore
group_name=kwargs.get("chat_info_group_name"), # type: ignore
group_platform=kwargs.get("chat_info_group_platform"), # type: ignore
)
chat_user_info = DatabaseUserInfo(
user_id=self.chat_info_user_id,
user_nickname=self.chat_info_user_nickname,
user_cardname=self.chat_info_user_cardname,
platform=self.chat_info_user_platform,
user_id=kwargs.get("chat_info_user_id"), # type: ignore
user_nickname=kwargs.get("chat_info_user_nickname"), # type: ignore
user_cardname=kwargs.get("chat_info_user_cardname"), # type: ignore
platform=kwargs.get("chat_info_user_platform"), # type: ignore
)
self.chat_info = DatabaseChatInfo(
stream_id=self.chat_info_stream_id,
platform=self.chat_info_platform,
create_time=self.chat_info_create_time,
last_active_time=self.chat_info_last_active_time,
stream_id=kwargs.get("chat_info_stream_id"), # type: ignore
platform=kwargs.get("chat_info_platform"), # type: ignore
create_time=kwargs.get("chat_info_create_time"), # type: ignore
last_active_time=kwargs.get("chat_info_last_active_time"), # type: ignore
user_info=chat_user_info,
group_info=self.group_info,
)