"""Lightweight file-based audit + error logging for Thermograph. Two log streams, written as JSON Lines (one self-contained JSON object per line): * ``logs/audit/audit-.jsonl`` — one line per graded run, recording the total wall-clock duration, a per-phase timing breakdown, how many days the run covered (the recent-window size and the historical span), and a ``run_type`` tag of **full** (a fresh ~45-year history fetch happened) or **partial** (the history came from cache, so only the recent days were fetched). * ``logs/errors/errors-.jsonl`` — a *separate folder* for anything that went wrong: each upstream failure is tagged ``retry`` (another attempt will be made) or ``error`` (final failure), and whole-run failures are tagged ``error``. Everything is best-effort: logging never raises into the request path. """ import contextlib import contextvars import datetime import json import os import threading import time import uuid LOG_ROOT = os.path.join(os.path.dirname(__file__), "..", "logs") AUDIT_DIR = os.path.join(LOG_ROOT, "audit") ERROR_DIR = os.path.join(LOG_ROOT, "errors") _lock = threading.Lock() # Set for the duration of a run so error/retry logs from deep in the fetch layer # can be correlated back to the run that triggered them. current_run_id: "contextvars.ContextVar[str | None]" = contextvars.ContextVar( "thermograph_run_id", default=None ) def _now_iso() -> str: return datetime.datetime.now().astimezone().isoformat(timespec="milliseconds") def _today() -> str: return datetime.date.today().isoformat() def _append(path: str, record: dict) -> None: try: os.makedirs(os.path.dirname(path), exist_ok=True) line = json.dumps(record, default=str) with _lock: with open(path, "a", encoding="utf-8") as f: f.write(line + "\n") except Exception: # noqa: BLE001 - logging must never break the request pass def log_event(tag: str, record: dict) -> dict: """Record an error or retry in the separate errors folder. ``tag`` is conventionally ``"retry"`` or ``"error"``. """ rec = {"ts": _now_iso(), "tag": tag, "run_id": current_run_id.get(), **record} _append(os.path.join(ERROR_DIR, f"errors-{_today()}.jsonl"), rec) return rec class RunAudit: """Times a graded run and writes one audit line when the ``with`` block exits. Usage:: with audit.RunAudit(endpoint="grade", lat=..., lon=...) as run: with run.phase("history"): ... run.set(run_type="full", history_rows=n) return response """ def __init__(self, **params): self.run_id = uuid.uuid4().hex[:12] self.fields = {"run_id": self.run_id, **params} self.phases: dict[str, float] = {} self._t0 = time.perf_counter() self._token = None def __enter__(self) -> "RunAudit": self._token = current_run_id.set(self.run_id) return self def set(self, **kw) -> "RunAudit": """Attach extra fields (e.g. run_type, history span) to the audit record.""" self.fields.update(kw) return self @contextlib.contextmanager def phase(self, name: str): """Time a named part of the run; the duration (ms) lands in ``phases``.""" start = time.perf_counter() try: yield finally: self.phases[name] = round((time.perf_counter() - start) * 1000, 1) def __exit__(self, exc_type, exc, tb) -> bool: total = round((time.perf_counter() - self._t0) * 1000, 1) status = "ok" if exc is not None: status = "error" self.fields["error"] = repr(exc) code = getattr(exc, "status_code", None) if code is not None: self.fields["http_status"] = code # Mirror whole-run failures into the separate errors folder. log_event("error", { "phase": "run", "error": repr(exc), **{k: self.fields.get(k) for k in ("endpoint", "lat", "lon", "cell_id")}, }) record = { "ts": _now_iso(), "status": status, "duration_ms": total, "phases": self.phases, **self.fields, } _append(os.path.join(AUDIT_DIR, f"audit-{_today()}.jsonl"), record) if self._token is not None: current_run_id.reset(self._token) return False # never suppress the exception