"""Provenance utilities: hashing, code version, timestamps, append-only ledgers.""" from __future__ import annotations import hashlib, json, os, subprocess, datetime, platform from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[2] def sha256_file(path: Path, chunk: int = 1 << 20) -> str: h = hashlib.sha256() with open(path, "rb") as f: while True: b = f.read(chunk) if not b: break h.update(b) return h.hexdigest() def sha256_text(s: str) -> str: return hashlib.sha256(s.encode("utf-8")).hexdigest() def now_iso() -> str: return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds") def code_version() -> str: try: out = subprocess.run(["git", "-C", str(ROOT), "rev-parse", "--short", "HEAD"], capture_output=True, text=True, timeout=5) v = out.stdout.strip() or "nogit" dirty = subprocess.run(["git", "-C", str(ROOT), "status", "--porcelain"], capture_output=True, text=True, timeout=5).stdout.strip() return v + ("-dirty" if dirty else "") except Exception: return "nogit" def host_info() -> dict: return {"platform": platform.platform(), "python": platform.python_version(), "machine": platform.machine(), "node": platform.node()} def stamp(extra: dict | None = None) -> dict: d = {"timestamp": now_iso(), "code_version": code_version(), "host": host_info()} if extra: d.update(extra) return d class Ledger: """Append-only JSONL ledger.""" def __init__(self, path: Path): self.path = Path(path); self.path.parent.mkdir(parents=True, exist_ok=True) def append(self, record: dict[str, Any]) -> None: rec = {"ts": now_iso(), **record} with open(self.path, "a") as f: f.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n") def read(self) -> list[dict]: if not self.path.exists(): return [] return [json.loads(l) for l in self.path.read_text().splitlines() if l.strip()] def expand(p: str | Path) -> Path: return Path(os.path.expanduser(str(p))).resolve()