53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
import arrow
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
||
from beanie import Document
|
||
|
||
|
||
class Message(Document):
|
||
"""消息会话模型"""
|
||
|
||
# 基本信息
|
||
msgid: str # 消息ID
|
||
action: str # 动作类型
|
||
from_user: str # 发送者，直接使用from_user字段
|
||
tolist: list[str] # 接收者列表
|
||
roomid: str | None = None # 群聊ID
|
||
text: dict # 消息内容
|
||
msgtime: datetime # 消息时间戳
|
||
msgtype: str # 消息类型
|
||
remark: str | None = None # 备注
|
||
|
||
# 通用字段
|
||
seq: int = 1000 # 排序序号
|
||
isEnabled: bool = True # 是否启用
|
||
created_at: datetime | None = arrow.now().datetime # 创建时间
|
||
updated_at: datetime | None = arrow.now().datetime # 更新时间
|
||
|
||
class Settings:
|
||
name = "messages" # 指定MongoDB集合名称
|
||
|
||
@field_validator("msgtime", mode="before")
|
||
@classmethod
|
||
def convert_timestamp_to_datetime(cls, v: Any) -> datetime:
|
||
"""将毫秒级时间戳转换为datetime对象"""
|
||
if isinstance(v, (int, float)):
|
||
# 假设是毫秒级时间戳
|
||
return datetime.fromtimestamp(v / 1000)
|
||
elif isinstance(v, str):
|
||
# 尝试解析字符串
|
||
try:
|
||
# 尝试解析为时间戳
|
||
timestamp = float(v)
|
||
return datetime.fromtimestamp(timestamp / 1000)
|
||
except ValueError:
|
||
# 尝试解析为ISO格式字符串
|
||
return datetime.fromisoformat(v.replace("Z", "+00:00"))
|
||
return v
|
||
|
||
class Config:
|
||
populate_by_name = True # 允许使用字段名和别名
|
||
arbitrary_types_allowed = True
|