strepsiades
85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
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() |