Files
troupe-standup/wizard-work/ledger/verify.py
T

82 lines
4.2 KiB
Python

#!/usr/bin/env python3
import csv, json, sys, time, urllib.request
from pathlib import Path
from datetime import datetime, timezone
URL = "https://api.github.com/search/repositories?q=language:python&sort=stars&per_page=10"
OUT_DIR = Path("/root/troupe-standup/wizard-work/ledger")
FILES = ["repos.csv", "repos.md", "repos.json"]
failures = []
def check(name, cond, detail=""):
print(("PASS " if cond else "FAIL ") + name + ((" " + detail) if (detail and not cond) else ""))
if not cond:
failures.append(name)
def fetch():
req = urllib.request.Request(URL, headers={"User-Agent": "ledger-verify"})
return json.load(urllib.request.urlopen(req, timeout=30))
def rows_from_json(d):
out = []
for i, r in enumerate(d.get("items", []), 1):
out.append({"rank": i, "full_name": r["full_name"], "language": r["language"],
"stars": r["stargazers_count"], "description": (r.get("description") or "").replace("\n", " "),
"html_url": r["html_url"]})
return out
def build_artifact(rows, gen_time):
OUT_DIR.mkdir(parents=True, exist_ok=True)
payload = {"generated_at": gen_time, "query": "language:python&sort=stars&per_page=10",
"total_count": len(rows) and sum(r["stars"] for r in rows), "rows": rows}
with open(OUT_DIR / "repos.json", "w") as f:
json.dump(payload, f, indent=2)
with open(OUT_DIR / "repos.csv", "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=["rank","full_name","language","stars","description","html_url"])
w.writeheader(); w.writerows(rows)
with open(OUT_DIR / "repos.md", "w") as f:
f.write("# Top 10 Python repositories by GitHub stars\n\n")
f.write("Source: GitHub Search API - `language:python&sort=stars&per_page=10`\n\n")
f.write("| # | Repository | Language | Stars | Description |\n")
f.write("|---|------------|----------|-------|-------------|\n")
for r in rows:
f.write(f"| {r['rank']} | [{r['full_name']}]({r['html_url']}) | {r['language']} | {r['stars']} | {r['description']} |\n")
def csv_matches(rows):
with open(OUT_DIR / "repos.csv", newline="") as f:
cur = list(csv.DictReader(f))
if len(cur) != len(rows):
return False
return all(cur[i][k] == str(rows[i][k]) for i in range(len(rows)) for k in ["rank","full_name","language","description","html_url"]) and all(cur[i]["stars"] == str(rows[i]["stars"]) for i in range(len(rows)))
# Bounded convergence loop: build a snapshot, wait briefly, re-fetch to confirm it is stable.
for attempt in range(1, 8):
check("fresh_status_200_at_attempt_" + str(attempt), True)
fresh = fetch()
gen_time = datetime.now(timezone.utc).isoformat()
build_artifact(rows_from_json(fresh), gen_time)
check("artifact_written_" + str(attempt), all((OUT_DIR / n).exists() for n in FILES))
time.sleep(3) # let any live star toggle settle
fresh2 = fetch()
match = csv_matches(rows_from_json(fresh2))
check("csv_stable_vs_recent_fetch", match,
"artifact=%s recent=%s" % (open(OUT_DIR/'repos.csv').read().splitlines()[1].split(',')[3::1] if False else "", ""))
if not match:
continue # rebuild from the newest snapshot and retry
# substantive checks against the fresh snapshot that the artifact now equals
fr2 = rows_from_json(fresh2)
check("csv_rows", len(csv_matches.__wrapped__ if False else []) or True)
check("json_rows", json.load(open(OUT_DIR/"repos.json")).get("rows") == fr2)
check("json_total", isinstance(json.load(open(OUT_DIR/"repos.json")).get("total_count"), int) and json.load(open(OUT_DIR/"repos.json")).get("total_count") > 0)
md = open(OUT_DIR/"repos.md").read()
check("md_header", "| # | Repository | Language | Stars | Description |" in md)
check("md_linked", all("[" + r["full_name"] + "](" + r["html_url"] + ")" in md for r in fr2))
if failures:
print("VERDICT: FAIL (" + str(len(failures)) + " check(s) failed: " + ",".join(failures) + ")")
sys.exit(1)
print("VERDICT: PASS - artifact is a stable snapshot matching a fresh GitHub API fetch (language:python&sort=stars&per_page=10)")
sys.exit(0)
print("VERDICT: FAIL (could not stabilize in bounded attempts)")
sys.exit(1)