collect.py

"""Daily snapshot of stars and Docker pull counts for self-hosted apps.

WHY THIS EXISTS. The two public signals people use to pick self-hosted software
disagree violently. Immich is 1st by GitHub stars and 12th by Docker pulls;
Heimdall is 17th by stars and 3rd by pulls, with one twelfth the stars and
eighteen times the pulls. At least one of those signals is badly wrong about
something, and nobody currently knows which apps people actually run.

Both signals are broken in known ways:

  stars  -- cumulative, never decay, measure interest at first sight rather than
            use. A project starred in 2019 and abandoned scores the same as one
            thriving today.
  pulls  -- cumulative, so they reward age; and inflated by auto-updaters, CI and
            mirrors. One Watchtower user generates ~365 pulls a year, which is
            most of why linuxserver.io images dominate the raw ranking.

RATE OF CHANGE SURVIVES BOTH PROBLEMS. Age bias vanishes when you difference,
and the auto-updater inflation is a roughly constant multiplier per project, so a
rising or falling daily delta still means adoption is rising or falling. Docker
Hub publishes no time series, so the only way to have one is to start recording.

Append-only. The dataset compounds from the first run, which is the entire reason
to start today rather than after more deliberation.

  python collect.py                 # appends one row per app to data/snapshots.csv
"""

from __future__ import annotations

import argparse
import concurrent.futures as futures
import csv
import json
import os
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

HERE = Path(__file__).resolve().parent
FIELDS = ["date", "slug", "stars", "pulls", "image", "archived"]


def hub_pulls(image: str) -> int | None:
    try:
        with urllib.request.urlopen(
                f"https://hub.docker.com/v2/repositories/{image}", timeout=20) as response:
            return json.load(response).get("pull_count")
    except Exception:
        return None


def github_stars(repo: str, token: str | None) -> int | None:
    request = urllib.request.Request(
        f"https://api.github.com/repos/{repo}",
        headers={"Accept": "application/vnd.github+json",
                 **({"Authorization": f"Bearer {token}"} if token else {})},
    )
    try:
        with urllib.request.urlopen(request, timeout=20) as response:
            return json.load(response).get("stargazers_count")
    except Exception:
        return None


def snapshot(app: dict, token: str | None, today: str) -> dict:
    return {
        "date": today,
        "slug": app["slug"],
        # Recorded even when null: a signal that stopped resolving is itself
        # data, and silently dropping the row would leave a gap that looks like
        # the app was never tracked.
        "stars": github_stars(app["github"], token) if app["github"] else None,
        "pulls": hub_pulls(app["image"]) if app["image"] else None,
        "image": app["image"] or "",
        "archived": int(bool(app["archived"])),
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--apps", default="apps.json")
    parser.add_argument("--out", default="data/snapshots.csv")
    parser.add_argument("--limit", type=int, default=0, help="for smoke tests")
    arguments = parser.parse_args()

    apps = json.loads((HERE / arguments.apps).read_text(encoding="utf-8"))
    apps = [a for a in apps if a["image"] or a["github"]]
    if arguments.limit:
        apps = apps[: arguments.limit]

    token = os.environ.get("GITHUB_TOKEN")
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")

    out_path = HERE / arguments.out
    out_path.parent.mkdir(parents=True, exist_ok=True)
    existing = out_path.exists()

    # Re-running on the same day must not double-count, or a delta computed from
    # the file would read as zero growth on the second run.
    if existing:
        with out_path.open(encoding="utf-8", newline="") as handle:
            if any(row["date"] == today for row in csv.DictReader(handle)):
                print(f"{today} already recorded in {out_path}; nothing to do.")
                return 0

    rows = []
    with futures.ThreadPoolExecutor(max_workers=12) as pool:
        for index, row in enumerate(
                pool.map(lambda a: snapshot(a, token, today), apps), 1):
            rows.append(row)
            if index % 100 == 0:
                print(f"  {index}/{len(apps)}", flush=True)

    with out_path.open("a", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=FIELDS)
        if not existing:
            writer.writeheader()
        writer.writerows(rows)

    with_pulls = sum(1 for r in rows if r["pulls"])
    with_stars = sum(1 for r in rows if r["stars"])
    print(f"\n{today}: appended {len(rows)} rows to {out_path}")
    print(f"  {with_stars} with stars, {with_pulls} with pulls")
    if not token:
        print("  note: no GITHUB_TOKEN, so star lookups are rate-limited to 60/hour")
    return 0


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