Your Name
110 lines
4.1 KiB
Python
110 lines
4.1 KiB
Python
"""
|
||
统一时间格式化工具
|
||
|
||
背景：MySQL DATETIME 不带时区，backend 用 datetime.utcnow() 写入 UTC naive datetime。
|
||
原 schema 直接 .isoformat() 输出，前端拿到无时区字符串后当成 Asia/Shanghai 本地时间显示，
|
||
导致所有时间统一少 8 小时。
|
||
|
||
本工具把任意 datetime（naive 或 aware）一律按 UTC 输出，带 'Z' 后缀。
|
||
前端收到 'Z' 后缀字符串后，按 UTC 解析后再 +8 小时显示为 Asia/Shanghai 时间。
|
||
|
||
也兼容 MySQL 返回 naive 时被错当成 local time 的情况：调用方传入数字或字符串时
|
||
不会出错，自动识别。
|
||
"""
|
||
from datetime import datetime, timezone, timedelta
|
||
from typing import Optional, Union
|
||
|
||
|
||
# 业务时区：服务器期望最终用户看到的时区。
|
||
# 目前写死 Asia/Shanghai (UTC+8)；如以后部署到其他时区，改这里即可。
|
||
BUSINESS_TZ = timezone(timedelta(hours=8))
|
||
BUSINESS_TZ_NAME = "Asia/Shanghai"
|
||
|
||
|
||
def to_utc_iso(dt: Optional[datetime]) -> Optional[str]:
|
||
"""
|
||
把 datetime 转成 UTC ISO8601 字符串（带 Z 后缀）。
|
||
- aware datetime 调 astimezone(UTC)
|
||
- naive datetime 假定为 UTC（与 datetime.utcnow() 写入策略一致）
|
||
- None 返回 None
|
||
"""
|
||
if dt is None:
|
||
return None
|
||
if dt.tzinfo is None:
|
||
# naive 一律视为 UTC（与 datetime.utcnow() 写入策略一致）
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
# 用 '+00:00' 替换成 'Z'，更标准的 ISO8601 形式
|
||
return dt.astimezone(timezone.utc).isoformat().replace('+00:00', 'Z')
|
||
|
||
|
||
def to_business_iso(dt: Optional[datetime]) -> Optional[str]:
|
||
"""
|
||
把 datetime 转成业务时区（Asia/Shanghai）的 ISO8601 字符串（带 +08:00 后缀）。
|
||
前端拿到后可直接当本地时间显示。
|
||
"""
|
||
if dt is None:
|
||
return None
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
return dt.astimezone(BUSINESS_TZ).isoformat()
|
||
|
||
|
||
def utc_str_to_business_str(s: Optional[str]) -> Optional[str]:
|
||
"""
|
||
把 'YYYY-MM-DDTHH:MM:SS[.ffffff][Z|+HH:MM]' 字符串按 UTC 解析，
|
||
转成业务时区 ISO8601 字符串。
|
||
用于后端已经输出 UTC 时，前端（或后端自己）做时区转换的辅助函数。
|
||
"""
|
||
if not s:
|
||
return None
|
||
try:
|
||
dt = datetime.fromisoformat(s.replace('Z', '+00:00'))
|
||
return to_business_iso(dt)
|
||
except (ValueError, TypeError):
|
||
return s
|
||
|
||
|
||
def now_business_iso() -> str:
|
||
"""供后端临时插入用：返回当前业务时区时间"""
|
||
return datetime.now(BUSINESS_TZ).isoformat()
|
||
|
||
|
||
def serialize_dt_fields(model, fields: Optional[list] = None) -> dict:
|
||
"""
|
||
把 ORM model 转成 dict，datetime 字段自动用业务时区 ISO 字符串输出。
|
||
|
||
用于 audit/snmp/alerts 等内联 dict 序列化的 endpoint，确保返回的
|
||
时间字段不会因为裸 .isoformat() 输出而少 8 小时。
|
||
|
||
用法：
|
||
return serialize_dt_fields(alert, ['created_at', 'updated_at'])
|
||
return serialize_dt_fields(alert) # 自动检测所有 datetime 字段
|
||
"""
|
||
from sqlalchemy import inspect as sa_inspect
|
||
result = {}
|
||
# inspect Class（不是 instance），可以拿 mapper.columns
|
||
mapper = sa_inspect(type(model)) if not isinstance(model, type) else sa_inspect(model)
|
||
datetime_fields: list = []
|
||
if fields is None:
|
||
for column in mapper.columns:
|
||
col_type = str(column.type).upper()
|
||
if 'DATETIME' in col_type or 'TIMESTAMP' in col_type:
|
||
datetime_fields.append(column.key)
|
||
else:
|
||
datetime_fields = fields
|
||
for column in mapper.columns:
|
||
value = getattr(model, column.key, None)
|
||
if column.key in datetime_fields:
|
||
result[column.key] = to_business_iso(value)
|
||
else:
|
||
if hasattr(value, 'value'): # Enum
|
||
result[column.key] = value.value
|
||
else:
|
||
result[column.key] = value
|
||
return result
|
||
|
||
|
||
def serialize_dt_list(models, fields: Optional[list] = None) -> list:
|
||
"""批量版本：返回每个 model 的 dict 列表"""
|
||
return [serialize_dt_fields(m, fields) for m in models] # type: ignore
|