52 lines
1.4 KiB
Python
52 lines
1.4 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import tempfile
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Iterable
|
||
|
||
|
||
ISO_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
|
||
|
||
|
||
def utcnow() -> datetime:
|
||
"""获取当前 UTC 时间，统一时间基准。"""
|
||
return datetime.now(timezone.utc)
|
||
|
||
|
||
def to_iso(dt: datetime | None) -> str | None:
|
||
if dt is None:
|
||
return None
|
||
return dt.astimezone(timezone.utc).replace(microsecond=0).strftime(ISO_FORMAT)
|
||
|
||
|
||
def parse_iso(value: str | None) -> datetime | None:
|
||
if not value:
|
||
return None
|
||
try:
|
||
if value.endswith("Z"):
|
||
return datetime.strptime(value, ISO_FORMAT).replace(tzinfo=timezone.utc)
|
||
# Fallback for ISO strings with offset
|
||
return datetime.fromisoformat(value).astimezone(timezone.utc)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def ensure_parent(path: str) -> None:
|
||
"""确保目标文件所在目录存在。"""
|
||
Path(path).parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
def atomic_write_json(path: str, data: Iterable[Any] | Any) -> None:
|
||
"""原子化写 JSON，避免被其它进程读到半成品。"""
|
||
ensure_parent(path)
|
||
directory = Path(path).parent
|
||
with tempfile.NamedTemporaryFile("w", dir=directory, delete=False) as tmp:
|
||
json.dump(data, tmp, separators=(",", ":"))
|
||
tmp.flush()
|
||
os.fsync(tmp.fileno())
|
||
temp_path = tmp.name
|
||
os.replace(temp_path, path)
|