[#2] master模块开发测试

src/master/app/util.py

@@ -0,0 +1,51 @@
+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)