Vibeslopped github PR stats collector

ghstats.py

#!/usr/bin/env python3
"""Collect, store, graph, and print GitHub pull-request search counts."""

from __future__ import annotations

import argparse
import calendar
import datetime as dt
import json
import os
import sqlite3
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path


FIRST_YEAR = 2020
REPORT_YEAR = 2026
REPORT_YEAR_END = "2026-09-01"
SEARCH_URL = "https://api.github.com/search/issues"
DATABASE_PATH = Path(__file__).with_name("ghstats.sqlite3")
ALL_PULL_REQUESTS_TERM = "__all"


class GitHubSearchClient:
    """Search client that paces requests conservatively below search API limits."""

    def __init__(self) -> None:
        self.token = os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN")
        # GitHub permits 10 unauthenticated search requests per minute. Keep below it.
        self.minimum_interval = 10.0 if not self.token else 2.1
        self.last_request_at: float | None = None

    def count_pull_requests(self, term: str, start: str, end: str) -> int:
        if self.last_request_at is not None:
            wait = self.minimum_interval - (time.monotonic() - self.last_request_at)
            if wait > 0:
                time.sleep(wait)

        query = (
            f"{term} in:title is:pr created:{start}..{end}"
            if term != ALL_PULL_REQUESTS_TERM
            else f"is:pr created:{start}..{end}"
        )
        request = urllib.request.Request(
            f"{SEARCH_URL}?{urllib.parse.urlencode({'q': query, 'per_page': 1})}",
            headers={
                "Accept": "application/vnd.github+json",
                "User-Agent": "ghstats",
                **({"Authorization": f"Bearer {self.token}"} if self.token else {}),
            },
        )
        self.last_request_at = time.monotonic()

        try:
            with urllib.request.urlopen(request, timeout=30) as response:
                payload = json.load(response)
        except urllib.error.HTTPError as error:
            if error.code in (403, 429):
                reset = error.headers.get("X-RateLimit-Reset")
                if reset and reset.isdigit():
                    wait = max(1, int(reset) - int(time.time()) + 1)
                    raise RuntimeError(
                        f"GitHub rate limit reached. Try again after {wait} seconds."
                    ) from error
                raise RuntimeError(
                    "GitHub rejected the search request, likely due to rate limiting. "
                    "Set GH_TOKEN to increase the limit and try again."
                ) from error
            raise RuntimeError(f"GitHub search failed: HTTP {error.code}") from error
        except urllib.error.URLError as error:
            raise RuntimeError(f"Could not reach GitHub: {error.reason}") from error

        total = payload.get("total_count")
        if not isinstance(total, int):
            raise RuntimeError("GitHub returned an unexpected search response.")
        return total


def connect_database() -> sqlite3.Connection:
    connection = sqlite3.connect(DATABASE_PATH)
    connection.execute(
        """
        CREATE TABLE IF NOT EXISTS searches (
            term TEXT PRIMARY KEY,
            updated_at TEXT NOT NULL
        )
        """
    )
    connection.execute(
        """
        CREATE TABLE IF NOT EXISTS yearly_counts (
            term TEXT NOT NULL REFERENCES searches(term) ON DELETE CASCADE,
            year INTEGER NOT NULL,
            pull_requests INTEGER NOT NULL CHECK (pull_requests >= 0),
            PRIMARY KEY (term, year)
        )
        """
    )
    connection.execute(
        """
        CREATE TABLE IF NOT EXISTS monthly_counts (
            term TEXT NOT NULL REFERENCES searches(term) ON DELETE CASCADE,
            year INTEGER NOT NULL,
            month INTEGER NOT NULL CHECK (month BETWEEN 1 AND 12),
            pull_requests INTEGER NOT NULL CHECK (pull_requests >= 0),
            PRIMARY KEY (term, year, month)
        )
        """
    )
    migrate_aggregate_counts(connection)
    return connection


