1"""Shared helpers: project root, settings, logging, run ledger, output manifest."""2from __future__ import annotations3import hashlib, json, logging, os, sqlite3, sys, time, pathlib4import yaml56ROOT = pathlib.Path(__file__).resolve().parents[1]7SETTINGS = yaml.safe_load((ROOT / "config" / "settings.yaml").read_text())8SEED = int(SETTINGS["seed"])9REGIONS = list(SETTINGS["regions"])10P = {k: ROOT / v for k, v in SETTINGS["paths"].items()}11for _p in P.values():12 _p.mkdir(parents=True, exist_ok=True)131415def get_logger(name: str) -> logging.Logger:16 log = logging.getLogger(name)17 if not log.handlers:18 log.setLevel(logging.INFO)19 fmt = logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s", "%H:%M:%S")20 h = logging.StreamHandler(sys.stdout); h.setFormatter(fmt); log.addHandler(h)21 fh = logging.FileHandler(P["logs"] / "pipeline.log"); fh.setFormatter(fmt); log.addHandler(fh)22 return log232425def snapshot_conn() -> sqlite3.Connection:26 path = ROOT / SETTINGS["snapshot"]["path"]27 return sqlite3.connect(f"file:{path}?mode=ro", uri=True)282930def sha256_file(path: pathlib.Path) -> str:31 h = hashlib.sha256()32 with open(path, "rb") as f:33 for chunk in iter(lambda: f.read(1 << 22), b""):34 h.update(chunk)35 return h.hexdigest()363738def ledger(step: str, **kw) -> None:39 """Append one JSON line to logs/run_ledger.jsonl (what ran, when, with which inputs/outputs)."""40 rec = {"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "step": step, **kw}41 with open(P["logs"] / "run_ledger.jsonl", "a") as f:42 f.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n")434445def register_output(path: pathlib.Path, step: str) -> None:46 """Record sha256 of a produced file in logs/manifest.json."""47 mf = P["logs"] / "manifest.json"48 d = json.loads(mf.read_text()) if mf.exists() else {}49 d[str(path.relative_to(ROOT))] = {"sha256": sha256_file(path), "bytes": path.stat().st_size, "step": step,50 "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}51 mf.write_text(json.dumps(d, indent=1, sort_keys=True))525354def write_json(obj, path: pathlib.Path, step: str = "") -> None:55 path.parent.mkdir(parents=True, exist_ok=True)56 path.write_text(json.dumps(obj, indent=1, ensure_ascii=False, default=str))57 if step:58 register_output(path, step)59