150 lines
4.8 KiB
Python
150 lines
4.8 KiB
Python
#!/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())
|