from datetime import datetime from typing import Annotated from pydantic import AfterValidator from functools import lru_cache import re # 预编译正则表达式,避免重复编译 DATE_PATTERN = re.compile(r'^\d{4}-\d{2}-\d{2}$') TIMESTAMP_PATTERNS = { 'iso_z': re.compile(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+Z$'), 'iso_offset': re.compile(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+\+00:00$'), 'simple_space': re.compile(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$'), 'simple_t': re.compile(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}$'), } @lru_cache(maxsize=1024) # 缓存最近1024个不同的日期 def _validate_date_cached(v: str) -> str: """缓存版本的日期验证""" # 快速格式检查(避免无效字符串进入 strptime) if not DATE_PATTERN.match(v): raise ValueError(f"日期格式应为 YYYY-MM-DD,实际为: {v}") # 验证日期有效性(例如排除 2023-02-30) datetime.strptime(v, "%Y-%m-%d") return v def validate_date(v: str) -> str: """校验日期格式为 YYYY-MM-DD""" if v is None or v == "": return v return _validate_date_cached(v) @lru_cache(maxsize=2048) # 时间戳变化更多,给更大的缓存 def _validate_timestamp_cached(v: str) -> str: """缓存版本的时间戳验证""" # 按常见程度排序,优先检查最可能的格式 if TIMESTAMP_PATTERNS['iso_z'].match(v): datetime.strptime(v, "%Y-%m-%dT%H:%M:%S.%fZ") elif TIMESTAMP_PATTERNS['simple_space'].match(v): datetime.strptime(v, "%Y-%m-%d %H:%M:%S") elif TIMESTAMP_PATTERNS['simple_t'].match(v): datetime.strptime(v, "%Y-%m-%dT%H:%M:%S") elif TIMESTAMP_PATTERNS['iso_offset'].match(v): datetime.strptime(v, "%Y-%m-%dT%H:%M:%S.%f+00:00") else: # 兜底逻辑:如果正则都不匹配,尝试原来的逻辑 try: if v.endswith('Z'): datetime.strptime(v, "%Y-%m-%dT%H:%M:%S.%fZ") elif v.endswith('+00:00'): datetime.strptime(v, "%Y-%m-%dT%H:%M:%S.%f+00:00") else: try: datetime.strptime(v, "%Y-%m-%d %H:%M:%S") except ValueError: datetime.strptime(v, "%Y-%m-%dT%H:%M:%S") except ValueError: raise ValueError("时间戳格式应为 YYYY-MM-DDTHH:MM:SS.sssZ、+00:00、YYYY-MM-DD HH:MM:SS 或 YYYY-MM-DDTHH:MM:SS") return v def validate_timestamp(v: str) -> str: """校验时间戳格式""" if v is None or v == "": return v return _validate_timestamp_cached(v) # 定义新类型 DateStr = Annotated[str, AfterValidator(validate_date)] TimestampStr = Annotated[str, AfterValidator(validate_timestamp)] # 可选:提供缓存统计功能用于调试 def get_cache_stats(): """获取缓存统计信息,用于性能调试""" return { 'date_cache': _validate_date_cached.cache_info(), 'timestamp_cache': _validate_timestamp_cached.cache_info(), } def clear_validation_cache(): """清空验证缓存(用于测试或内存清理)""" _validate_date_cached.cache_clear() _validate_timestamp_cached.cache_clear()