| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566 |
- import datetime as dt
- import json
- import os
- import tempfile
- def utc_now_iso():
- """Return current UTC time in ISO 8601 format (e.g. 2026-10-13T12:00:00Z)."""
- return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
- def read_json_file(path, default=None):
- """Read and parse JSON from file, returning `default` if file missing or corrupt."""
- if default is None:
- default = {}
- if not os.path.exists(path):
- return default
- try:
- with open(path, "r", encoding="utf-8") as f:
- return json.load(f)
- except (ValueError, json.JSONDecodeError):
- return default
- def _atomic_write(path, payload_bytes):
- """Write `payload_bytes` to `path` atomically.
- Strategy: write to a sibling temp file, fsync, then os.replace.
- Concurrent readers either see the previous version or the new
- one — never a truncated/empty file. Required for state.json,
- which is the sole fallback source on the next run.
- """
- parent = os.path.dirname(path) or "."
- os.makedirs(parent, exist_ok=True)
- fd, tmp_path = tempfile.mkstemp(
- prefix=".{}.".format(os.path.basename(path)),
- suffix=".tmp",
- dir=parent,
- )
- try:
- with os.fdopen(fd, "wb") as f:
- f.write(payload_bytes)
- f.flush()
- try:
- os.fsync(f.fileno())
- except OSError:
- # fsync may fail on some filesystems (e.g. tmpfs without backing store)
- pass
- os.replace(tmp_path, path)
- except Exception:
- try:
- os.unlink(tmp_path)
- except OSError:
- pass
- raise
- def write_json_file(path, data):
- """Atomically write data as formatted JSON."""
- payload = json.dumps(data, ensure_ascii=True, indent=2)
- _atomic_write(path, payload.encode("utf-8"))
- def write_text_file(path, data):
- """Atomically write text string to file."""
- _atomic_write(path, data.encode("utf-8"))
|
