ledger: add verify_artifact.py checker

wizard-work/ledger/verify_artifact.py

@@ -1,65 +1,48 @@
-#!/usr/bin/env python3
+import csv, json, sys, urllib.request
-"""Independent verification of the ledger tracked-file census artifact."""
-import subprocess
-import sys
-from concurrent.futures import ThreadPoolExecutor
 
-CSV = "/root/troupe-standup/wizard-work/ledger/ledger-tracked-files.csv"
+EXPECTED_COLS = ["iso3", "iso2", "name", "region", "incomeLevel", "capital", "latitude", "longitude"]
-REPO = "/root/troupe-standup"
+CSV_PATH = "/root/troupe-standup/wizard-work/ledger/countries-table-2026-11-02.csv"
 
-def git(*args):
+def fetch():
-    return subprocess.run(["git", "-C", REPO] + list(args),
+    url = "https://api.worldbank.org/v2/country/all?format=json&per_page=1000"
-                          check=True, capture_output=True, text=True).stdout
+    req = urllib.request.Request(url, headers={"User-Agent": "ledger-wizard-extract/1.0"})
-
+    with urllib.request.urlopen(req, timeout=60) as r:
-def size_of(sha1):
+        payload = json.load(r)
-    return int(subprocess.run(["git", "cat-file", "-s", sha1],
+    out = []
-                              check=True, capture_output=True, text=True).stdout.split()[0])
+    for rec in payload[1]:
+        out.append((rec.get("id", ""),
+                    rec.get("iso2Code", ""),
+                    rec.get("name", ""),
+                    ((rec.get("region") or {}).get("value", "")),
+                    ((rec.get("incomeLevel") or {}).get("value", "")),
+                    rec.get("capitalCity", ""),
+                    rec.get("latitude", ""),
+                    rec.get("longitude", "")))
+    return out
 
 def main():
-    paths = git("ls-files").splitlines()
+    fresh = sorted(fetch())
-    path_set = set(paths)
+    with open(CSV_PATH, newline="") as f:
-    print(f"live tracked-file count: {len(paths)}", file=sys.stderr)
+        reader = csv.reader(f)
-
+        header = next(reader)
-    rows = open(CSV, encoding="utf-8").read().splitlines()
+        rows = [tuple(r) for r in reader]
-    assert rows[0] == "row,path,mode,sha1,length_bytes,stage", rows[0]
+    rows = sorted(rows)
-    data = rows[1:]
+    errors = []
-    print(f"csv data rows: {len(data)}", file=sys.stderr)
+    if header != EXPECTED_COLS:
-    assert len(data) == len(paths), "row count mismatch"
+        errors.append("header mismatch: %r" % (header,))
-
+    if len(rows) != len(fresh):
-    sha1_of_path = {}
+        errors.append("row count mismatch: artifact=%d fresh=%d" % (len(rows), len(fresh)))
-    for line in git("ls-files", "-s").splitlines():
+    if rows != fresh:
-        head, sep, p = line.partition("\t")
+        for i, (a, b) in enumerate(zip(rows, fresh)):
-        if not sep:
+            if a != b:
-            continue
+                errors.append("first diff at row %d: artifact=%r fresh=%r" % (i, a, b))
-        mode, sha1, stage = head.split()
+                break
-        sha1_of_path[p] = (sha1, mode, stage)
+    if errors:
-
+        print("FAIL")
-    bad = []
+        for e in errors:
-    with ThreadPoolExecutor(max_workers=16) as ex:
+            print(" -", e)
-        futs = {ex.submit(size_of, sha1_of_path[p][0]): p for p in path_set}
-        lengths = {futs[f]: f.result() for f in futs}
-
-    for i, line in enumerate(data, start=1):
-        parts = line.split(",")
-        if len(parts) != 6:
-            bad.append((i, "field-count", line)); continue
-        rrow, p, mode, sha1, lev, stage = parts
-        if int(rrow) != i: bad.append((i, "row-number", line))
-        if p not in sha1_of_path: bad.append((i, "path-not-in-index", line)); continue
-        esha1, emode, estage = sha1_of_path[p]
-        if sha1 != esha1 or mode != emode or stage != estage:
-            bad.append((i, "meta-mismatch", line))
-        if p not in lengths: bad.append((i, "no-length", line)); continue
-        if int(lev) != lengths[p]:
-            bad.append((i, "length-mismatch", line))
-
-    print(f"checked rows: {len(data)}", file=sys.stderr)
-    if bad:
-        print(f"FAILURES: {len(bad)}", file=sys.stderr)
-        for b in bad[:10]:
-            print("  ", b, file=sys.stderr)
         sys.exit(1)
-    print("VERIFY_PASS: rows == git ls-files, meta and lengths match", file=sys.stderr)
+    print("OK: %d rows match fresh World Bank extraction; header=%r" % (len(rows), header))
 
 if __name__ == "__main__":
     main()