def migrate_aggregate_counts(connection: sqlite3.Connection) -> None:
    tables = {
        row[0]
        for row in connection.execute(
            "SELECT name FROM sqlite_master WHERE type = 'table'"
        )
    }
    if not {"total_yearly_counts", "total_monthly_counts"} & tables:
        return

    updated_at = dt.datetime.now(dt.timezone.utc).isoformat()
    connection.execute(
        """
        INSERT INTO searches (term, updated_at) VALUES (?, ?)
        ON CONFLICT(term) DO UPDATE SET updated_at = excluded.updated_at
        """,
        (ALL_PULL_REQUESTS_TERM, updated_at),
    )
    if "total_yearly_counts" in tables:
        connection.execute(
            """
            INSERT OR REPLACE INTO yearly_counts (term, year, pull_requests)
            SELECT ?, year, pull_requests FROM total_yearly_counts
            """,
            (ALL_PULL_REQUESTS_TERM,),
        )
        connection.execute("DROP TABLE total_yearly_counts")
    if "total_monthly_counts" in tables:
        connection.execute(
            """
            INSERT OR REPLACE INTO monthly_counts (term, year, month, pull_requests)
            SELECT ?, year, month, pull_requests FROM total_monthly_counts
            """,
            (ALL_PULL_REQUESTS_TERM,),
        )
        connection.execute("DROP TABLE total_monthly_counts")


def date_range_for(year: int) -> tuple[str, str]:
    start = f"{year}-01-01"
    end = REPORT_YEAR_END if year == REPORT_YEAR else f"{year}-12-31"
    return start, end


def update_counts(term: str, client: GitHubSearchClient | None = None) -> None:
    client = client or GitHubSearchClient()
    counts: list[tuple[int, int]] = []
    for year in range(FIRST_YEAR, REPORT_YEAR + 1):
        start, end = date_range_for(year)
        count = client.count_pull_requests(term, start, end)
        counts.append((year, count))
        print(f"{year}: {count}", file=sys.stderr)

    with connect_database() as connection:
        connection.execute("DELETE FROM yearly_counts WHERE term = ?", (term,))
        connection.execute(
            """
            INSERT INTO searches (term, updated_at) VALUES (?, ?)
            ON CONFLICT(term) DO UPDATE SET updated_at = excluded.updated_at
            """,
            (term, dt.datetime.now(dt.timezone.utc).isoformat()),
        )
        connection.executemany(
            "INSERT INTO yearly_counts (term, year, pull_requests) VALUES (?, ?, ?)",
            [(term, year, count) for year, count in counts],
        )
    print(f"Updated {len(counts)} yearly counts for {term!r} in {DATABASE_PATH}.")


def month_ranges_for(year: int) -> list[tuple[int, str, str]]:
    if not FIRST_YEAR <= year <= REPORT_YEAR:
        raise RuntimeError(f"Year must be between {FIRST_YEAR} and {REPORT_YEAR}.")
    final_month = 8 if year == REPORT_YEAR else 12
    return [
        (
            month,
            f"{year}-{month:02d}-01",
            f"{year}-{month:02d}-{calendar.monthrange(year, month)[1]:02d}",
        )
        for month in range(1, final_month + 1)
    ]


def update_monthly_counts(
    term: str, year: int, client: GitHubSearchClient | None = None
) -> list[tuple[int, int]]:
    client = client or GitHubSearchClient()
    counts: list[tuple[int, int]] = []
    for month, start, end in month_ranges_for(year):
        count = client.count_pull_requests(term, start, end)
        counts.append((month, count))
        print(f"{year}-{month:02d}: {count}", file=sys.stderr)

    with connect_database() as connection:
        connection.execute(
            """
            INSERT INTO searches (term, updated_at) VALUES (?, ?)
            ON CONFLICT(term) DO UPDATE SET updated_at = excluded.updated_at
            """,
            (term, dt.datetime.now(dt.timezone.utc).isoformat()),
        )
        connection.execute(
            "DELETE FROM monthly_counts WHERE term = ? AND year = ?", (term, year)
        )
        connection.executemany(
            """
            INSERT INTO monthly_counts (term, year, month, pull_requests)
            VALUES (?, ?, ?, ?)
            """,
            [(term, year, month, count) for month, count in counts],
        )
    return counts


