1"""Dataset manifests, hashes and code versions (no raw data, no secrets)."""2from __future__ import annotations34import hashlib5import json6import os7import subprocess8import time9from pathlib import Path10from typing import Any, Iterable111213def sha256_file(path: str | os.PathLike, chunk: int = 1 << 20) -> str:14 h = hashlib.sha256()15 with open(path, "rb") as fh:16 for block in iter(lambda: fh.read(chunk), b""):17 h.update(block)18 return h.hexdigest()192021def sha256_text(text: str) -> str:22 return hashlib.sha256(text.encode()).hexdigest()232425def git_sha(repo: str | os.PathLike | None = None) -> str:26 try:27 return subprocess.check_output(["git", "rev-parse", "--short=12", "HEAD"], cwd=repo, text=True,28 stderr=subprocess.DEVNULL).strip()29 except Exception:30 return "nogit"313233def now_iso() -> str:34 return time.strftime("%Y-%m-%dT%H:%M:%S%z")353637def append_jsonl(path: str | os.PathLike, record: dict[str, Any]) -> None:38 p = Path(path)39 p.parent.mkdir(parents=True, exist_ok=True)40 with open(p, "a") as fh:41 fh.write(json.dumps(record, default=str, sort_keys=True) + "\n")424344def read_jsonl(path: str | os.PathLike) -> list[dict[str, Any]]:45 p = Path(path)46 if not p.exists():47 return []48 with open(p) as fh:49 return [json.loads(line) for line in fh if line.strip()]505152def manifest_index(path: str | os.PathLike, keys: Iterable[str]) -> dict[tuple, dict[str, Any]]:53 keys = tuple(keys)54 return {tuple(r.get(k) for k in keys): r for r in read_jsonl(path)}555657def write_json(path: str | os.PathLike, obj: Any) -> None:58 p = Path(path)59 p.parent.mkdir(parents=True, exist_ok=True)60 with open(p, "w") as fh:61 json.dump(obj, fh, indent=1, default=str, sort_keys=True)62