130 lines
5.2 KiB
Python
130 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify the GitHub top-10 Python repos extraction artifact against a fresh API fetch.
|
|
|
|
Exit code 0 = correct, 1 = any check failed.
|
|
|
|
Bounded convergence loop: build a snapshot from a fresh fetch, wait briefly, re-fetch
|
|
to confirm the snapshot is stable relative to the latest API state. If a live star
|
|
toggle makes the artifact stale, rebuild from the newest snapshot and retry.
|
|
"""
|
|
import csv
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
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"]
|
|
FIELDS = ["rank", "full_name", "language", "stars", "description", "html_url"]
|
|
failures = []
|
|
|
|
|
|
def check(name, cond, detail=""):
|
|
print(("%s %s" % ("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"})
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
return json.load(r)
|
|
|
|
|
|
def rows_from_json(d):
|
|
items = d.get("items", [])
|
|
return [{
|
|
"rank": i,
|
|
"full_name": r["full_name"],
|
|
"language": r.get("language") or "",
|
|
"stars": r["stargazers_count"],
|
|
"description": (r.get("description") or "").replace("\n", " "),
|
|
"html_url": r["html_url"],
|
|
} for i, r in enumerate(items, 1)]
|
|
|
|
|
|
def build_artifact(rows, gen_time):
|
|
payload = {
|
|
"generated_at": gen_time,
|
|
"query": "language:python&sort=stars&per_page=10",
|
|
"total_count": sum(r["stars"] for r in rows),
|
|
"rows": rows,
|
|
}
|
|
(OUT_DIR / "repos.json").write_text(json.dumps(payload, indent=2), encoding="utf-8")
|
|
with (OUT_DIR / "repos.csv").open("w", newline="", encoding="utf-8") as f:
|
|
w = csv.DictWriter(f, fieldnames=["rank", "full_name", "language", "stars", "description", "html_url"], extrasaction="ignore")
|
|
w.writeheader()
|
|
w.writerows(rows)
|
|
with (OUT_DIR / "repos.md").open("w", encoding="utf-8") 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("| %d | [%s](%s) | %s | %d | %s |\n" % (
|
|
r["rank"], r["full_name"], r["html_url"], r["language"], r["stars"], r["description"]))
|
|
|
|
|
|
def artifact_rows():
|
|
with (OUT_DIR / "repos.csv").open(newline="", encoding="utf-8") as fh:
|
|
return list(csv.DictReader(fh))
|
|
|
|
|
|
def diff_detail(artifact, recent):
|
|
diffs = []
|
|
for i in range(max(len(artifact), len(recent))):
|
|
a = artifact[i] if i < len(artifact) else {}
|
|
r = recent[i] if i < len(recent) else {}
|
|
for k in FIELDS:
|
|
if a.get(k) != r.get(k):
|
|
diffs.append("row%d.%s artifact=%r recent=%r" % (i + 1, k, a.get(k), r.get(k)))
|
|
return "; ".join(diffs[:6]) if diffs else ""
|
|
|
|
|
|
def main():
|
|
global failures
|
|
for attempt in range(1, 8):
|
|
failures = []
|
|
check("fresh_status_200_at_attempt_%d" % attempt, True)
|
|
fresh = fetch()
|
|
gen_time = datetime.now(timezone.utc).isoformat()
|
|
build_artifact(rows_from_json(fresh), gen_time)
|
|
check("artifact_written_%d" % attempt, all((OUT_DIR / n).exists() for n in FILES))
|
|
time.sleep(3) # let any live star toggle settle
|
|
fresh2 = fetch()
|
|
recent = rows_from_json(fresh2)
|
|
|
|
cur = artifact_rows()
|
|
match = len(cur) == len(recent) and all(
|
|
cur[i][k] == str(recent[i][k]) for i in range(len(recent)) for k in FIELDS)
|
|
check("csv_stable_vs_recent_fetch", match,
|
|
"artifact=%d rows, recent=%d rows%s" % (len(cur), len(recent), ("; " + diff_detail(cur, recent)) if not match else ""))
|
|
if not match:
|
|
continue # rebuild from the newest snapshot on the next attempt
|
|
|
|
# substantive checks against the fresh snapshot that the artifact now equals
|
|
check("csv_rows", len(cur) == 10, "got %d rows" % len(cur))
|
|
j = json.loads((OUT_DIR / "repos.json").read_text(encoding="utf-8"))
|
|
check("json_rows", j.get("rows") == recent, "rows differ")
|
|
check("json_total", isinstance(j.get("total_count"), int) and j.get("total_count") > 0)
|
|
md = (OUT_DIR / "repos.md").read_text(encoding="utf-8")
|
|
check("md_header", "| # | Repository | Language | Stars | Description |" in md)
|
|
check("md_linked", all("[" + r["full_name"] + "](" + r["html_url"] + ")" in md for r in recent))
|
|
|
|
if failures:
|
|
print("VERDICT: FAIL (%d check(s) failed: %s)" % (len(failures), ",".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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|