def local_counts(term: str) -> list[tuple[int, int]]:
    with connect_database() as connection:
        rows = connection.execute(
            "SELECT year, pull_requests FROM yearly_counts WHERE term = ? ORDER BY year",
            (term,),
        ).fetchall()
    if not rows:
        raise RuntimeError(f"No local data for {term!r}. Run `ghstats.py prs {term!r}` first.")
    return [(int(year), int(count)) for year, count in rows]


def local_monthly_counts(term: str, year: int) -> list[tuple[int, int]]:
    with connect_database() as connection:
        rows = connection.execute(
            """
            SELECT month, pull_requests
            FROM monthly_counts
            WHERE term = ? AND year = ?
            ORDER BY month
            """,
            (term, year),
        ).fetchall()
    if not rows:
        raise RuntimeError(
            f"No local monthly data for {term!r} in {year}. "
            f"Run `ghstats.py pr-month {year} {term!r}` first."
        )
    return [(int(month), int(count)) for month, count in rows]


def list_searches() -> list[tuple[str, list[int]]]:
    with connect_database() as connection:
        rows = connection.execute(
            """
            SELECT searches.term, yearly_counts.year
            FROM searches
            LEFT JOIN yearly_counts ON yearly_counts.term = searches.term
            ORDER BY searches.term COLLATE NOCASE, yearly_counts.year
            """
        ).fetchall()

    searches: dict[str, list[int]] = {}
    for term, year in rows:
        searches.setdefault(str(term), [])
        if year is not None:
            searches[str(term)].append(int(year))
    return list(searches.items())


def print_searches(searches: list[tuple[str, list[int]]]) -> None:
    if not searches:
        print("No terms are stored in the database.")
        return

    print("Term  Years")
    print("----  -----")
    for term, years in searches:
        print(f"{term}  {', '.join(map(str, years)) or '(no counts)'}")


def counts_starting_at(
    counts: list[tuple[int, int]], start_year: int | None
) -> list[tuple[int, int]]:
    if start_year is None:
        return counts
    filtered_counts = [(year, count) for year, count in counts if year >= start_year]
    if not filtered_counts:
        raise RuntimeError(f"No local data from {start_year} onward.")
    return filtered_counts


def output_path_for(term: str, graph_type: str, year: int | None = None) -> Path:
    safe_name = "".join(char if char.isalnum() else "-" for char in term).strip("-")
    suffix = f"-{year}" if year is not None else ""
    return Path.cwd() / f"ghstats-{safe_name[:60] or 'search'}-{graph_type}{suffix}.png"


def write_bar_graph(
    term: str,
    labels: list[str],
    counts: list[int],
    graph_type: str,
    *,
    year: int | None = None,
) -> Path:
    import matplotlib

    matplotlib.use("Agg")
    import matplotlib.pyplot as pyplot
    from matplotlib.ticker import StrMethodFormatter

    path = output_path_for(term, graph_type, year)
    figure, axis = pyplot.subplots(figsize=(11, 6), layout="constrained")
    axis.bar(labels, counts, color="#0969da", width=0.7)
    title_suffix = f" by month in {year}" if year is not None else " by year"
    axis.set_title(f"GitHub pull requests matching: {term}{title_suffix}", weight="bold")
    axis.set_xlabel("Month" if year is not None else "Year")
    axis.set_ylabel("Pull requests")
    axis.yaxis.set_major_formatter(StrMethodFormatter("{x:,.0f}"))
    axis.grid(axis="y", alpha=0.3)
    if year == REPORT_YEAR:
        axis.text(
            0.5,
            -0.18,
            "2026 includes pull requests created through August 31.",
            transform=axis.transAxes,
            ha="center",
            color="#57606a",
        )
    figure.savefig(path, dpi=180)
    pyplot.close(figure)
    return path


