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"])10LABEL = os.environ.get("WP38_LABEL", "rule") # "rule" (platform keyword labels) or "model" (A-002 classifier labels)11SUFFIX = "" if LABEL == "rule" else f"_{LABEL}"12P = {k: ROOT / v for k, v in SETTINGS["paths"].items()}13P["processed"] = ROOT / f"data/processed{SUFFIX}"; P["experiments"] = ROOT / f"experiments{SUFFIX}"14P["figures"] = ROOT / f"paper/figures{SUFFIX}"; P["tables"] = ROOT / f"paper/tables{SUFFIX}"15for _p in P.values():16 _p.mkdir(parents=True, exist_ok=True)17PROC_DIR, EXP_DIR = P["processed"], P["experiments"]18BASE_PROC, BASE_EXP = ROOT / "data/processed", ROOT / "experiments" # label-independent artefacts (cleaning, audits, matching)19BASE_PROC.mkdir(parents=True, exist_ok=True); BASE_EXP.mkdir(parents=True, exist_ok=True)202122def get_logger(name: str) -> logging.Logger:23 log = logging.getLogger(name)24 if not log.handlers:25 log.setLevel(logging.INFO)26 fmt = logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s", "%H:%M:%S")27 h = logging.StreamHandler(sys.stdout); h.setFormatter(fmt); log.addHandler(h)28 fh = logging.FileHandler(P["logs"] / "pipeline.log"); fh.setFormatter(fmt); log.addHandler(fh)29 return log303132def snapshot_conn() -> sqlite3.Connection:33 path = ROOT / SETTINGS["snapshot"]["path"]34 return sqlite3.connect(f"file:{path}?mode=ro", uri=True)353637def sha256_file(path: pathlib.Path) -> str:38 h = hashlib.sha256()39 with open(path, "rb") as f:40 for chunk in iter(lambda: f.read(1 << 22), b""):41 h.update(chunk)42 return h.hexdigest()434445def ledger(step: str, **kw) -> None:46 """Append one JSON line to logs/run_ledger.jsonl (what ran, when, with which inputs/outputs)."""47 rec = {"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "step": step, **kw}48 with open(P["logs"] / "run_ledger.jsonl", "a") as f:49 f.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n")505152def register_output(path: pathlib.Path, step: str) -> None:53 """Record sha256 of a produced file in logs/manifest.json."""54 mf = P["logs"] / "manifest.json"55 d = json.loads(mf.read_text()) if mf.exists() else {}56 d[str(path.relative_to(ROOT))] = {"sha256": sha256_file(path), "bytes": path.stat().st_size, "step": step,57 "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}58 mf.write_text(json.dumps(d, indent=1, sort_keys=True))596061def write_json(obj, path: pathlib.Path, step: str = "") -> None:62 path.parent.mkdir(parents=True, exist_ok=True)63 path.write_text(json.dumps(obj, indent=1, ensure_ascii=False, default=str))64 if step:65 register_output(path, step)66