troupe-standup/wizard-work/ledger/verify_github_repos.py

#!/usr/bin/env python3
# Q63 (2026-10-24): no committed gate invokes this file -- `git grep -F -n verify_github_repos.py HEAD
#   -- '*.py' '*.sh' CHECK` matches nothing outside this file -- so it is a one-off
#   artifact, run by hand against its subject, not a standing check.
"""Verify the GitHub-repos-by-stars extraction artifact.

Exit code 0 = correct, 1 = any check failed.

Checks:
  A. Structure of github_repos_stars.{csv,json}
  B. md5 fingerprint of the CSV matches the stored fingerprint
  C. Reproducibility: re-fetch the SAME search and diff the derived rows
  D. Live cross-check: rank-1 repo reported by the /repos/ endpoint matches
     the artifact's rank-1 cells
No network is needed for A/B; C and D are cross-source anchors.
"""

import csv
import hashlib
import json
import sys
import urllib.request

OUT = sys.argv[1] if len(sys.argv) > 1 else "/root/troupe-standup/wizard-work/ledger"
CSV_PATH = OUT + "/github_repos_stars.csv"
JSON_PATH = OUT + "/github_repos_stars.json"
MD5_PATH = OUT + "/github_repos_stars.md5"
SEARCH_URL = (
    "https://api.github.com/search/repositories"
    "?q=stars:>10000&sort=stars&order=desc&per_page=100"
)
HEADERS = {"User-Agent": "ledger-wizard"}

REQUIRED = [
    "rank", "full_name", "owner", "description", "language",
    "stargazers_count", "updated_at", "html_url",
]

failures = []


def check(name, cond, msg=None):
    if not cond:
        failures.append((name, msg or ""))


def derive_rows(items):
    return [
        {
            "rank": i + 1,
            "full_name": it["full_name"],
            "owner": (it.get("owner") or {}).get("login", ""),
            "description": it.get("description") or "",
            "language": it.get("language") or "",
            "stargazers_count": it.get("stargazers_count"),
            "updated_at": it.get("updated_at") or "",
            "html_url": it.get("html_url") or "",
        }
        for i, it in enumerate(items[:100])
    ]


def fetch_json(url):
    req = urllib.request.Request(url, headers=HEADERS)
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.load(r)


def load_artifact():
    with open(CSV_PATH, encoding="utf-8", newline="") as fh:
        reader = csv.DictReader(fh)
        check("csv.headers", reader.fieldnames == REQUIRED,
              "got %r" % (reader.fieldnames,))
        csv_rows = list(reader)
    with open(JSON_PATH, encoding="utf-8") as fh:
        json_rows = json.load(fh)

    norm = []
    for r in csv_rows:
        norm.append({
            "rank": int(r["rank"]),
            "full_name": r["full_name"],
            "owner": r["owner"],
            "description": r["description"],
            "language": r["language"],
            "stargazers_count": int(r["stargazers_count"]),
            "updated_at": r["updated_at"],
            "html_url": r["html_url"],
        })
    check("json.match_csv", json_rows == norm,
          "JSON row count %d vs CSV %d" % (len(json_rows), len(csv_rows)))
    return norm


def main():
    # --- A. structure ------------------------------------------------------
    rows = load_artifact()
    check("csv.rowcount", len(rows) == 100, "got %d" % len(rows))
    check("csv.ranks", [r["rank"] for r in rows] == list(range(1, 101)))
    check("csv.no_empty_required",
          all(r["full_name"] and r["owner"] and r["html_url"]
              and r["updated_at"] for r in rows))
    check("csv.stars_int_positive",
          all(isinstance(r["stargazers_count"], int) and r["stargazers_count"] > 0
              for r in rows))
    desc_nonempty = sum(1 for r in rows if r["description"].strip())
    check("csv.description_nonempty_ge_90", desc_nonempty >= 90,
          "%d/100 non-empty" % desc_nonempty)
    # non-increasing star counts (strict descending would be ideal; allow ties)
    stars = [r["stargazers_count"] for r in rows]
    check("csv.sorted_desc",
          all(stars[i] >= stars[i + 1] for i in range(len(stars) - 1)))

    # --- B. md5 fingerprint ------------------------------------------------
    digest = hashlib.md5(open(CSV_PATH, "rb").read()).hexdigest()
    stored = open(MD5_PATH, encoding="utf-8").read().split()[0]
    check("md5.match", digest == stored, "computed %s vs stored %s" % (digest, stored))

    # --- C. live cross-check (identity is stable; star counts move, so we
    # cross-check identity + a wide count bound; the full re-fetch is not
    # bit-identical between requests because the source mutates). -----------
    data = fetch_json(SEARCH_URL)
    check("search.total_count",
          isinstance(data.get("total_count"), int)
          and data.get("total_count") >= 1000,
          "total_count %r" % data.get("total_count"))

    top = rows[0]

    # --- D. live spot-check (rank 1) ---------------------------------------
    # Identity is stable; star counts move between requests, so we only
    # cross-check stable fields and a wide sanity bound on the count.
    live = fetch_json("https://api.github.com/repos/%s" % top["full_name"])
    live_owner = (live.get("owner") or {}).get("login", "")
    check("live.rank1_identity",
          live.get("full_name") == top["full_name"]
          and live_owner == top["owner"],
          "live %r/%r vs artifact %r/%r" % (
              live.get("full_name"), live_owner, top["full_name"], top["owner"]))
    live_stars = live.get("stargazers_count")
    check("live.rank1_stars_positive",
          isinstance(live_stars, int) and live_stars >= 100000,
          "live stars %r" % live_stars)

    if failures:
        print("artifact rows=%d total_count=%s" % (len(rows), data.get("total_count")))
        print("FAILED CHECKS:")
        for name, msg in failures:
            print("  - %s%s" % (name, (" : " + msg) if msg else ""))
        sys.exit(1)

    print("member_stars: %d | rank1=%s (%s) | md5=%s" % (
        len(rows), top["full_name"], top["stargazers_count"], digest))
    print("ALL CHECKS PASSED")


if __name__ == "__main__":
    main()