def print_counts(term: str, counts: list[tuple[int, int]]) -> None:
    print(f"GitHub pull requests matching: {term}")
    print("Year  Pull requests")
    print("----  -------------")
    for year, count in counts:
        print(f"{year}  {count:,}")

    before_2026 = sum(count for year, count in counts if year < REPORT_YEAR)
    count_2026 = sum(count for year, count in counts if year == REPORT_YEAR)
    ratio = "n/a (no PRs before 2026)" if before_2026 == 0 else f"{count_2026 / before_2026:.6f}"
    print()
    print(f"PRs before 2026: {before_2026:,}")
    print(f"PRs in 2026 (through September 1): {count_2026:,}")
    print(f"2026 / before-2026 ratio: {ratio}")


def print_monthly_counts(term: str, year: int, counts: list[tuple[int, int]]) -> None:
    print(f"GitHub pull requests matching: {term} ({year})")
    print("Month      Pull requests")
    print("---------  -------------")
    for month, count in counts:
        print(f"{calendar.month_name[month]:<9}  {count:,}")


def parse_arguments() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Store and analyze yearly GitHub pull-request search counts."
    )
    commands = parser.add_subparsers(dest="command", required=True)
    prs = commands.add_parser("prs", help="Fetch and refresh yearly PR counts.")
    prs.add_argument("term", help="GitHub search term or expression")
    pr_month = commands.add_parser(
        "pr-month", help="Fetch and refresh one year's monthly PR counts."
    )
    pr_month.add_argument("year", type=int, help=f"Year ({FIRST_YEAR}-{REPORT_YEAR})")
    pr_month.add_argument("term", help="GitHub search term or expression")
    graph_year = commands.add_parser(
        "graph-year", help="Write a PNG yearly bar chart from local data."
    )
    graph_year.add_argument("term", help="GitHub search term or expression")
    graph_year.add_argument(
        "--start", type=int, metavar="YEAR", help="Include data from this year onward."
    )
    graph_month = commands.add_parser(
        "graph-month", help="Write a PNG monthly bar chart from local data."
    )
    graph_month.add_argument("year", type=int, help="Year of saved monthly data")
    graph_month.add_argument("term", help="GitHub search term or expression")
    for name, help_text in (
        ("print", "Print locally stored yearly data and summary totals."),
        ("show", "Alias for print."),
    ):
        command = commands.add_parser(name, help=help_text)
        command.add_argument("term", help="GitHub search term or expression")
    commands.add_parser("list", help="List stored search terms and their available years.")
    return parser.parse_args()


def main() -> int:
    args = parse_arguments()
    try:
        if args.command == "prs":
            update_counts(args.term)
        elif args.command == "pr-month":
            print_monthly_counts(
                args.term, args.year, update_monthly_counts(args.term, args.year)
            )
        elif args.command == "list":
            print_searches(list_searches())
        elif args.command == "graph-year":
            counts = counts_starting_at(local_counts(args.term), args.start)
            path = write_bar_graph(
                args.term,
                [str(year) for year, _ in counts],
                [count for _, count in counts],
                "year",
            )
            print(f"Wrote yearly bar chart to {path}.")
        elif args.command == "graph-month":
            counts = local_monthly_counts(args.term, args.year)
            path = write_bar_graph(
                args.term,
                [calendar.month_abbr[month] for month, _ in counts],
                [count for _, count in counts],
                "month",
                year=args.year,
            )
            print(f"Wrote monthly bar chart to {path}.")
        else:
            counts = local_counts(args.term)
            print_counts(args.term, counts)
    except RuntimeError as error:
        print(f"error: {error}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

requirements.txt

matplotlib==3.11.1

These "LLM-isms" are insufferable. I can only imagine these spikes will now get back into the training pipelines and amplify their signal even more, though I cannot fathom any one person suffering much more than what Claude 5 series already does...

and of course "bearing" (Claude really likes "load-bearing" things):