""" 统一时间格式化工具 背景: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