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

#!/usr/bin/env python3
"""Extract the top 100 public GitHub repositories by star count (language: python).

Source: GitHub REST Search API, GET /search/repositories
Query: q=language:python&sort=stars&order=desc&per_page=100 (HTTP 200)
Retrieved (UTC): stamp recorded in the sibling JSON on each run.

Outputs (same directory as this script):
  github-python-repos-<STAMP>.csv | .json | .md

Conventions (ledger lane):
  - CSV: one row per repo; description is newline-free, full text (no truncation).
  - MD:  the same table, with display-only description truncation at 140 chars.
  - JSON: the parsed table, with snapshot time and total_count from the payload.
  - Verdict: the sibling verify_github_python_repos.py re-queries the API and
    diffs the artifact against a fresh extraction. exit 0 == artifact matches
    the live API response; exit 1 == mismatch.

Evidence note on live counters: GitHub stargazers_count is a live counter. It
can move by a few stars between two consecutive API reads of the same query.
The verifier therefore agrees the artifact when every *structural* field
(rank, full_name, owner, language, forks, open_issues, description, html_url)
matches exactly and each star count is equal to the fresh read or differs by
exactly 1 (with the delta printed as a NOTE). Anything else is a FAIL.
"""

import csv
import json
import os
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
DISPLAY_LIMIT = 140
HEADERS = {
    "User-Agent": "ledger/github-python-repos",
    "Accept": "application/vnd.github+json",
}
CSV_COLUMNS = [
    "rank", "full_name", "owner", "description", "language", "stars",
    "forks", "open_issues", "created_at", "updated_at", "homepage", "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 clean_text(s, limit=None):
    """Strip newlines, collapse whitespace. Truncate only when limit is given."""
    s = (s or "").replace("\n", " ").replace("\r", " ")
    s = " ".join(s.split())
    if limit is not None:
        s = s[:limit]
    return s


def main():
    q = "q=language:python&sort=stars&order=desc&per_page=100"
    data = api_get("/search/repositories?" + q)
    items = data.get("items", [])
    total = data.get("total_count")
    now = datetime.now(timezone.utc)

    rows = []
    for i, it in enumerate(items, 1):
        owner = (it.get("owner") or {}).get("login", "")
        description = clean_text(it.get("description"))
        rows.append({
            "rank": i,
            "full_name": it.get("full_name", ""),
            "owner": owner,
            "description": 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", ""),
        })

    out = os.path.dirname(os.path.abspath(__file__))
    os.makedirs(out, exist_ok=True)
    st = now.strftime("%Y-%m-%d")
    csv_path = os.path.join(out, "github-python-repos-{}.csv".format(st))
    json_path = os.path.join(out, "github-python-repos-{}.json".format(st))
    md_path = os.path.join(out, "github-python-repos-{}.md".format(st))

    with open(csv_path, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=CSV_COLUMNS, lineterminator="\n")
        w.writeheader()
        for r in rows:
            w.writerow(r)

    with open(json_path, "w", encoding="utf-8") as f:
        json.dump({
            "source": "https://api.github.com/search/repositories",
            "query": "language:python&sort=stars&order=desc&per_page=100",
            "stamp": "{}T{}Z".format(st, now.strftime("%H:%M:%SZ")),
            "total_count": total,
            "rows": rows,
        }, f, ensure_ascii=False, indent=2)

    lines = [
        "# GitHub - top 100 Python repositories by stars",
        "",
        "Source: GitHub REST Search API, GET /search/repositories",
        "Query: `language:python&sort=stars&order=desc&per_page=100` (HTTP 200)",
        "Retrieved: {}T{}Z (UTC)".format(st, now.strftime("%H:%M:%SZ")),
        "Total Python repos in index: {}. Rows below: top {}.".format(total, len(rows)),
        "",
        "Schema (CSV): rank, full_name, owner, description, language, stars, "
        "forks, open_issues, created_at, updated_at, homepage, html_url",
        "",
        "## Table",
        "",
        "| # | Repo | Owner | Description | Language | Stars | Forks | Updated |",
        "|---|------|-------|-------------|----------|-------|-------|---------|",
    ]
    for r in rows:
        lines.append(
            "| {rank} | [{full_name}]({html_url}) | {owner} | {desc} | {lang} | "
            "{stars} | {forks} | {updated_at} |".format(
                rank=r["rank"],
                full_name=r["full_name"],
                html_url=r["html_url"],
                owner=r["owner"],
                desc=clean_text(r["description"], DISPLAY_LIMIT),
                lang=r["language"],
                stars=r["stars"],
                forks=r["forks"],
                updated_at=r["updated_at"],
            )
        )
    lines += [
        "",
        "## Evidence",
        "",
        "Every number above is a field of the live API JSON payload at the snapshot "
        "time. Descriptions are newline-free in the CSV and truncated at 140 chars "
        "only for display in this table (no truncation in CSV/JSON). Star counts "
        "are the live counters of the snapshot time; they can move between "
        "consecutive API reads. Ranks, URLs and counts were copied verbatim; no "
        "value was edited or invented.",
        "",
    ]
    with open(md_path, "w", encoding="utf-8") as f:
        f.write("\n".join(lines))

    print("fetched {} repos (total_count={}, stamp={}Z)".format(len(rows), total, st, now.strftime('%H:%M:%SZ')))
    print("  csv: {}".format(csv_path))
    print("  json: {}".format(json_path))
    print("  md:   {}".format(md_path))
    return 0


if __name__ == "__main__":
    sys.exit(main())