Count commits per cloud provider in kubernetes/autoscaler/cluster-autoscaler/cloudprovider over a time window

count_provider_commits.py

#!/usr/bin/env python3
"""
Count commits per cloud provider in kubernetes/autoscaler's
cluster-autoscaler/cloudprovider/ tree over a given time window.

How it works
------------
GitHub's REST Commits API supports two filters we exploit:
  - `path`  : restrict to commits that touched files under this path
  - `since` : ISO-8601 timestamp lower bound

To get a *count* without paginating through every commit, request
`per_page=1` and read the `Link` header. GitHub returns:
    <...&page=N>; rel="last"
where N is the total number of commits matching the query.

If the response body is empty, the count is 0.
If there's no `rel="last"` link but the body is non-empty, there's exactly 1.

Auth: works unauthenticated for small runs (60 req/hr/IP). For repeated
use, set GITHUB_TOKEN to lift the limit to 5000 req/hr.

Usage:
    python3 count_provider_commits.py
    GITHUB_TOKEN=ghp_xxx python3 count_provider_commits.py
"""

import json
import os
import re
import sys
import time
import urllib.parse
import urllib.request

REPO   = "kubernetes/autoscaler"
BRANCH = "master"
ROOT   = "cluster-autoscaler/cloudprovider"
SINCE  = "2025-11-22T00:00:00Z"   # adjust as needed
BASE   = f"https://api.github.com/repos/{REPO}/commits"

HEADERS = {
    "Accept": "application/vnd.github+json",
    "User-Agent": "autoscaler-provider-stats",
}
token = os.environ.get("GITHUB_TOKEN")
if token:
    HEADERS["Authorization"] = f"Bearer {token}"


def list_provider_dirs():
    """List immediate subdirectories under cluster-autoscaler/cloudprovider."""
    url = (f"https://api.github.com/repos/{REPO}/contents/{ROOT}"
           f"?ref={BRANCH}")
    req = urllib.request.Request(url, headers=HEADERS)
    with urllib.request.urlopen(req) as r:
        entries = json.load(r)
    return sorted(e["name"] for e in entries if e["type"] == "dir")


def count_commits(path):
    """Return number of commits to `path` on BRANCH since SINCE."""
    qs = urllib.parse.urlencode({
        "path": path,
        "since": SINCE,
        "sha": BRANCH,
        "per_page": 1,
    })
    req = urllib.request.Request(f"{BASE}?{qs}", headers=HEADERS)
    with urllib.request.urlopen(req) as r:
        link = r.headers.get("Link", "")
        body = json.load(r)

    if not body:
        return 0
    m = re.search(r'[?&]page=(\d+)[^>]*>;\s*rel="last"', link)
    return int(m.group(1)) if m else 1


def main():
    # Exclude non-provider helper dirs.
    SKIP = {"builder", "mocks", "test"}

    providers = [p for p in list_provider_dirs() if p not in SKIP]
    print(f"Found {len(providers)} provider directories.\n"
          f"Counting commits since {SINCE} on {BRANCH}...\n", file=sys.stderr)

    results = []
    for name in providers:
        path = f"{ROOT}/{name}"
        try:
            n = count_commits(path)
        except Exception as e:
            print(f"ERR  {name}: {e}", file=sys.stderr)
            continue
        results.append((name, n))
        print(f"{n:4d}  {name}", flush=True)
        time.sleep(0.2)  # be polite to the API

    results.sort(key=lambda x: (-x[1], x[0]))
    print("\n=== Ranking ===")
    for name, n in results:
        print(f"{n:4d}  {name}")


if __name__ == "__main__":
    main()