ledger: us nominal gdp annual 2018-2025 csv

wizard-work/ledger/verify_github_python_repos.py

@@ -0,0 +1,149 @@
+#!/usr/bin/env python3
+"""Verify the GitHub top-100 Python repositories artifact against a fresh API fetch.
+
+Re-queries the same search endpoint and diffs the produced table against the
+committed artifact (github-python-repos-<STAMP>.csv / .json). Exit 0 only when
+every strict field matches the live response within live-counter tolerance;
+nonzero otherwise.
+
+Conventions followed:
+  - Strict fields (must match exactly): rank, full_name, owner, language,
+    forks, open_issues, description (whitespace-normalised), html_url.
+  - stars (live counter): the artifact matches when it equals the fresh read or
+    differs by exactly 1; any larger delta is a FAIL. Each off-by-one is printed
+    as a NOTE with the delta.
+"""
+
+import csv
+import json
+import os
+import re
+import sys
+import urllib.request
+from datetime import datetime, timezone
+
+API_BASE = "https://api.github.com"
+QUERY = "language:python"
+SORT = "stars"
+ORDER = "desc"
+PER_PAGE = 100
+HEADERS = {
+    "User-Agent": "ledger/github-python-repos-verify",
+    "Accept": "application/vnd.github+json",
+}
+
+OUT_DIR = os.path.dirname(os.path.abspath(__file__))
+STAMPS = sorted(
+    f for f in os.listdir(OUT_DIR)
+    if re.match(r"github-python-repos-\d{4}-\d{2}-\d{2}\.csv$", f)
+)
+ARTIFACT = os.path.join(OUT_DIR, STAMPS[-1]) if STAMPS else None
+CSV_PATH = ARTIFACT
+JSON_PATH = os.path.splitext(ARTIFACT)[0] + ".json"
+STAR_TOLERANCE = 1
+
+STRICT_FIELDS = [
+    "rank", "full_name", "owner", "language", "forks", "open_issues",
+    "description", "html_url",
+]
+
+
+def api_get(path):
+    req = urllib.request.Request(API_BASE + path, headers=HEADERS, method="GET")
+    with urllib.request.urlopen(req, timeout=60) as r:
+        return json.loads(r.read().decode("utf-8"))
+
+
+def norm(s):
+    return re.sub(r"\s+", " ", (s or "").strip())
+
+
+def load_artifact():
+    rows = []
+    with open(CSV_PATH, newline="", encoding="utf-8") as f:
+        for r in csv.DictReader(f):
+            rows.append(r)
+    return rows
+
+
+def fresh_rows():
+    data = api_get(
+        "/search/repositories?q=language:python&sort=stars&order=desc&per_page=100"
+    )
+    out = []
+    for i, it in enumerate(data.get("items", []), 1):
+        out.append({
+            "rank": i,
+            "full_name": it.get("full_name", ""),
+            "owner": (it.get("owner") or {}).get("login", ""),
+            "description": norm(it.get("description")),
+            "language": it.get("language") or "",
+            "stars": int(it.get("stargazers_count", 0)),
+            "forks": int(it.get("forks_count", 0)),
+            "open_issues": int(it.get("open_issues_count", 0)),
+            "created_at": it.get("created_at", ""),
+            "updated_at": it.get("updated_at", ""),
+            "homepage": it.get("homepage") or "",
+            "html_url": it.get("html_url", ""),
+        })
+    return out
+
+
+def main():
+    if ARTIFACT is None:
+        print("FAIL: no github-python-repos-*.csv found in {}".format(OUT_DIR))
+        return 1
+
+    art = load_artifact()
+    fresh = fresh_rows()
+    print("artifact: {} (rows {})".format(CSV_PATH, len(art)))
+    fresh_payload = api_get(
+        "/search/repositories?q=language:python&sort=stars&order=desc&per_page=100"
+    )
+    print("fresh API rows: {} (total_count from payload: {})".format(len(fresh), fresh_payload.get("total_count")))
+
+    failures = []
+    notes = []
+    n = max(len(art), len(fresh))
+    for i in range(n):
+        a = art[i] if i < len(art) else None
+        f = fresh[i] if i < len(fresh) else None
+        if a is None or f is None:
+            failures.append((i + 1, "count mismatch", a, f))
+            continue
+        for field in STRICT_FIELDS:
+            av = a.get(field, "")
+            fv = f.get(field, "")
+            if field == "description":
+                av = norm(av)
+                fv = norm(fv)
+            if str(av) != str(fv):
+                failures.append((i + 1, field, av, fv))
+                break
+        da = int(a.get("stars", 0))
+        df = int(f.get("stars", 0))
+        if da != df:
+            delta = df - da
+            if abs(delta) > STAR_TOLERANCE:
+                failures.append((i + 1, "stars", da, df))
+            else:
+                notes.append((i + 1, da, df, delta))
+
+    if failures:
+        print("FAIL: {} row(s) differ from the live API response".format(len(failures)))
+        for rank, field, av, fv in failures[:10]:
+            print("  rank={} field={!r} artifact={!r} fresh={!r}".format(rank, field, av, fv))
+        return 1
+
+    for rank, da, df, delta in notes:
+        print("  NOTE rank={} stars artifact={} fresh={} delta={}".format(rank, da, df, delta))
+
+    if notes:
+        print("verdict: PASS - structural fields match the live API ({} star-count NOTE(s))".format(len(notes)))
+    else:
+        print("verdict: PASS - artifact matches the live API response exactly ({} rows)".format(len(fresh)))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())