Report the monthly rate of issue and pull request creation for a GitHub Repository

github-stats.py

#!/usr/bin/env python3
"""
Report the monthly rate of issue and pull request creation for a GitHub
repository over the last few years.

This uses the GitHub search API (via the `gh` CLI, so it picks up your
existing `gh auth login` credentials) to count how many issues and pull
requests were *created* in each calendar month. It prints a digestible
table with per-year subtotals and a simple text bar chart, and can
optionally emit CSV.

Examples:

    bin/github-stats.py
    bin/github-stats.py --repo rclone/rclone --years 5
    bin/github-stats.py --csv > stats.csv

Requires the `gh` CLI (https://cli.github.com/) to be installed and
authenticated (`gh auth login`).
"""

import argparse
import csv
import datetime
import json
import subprocess
import sys
import time

# GitHub limits authenticated search to 30 requests/minute, so pace
# requests to stay comfortably under that.
SEARCH_DELAY = 2.1


def gh_count(repo, kind, month):
    """
    Return the number of items of kind ("issue" or "pr") created in the
    given month (a "YYYY-MM" string) in repo, using the GitHub search API.

    Retries with a back off if the secondary rate limit is hit.
    """
    query = f"repo:{repo} type:{kind} created:{month}"
    args = [
        "gh", "api", "-X", "GET", "search/issues",
        "-f", f"q={query}",
        "-F", "per_page=1",
        "--jq", ".total_count",
    ]
    for attempt in range(5):
        result = subprocess.run(args, stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE, text=True)
        if result.returncode == 0:
            return int(result.stdout.strip())
        stderr = result.stderr.strip()
        # Rate limited - back off and retry
        if "rate limit" in stderr.lower() or "secondary" in stderr.lower():
            wait = 30 * (attempt + 1)
            print(f"Rate limited, waiting {wait}s...", file=sys.stderr)
            time.sleep(wait)
            continue
        raise RuntimeError(f"gh api failed for {query}: {stderr}")
    raise RuntimeError(f"gh api repeatedly rate limited for {query}")


def months_back(years):
    """
    Yield "YYYY-MM" strings for each whole month in the last `years` years
    up to and including the current month, oldest first.
    """
    today = datetime.date.today()
    total = years * 12
    # Walk back `total` months from the first of this month
    year, month = today.year, today.month
    result = []
    for _ in range(total):
        result.append(f"{year:04d}-{month:02d}")
        month -= 1
        if month == 0:
            month = 12
            year -= 1
    return list(reversed(result))


def bar(value, scale):
    """Return a text bar of length proportional to value."""
    if scale <= 0:
        return ""
    return "#" * round(value / scale * 40)


def main():
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--repo", default="rclone/rclone",
                        help="GitHub repository (default: rclone/rclone)")
    parser.add_argument("--years", type=int, default=5,
                        help="Number of years to report (default: 5)")
    parser.add_argument("--csv", action="store_true",
                        help="Output CSV instead of a formatted report")
    args = parser.parse_args()

    months = months_back(args.years)

    rows = []
    for month in months:
        issues = gh_count(args.repo, "issue", month)
        time.sleep(SEARCH_DELAY)
        prs = gh_count(args.repo, "pr", month)
        time.sleep(SEARCH_DELAY)
        rows.append((month, issues, prs))
        if not args.csv:
            print(f"  fetched {month}: {issues} issues, {prs} PRs",
                  file=sys.stderr)

    if args.csv:
        writer = csv.writer(sys.stdout)
        writer.writerow(["month", "issues", "prs", "total"])
        for month, issues, prs in rows:
            writer.writerow([month, issues, prs, issues + prs])
        return

    print()
    print(f"Monthly issue and pull request creation for {args.repo}")
    print(f"Last {args.years} years, as of {datetime.date.today()}")
    print()

    peak = max((i + p for _, i, p in rows), default=1)
    scale = peak

    header = f"{'Month':<8} {'Issues':>7} {'PRs':>6} {'Total':>6}  Total (bar)"
    print(header)
    print("-" * len(header))

    year_issues = year_prs = 0
    current_year = rows[0][0][:4] if rows else None
    grand_issues = grand_prs = 0

    for month, issues, prs in rows:
        year = month[:4]
        if year != current_year:
            print(f"{current_year + ' total':<8} "
                  f"{year_issues:>7} {year_prs:>6} "
                  f"{year_issues + year_prs:>6}")
            print()
            year_issues = year_prs = 0
            current_year = year
        total = issues + prs
        print(f"{month:<8} {issues:>7} {prs:>6} {total:>6}  {bar(total, scale)}")
        year_issues += issues
        year_prs += prs
        grand_issues += issues
        grand_prs += prs

    if current_year is not None:
        print(f"{current_year + ' total':<8} "
              f"{year_issues:>7} {year_prs:>6} "
              f"{year_issues + year_prs:>6}")

    print()
    n = len(rows)
    print(f"Totals over {n} months: "
          f"{grand_issues} issues, {grand_prs} PRs, "
          f"{grand_issues + grand_prs} combined")
    print(f"Monthly average: "
          f"{grand_issues / n:.1f} issues, {grand_prs / n:.1f} PRs, "
          f"{(grand_issues + grand_prs) / n:.1f} combined")


if __name__ == "__main__":
    main()