1"""Provenance utilities: hashing, code version, timestamps, append-only ledgers."""2from __future__ import annotations3import hashlib, json, os, subprocess, datetime, platform4from pathlib import Path5from typing import Any67ROOT = Path(__file__).resolve().parents[2]89def sha256_file(path: Path, chunk: int = 1 << 20) -> str:10 h = hashlib.sha256()11 with open(path, "rb") as f:12 while True:13 b = f.read(chunk)14 if not b: break15 h.update(b)16 return h.hexdigest()1718def sha256_text(s: str) -> str:19 return hashlib.sha256(s.encode("utf-8")).hexdigest()2021def now_iso() -> str:22 return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")2324def code_version() -> str:25 try:26 out = subprocess.run(["git", "-C", str(ROOT), "rev-parse", "--short", "HEAD"], capture_output=True, text=True, timeout=5)27 v = out.stdout.strip() or "nogit"28 dirty = subprocess.run(["git", "-C", str(ROOT), "status", "--porcelain"], capture_output=True, text=True, timeout=5).stdout.strip()29 return v + ("-dirty" if dirty else "")30 except Exception:31 return "nogit"3233def host_info() -> dict:34 return {"platform": platform.platform(), "python": platform.python_version(), "machine": platform.machine(), "node": platform.node()}3536def stamp(extra: dict | None = None) -> dict:37 d = {"timestamp": now_iso(), "code_version": code_version(), "host": host_info()}38 if extra: d.update(extra)39 return d4041class Ledger:42 """Append-only JSONL ledger."""43 def __init__(self, path: Path):44 self.path = Path(path); self.path.parent.mkdir(parents=True, exist_ok=True)45 def append(self, record: dict[str, Any]) -> None:46 rec = {"ts": now_iso(), **record}47 with open(self.path, "a") as f:48 f.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n")49 def read(self) -> list[dict]:50 if not self.path.exists(): return []51 return [json.loads(l) for l in self.path.read_text().splitlines() if l.strip()]5253def expand(p: str | Path) -> Path:54 return Path(os.path.expanduser(str(p))).resolve()55