57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
from datetime import datetime, timedelta, timezone
|
||
from pydantic import BaseModel, field_serializer
|
||
CN_TZ = timezone(timedelta(hours=8))
|
||
|
||
|
||
def now_cn() -> datetime:
|
||
"""
|
||
获取中国时区的当前时间
|
||
返回带有中国时区信息的 datetime 对象
|
||
"""
|
||
return datetime.now(CN_TZ)
|
||
|
||
def parse_time(val: str | int, is_end: bool = False) -> datetime:
|
||
"""
|
||
将传入的字符串或时间戳解析为中国时区的 datetime，用于数据库查询时间比较。
|
||
支持格式:
|
||
- "YYYY-MM-DD"
|
||
- "YYYY-MM-DD HH:mm:ss"
|
||
- 10 位时间戳（秒）
|
||
- 13 位时间戳（毫秒）
|
||
"""
|
||
dt_cn: datetime
|
||
|
||
if isinstance(val, int) or (isinstance(val, str) and val.isdigit()):
|
||
ts = int(val)
|
||
# 根据量级判断是秒还是毫秒
|
||
if ts >= 10**12:
|
||
dt_cn = datetime.fromtimestamp(ts / 1000, CN_TZ)
|
||
else:
|
||
dt_cn = datetime.fromtimestamp(ts, CN_TZ)
|
||
else:
|
||
try:
|
||
dt_cn = datetime.strptime(val, "%Y-%m-%d").replace(tzinfo=CN_TZ)
|
||
if is_end:
|
||
dt_cn = dt_cn.replace(hour=23, minute=59, second=59, microsecond=999999)
|
||
except ValueError:
|
||
try:
|
||
dt_cn = datetime.strptime(val, "%Y-%m-%d %H:%M:%S").replace(tzinfo=CN_TZ)
|
||
except ValueError:
|
||
raise ValueError("时间格式错误，支持 'YYYY-MM-DD' 或 'YYYY-MM-DD HH:mm:ss' 或 10/13位时间戳")
|
||
|
||
# 与 ORM 配置保持一致（use_tz=False），返回本地时区的“朴素”时间
|
||
return dt_cn.replace(tzinfo=None)
|
||
|
||
|
||
# 自动把 datetime 序列化为 13位时间戳的基类
|
||
class TimestampModel(BaseModel):
|
||
"""自动把 datetime 序列化为 13位时间戳的基类"""
|
||
|
||
model_config = {"arbitrary_types_allowed": True}
|
||
|
||
@field_serializer("*", when_used="json", check_fields=False) # "*" 表示作用于所有字段
|
||
def serialize_datetime(self, value):
|
||
if isinstance(value, datetime):
|
||
return int(value.timestamp()*1000) # 转成 13 位 int 时间戳
|
||
return value
|