GitHub stargazer vs org membership analysis script (Typer + Rich)

github_star_membership_analysis.py

#!/usr/bin/env -S uv run
# /// script
# requires-python = ">=3.11"
# dependencies = [
#   "requests>=2.31",
#   "rich>=13.7",
#   "typer>=0.12",
# ]
# ///

from __future__ import annotations

import json
import os
from collections import Counter
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Iterable

import requests
import typer
from rich.columns import Columns
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.text import Text

API_BASE = "https://api.github.com"
DEFAULT_REPO = "liatrio-labs/spec-driven-workflow"
DEFAULT_ORG = "liatrio"
DEFAULT_OUT_JSON = "temp/star-membership-analysis.json"
DEFAULT_TOKEN_ENV = "GITHUB_LIATRIO_STARGAZER_TOKEN"

app = typer.Typer(
    add_completion=False,
    rich_markup_mode="rich",
    pretty_exceptions_enable=False,
    help="Analyze GitHub stargazers vs org membership and render a rich terminal report.",
)


@dataclass(frozen=True)
class StarEvent:
    login: str
    account_type: str
    starred_at: str


def sort_logins(logins: Iterable[str]) -> list[str]:
    return sorted(logins, key=str.casefold)


def build_headers(token: str | None) -> dict[str, str]:
    headers = {
        "Accept": "application/vnd.github.star+json",
        "X-GitHub-Api-Version": "2022-11-28",
    }
    if token:
        headers["Authorization"] = f"Bearer {token}"
    return headers


def parse_error(response: requests.Response) -> str:
    status = response.status_code
    try:
        payload = response.json()
        message = payload.get("message", response.text)
    except ValueError:
        message = response.text

    if status == 401:
        return f"401 Unauthorized: {message}"
    if status == 403:
        remaining = response.headers.get("X-RateLimit-Remaining")
        reset = response.headers.get("X-RateLimit-Reset")
        return (
            f"403 Forbidden: {message}. "
            f"RateLimitRemaining={remaining}, RateLimitReset={reset}"
        )
    if status == 404:
        return f"404 Not Found: {message}"
    return f"HTTP {status}: {message}"


def request_json(
    session: requests.Session, url: str, headers: dict[str, str], params: dict[str, int]
) -> list[dict]:
    response = session.get(url, headers=headers, params=params, timeout=30)
    if response.status_code != 200:
        raise RuntimeError(parse_error(response))
    payload = response.json()
    if not isinstance(payload, list):
        raise RuntimeError(f"Unexpected API response for {url}: expected a list.")
    return payload


def fetch_stargazers(
    session: requests.Session, repo: str, headers: dict[str, str], per_page: int
) -> list[StarEvent]:
    page = 1
    events: list[StarEvent] = []
    url = f"{API_BASE}/repos/{repo}/stargazers"

    while True:
        rows = request_json(
            session,
            url,
            headers,
            params={"per_page": per_page, "page": page},
        )
        if not rows:
            break
        for row in rows:
            user = row.get("user") or {}
            login = user.get("login")
            starred_at = row.get("starred_at")
            if not login or not starred_at:
                continue
            events.append(
                StarEvent(
                    login=login,
                    account_type=user.get("type", "Unknown"),
                    starred_at=starred_at,
                )
            )
        page += 1

    return sorted(events, key=lambda event: event.starred_at)


def fetch_org_members(
    session: requests.Session, org: str, headers: dict[str, str], per_page: int
) -> set[str]:
    page = 1
    members: set[str] = set()
    url = f"{API_BASE}/orgs/{org}/members"

    while True:
        rows = request_json(
            session,
            url,
            headers,
            params={"per_page": per_page, "page": page},
        )
        if not rows:
            break
        for row in rows:
            login = row.get("login")
            if login:
                members.add(login.lower())
        page += 1

    return members


def check_org_membership(
    session: requests.Session, org: str, login: str, headers: dict[str, str]
) -> bool:
    url = f"{API_BASE}/orgs/{org}/memberships/{login}"
    response = session.get(url, headers=headers, timeout=30)

    if response.status_code == 404:
        return False
    if response.status_code != 200:
        raise RuntimeError(f"Membership probe failed for {login}: {parse_error(response)}")

    payload = response.json()
    state = payload.get("state")
    return state == "active"


def is_membership_probe_forbidden(error: RuntimeError) -> bool:
    message = str(error).lower()
    return "membership probe failed" in message and "403 forbidden" in message


def is_bot(login: str, account_type: str) -> bool:
    return account_type.lower() == "bot" or login.endswith("[bot]")


def to_dt(stamp: str) -> datetime:
    return datetime.fromisoformat(stamp.replace("Z", "+00:00")).astimezone(UTC)


def compute_monthly_counts(events: list[StarEvent]) -> list[tuple[str, int]]:
    counter: Counter[str] = Counter()
    for event in events:
        counter[to_dt(event.starred_at).strftime("%Y-%m")] += 1
    return sorted(counter.items(), key=lambda item: item[0], reverse=True)


def write_json(path: Path, payload: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")


def print_error(console: Console, message: str) -> None:
    console.print(Panel(message, title="Error", border_style="red"))


def print_warning(console: Console, message: str) -> None:
    console.print(Panel(message, title="Warning", border_style="yellow"))


def render_header(
    console: Console,
    repo: str,
    org: str,
    generated_at: str,
    membership_basis: str,
    probe_attempted: bool,
    probe_blocked: bool,
) -> None:
    mode = "full" if membership_basis == "all_org_members" else "list-only (token-limited)"
    details = (
        f"[bold]Repository:[/bold] {repo}\n"
        f"[bold]Org benchmark:[/bold] {org}\n"
        f"[bold]Generated (UTC):[/bold] {generated_at}\n"
        f"[bold]Membership mode:[/bold] {mode}\n"
        f"[bold]Probe attempted:[/bold] {probe_attempted}\n"
        f"[bold]Probe blocked:[/bold] {probe_blocked}"
    )
    console.print(Panel(details, title="GitHub Stargazer Analysis", border_style="cyan"))


def render_kpis(
    console: Console,
    total_human: int,
    internal: int,
    external: int,
    external_percent: float,
    excluded_bots: int,
    first_star: str | None,
    last_star: str | None,
    stars_last_30d: int,
) -> None:
    table = Table(title="Summary Metrics", show_header=True, header_style="bold magenta")
    table.add_column("Metric", style="bold", no_wrap=True)
    table.add_column("Value", justify="right")
    table.add_row("Total human stargazers", str(total_human))
    table.add_row("Internal stargazers", str(internal))
    table.add_row("External stargazers", str(external))
    table.add_row("External popularity", f"{external_percent:.2f}%")
    table.add_row("Excluded bot stargazers", str(excluded_bots))
    table.add_row("First human star date", first_star or "N/A")
    table.add_row("Latest human star date", last_star or "N/A")
    table.add_row("Human stars in last 30 days", str(stars_last_30d))
    console.print(table)


def render_monthly(console: Console, monthly_counts: list[tuple[str, int]], limit: int) -> None:
    shown = monthly_counts[:limit]
    table = Table(title=f"Monthly Human Stars (latest {len(shown)})", show_header=True)
    table.add_column("Month", style="cyan", no_wrap=True)
    table.add_column("Stars", justify="right", style="green")
    for month, count in shown:
        table.add_row(month, str(count))
    console.print(table)


def render_user_panel(
    console: Console, title: str, users: list[str], limit: int, columns: int = 4
) -> None:
    shown = users[:limit]
    subtitle = f"showing {len(shown)} of {len(users)}"
    if not shown:
        console.print(Panel("[dim]None[/dim]", title=title, subtitle=subtitle))
        return

    items = [Text(user, style="bright_white") for user in shown]
    col_count = max(1, columns)
    chunks = [items[i : i + col_count] for i in range(0, len(items), col_count)]
    rows = []
    for chunk in chunks:
        row = Table.grid(expand=True)
        for _ in range(col_count):
            row.add_column(ratio=1)
        padded = list(chunk) + [Text("")] * (col_count - len(chunk))
        row.add_row(*padded)
        rows.append(row)

    console.print(
        Panel(
            Columns(rows, expand=True),
            title=title,
            subtitle=subtitle,
            border_style="blue",
        )
    )


@app.command()
def main(
    repo: str = typer.Option(DEFAULT_REPO, help="Repository in owner/name format."),
    org: str = typer.Option(DEFAULT_ORG, help="GitHub organization slug."),
    token_env: str = typer.Option(
        DEFAULT_TOKEN_ENV,
        help="Environment variable name containing a GitHub token.",
    ),
    out_json: str = typer.Option(DEFAULT_OUT_JSON, help="Path to write JSON analysis output."),
    sample_size: int = typer.Option(
        20,
        help="Legacy compatibility flag for external sampling (ignored when rich panels are used).",
    ),
    per_page: int = typer.Option(100, min=1, max=100, help="GitHub API pagination size."),
    show_internal_limit: int = typer.Option(
        100, min=0, help="Maximum internal usernames to display in terminal output."
    ),
    show_external_limit: int = typer.Option(
        100, min=0, help="Maximum external usernames to display in terminal output."
    ),
    show_monthly_limit: int = typer.Option(
        12, min=1, help="Number of months to show in monthly trend table."
    ),
    no_color: bool = typer.Option(False, help="Disable colored output."),
    verbose_diagnostics: bool = typer.Option(
        False, help="Show additional token/membership diagnostics."
    ),
) -> None:
    del sample_size
    console = Console(no_color=no_color)
    err_console = Console(stderr=True, no_color=no_color)

    token = os.getenv(token_env)
    if not token:
        print_error(
            err_console,
            f"Missing GitHub token. Set [bold]{token_env}[/bold] with a token that can read org members.",
        )
        raise typer.Exit(code=2)

    headers = build_headers(token)
    try:
        with requests.Session() as session:
            events = fetch_stargazers(session, repo, headers, per_page)
            org_members = fetch_org_members(session, org, headers, per_page)
    except RuntimeError as exc:
        print_error(err_console, f"GitHub API error: {exc}")
        raise typer.Exit(code=1)
    except requests.RequestException as exc:
        print_error(err_console, f"Network error: {exc}")
        raise typer.Exit(code=1)

    latest_event_by_login: dict[str, StarEvent] = {}
    for event in events:
        latest_event_by_login[event.login] = event

    bots: list[str] = []
    human_events: list[StarEvent] = []
    for event in latest_event_by_login.values():
        if is_bot(event.login, event.account_type):
            bots.append(event.login)
            continue
        human_events.append(event)
    human_events = sorted(human_events, key=lambda event: event.starred_at)

    internal_users = sort_logins(
        event.login for event in human_events if event.login.lower() in org_members
    )

    membership_probe_attempted = False
    membership_probe_blocked = False
    if not internal_users and human_events:
        membership_probe_attempted = True
        try:
            with requests.Session() as probe_session:
                internal_users = sort_logins(
                    event.login
                    for event in human_events
                    if check_org_membership(probe_session, org, event.login, headers)
                )
        except RuntimeError as exc:
            if is_membership_probe_forbidden(exc):
                membership_probe_blocked = True
                print_warning(
                    err_console,
                    "Token cannot access per-user org membership checks. "
                    "Using org members list only; internal users may be undercounted.",
                )
            else:
                print_error(err_console, f"GitHub API error: {exc}")
                raise typer.Exit(code=1)

    internal_set = set(internal_users)
    external_users = sort_logins(
        event.login for event in human_events if event.login not in internal_set
    )

    total_human = len(human_events)
    external = len(external_users)
    internal = len(internal_users)
    external_percent = (external / total_human * 100.0) if total_human else 0.0

    first_star = to_dt(human_events[0].starred_at).date().isoformat() if human_events else None
    last_star = to_dt(human_events[-1].starred_at).date().isoformat() if human_events else None
    since = datetime.now(tz=UTC) - timedelta(days=30)
    stars_last_30d = sum(1 for event in human_events if to_dt(event.starred_at) >= since)

    generated_at = datetime.now(tz=UTC).isoformat()
    membership_basis = (
        "org_members_list_only" if membership_probe_blocked else "all_org_members"
    )
    output = {
        "repo": repo,
        "org": org,
        "generated_at": generated_at,
        "assumptions": {
            "membership_basis": membership_basis,
            "bots_excluded": True,
            "internal_org": org,
            "membership_probe_attempted": membership_probe_attempted,
            "membership_probe_blocked": membership_probe_blocked,
        },
        "counts": {
            "total_human_stargazers": total_human,
            "internal": internal,
            "external": external,
            "external_percent": round(external_percent, 4),
            "excluded_bots": len(bots),
        },
        "excluded": {"bots": sorted(bots)},
        "usernames": {"internal": internal_users, "external": external_users},
    }

    out_path = Path(out_json)
    write_json(out_path, output)

    render_header(
        console=console,
        repo=repo,
        org=org,
        generated_at=generated_at,
        membership_basis=membership_basis,
        probe_attempted=membership_probe_attempted,
        probe_blocked=membership_probe_blocked,
    )
    render_kpis(
        console=console,
        total_human=total_human,
        internal=internal,
        external=external,
        external_percent=external_percent,
        excluded_bots=len(bots),
        first_star=first_star,
        last_star=last_star,
        stars_last_30d=stars_last_30d,
    )
    render_monthly(console, compute_monthly_counts(human_events), show_monthly_limit)
    render_user_panel(
        console,
        title="Internal Stargazers",
        users=internal_users,
        limit=show_internal_limit,
    )
    render_user_panel(
        console,
        title="External Stargazers",
        users=external_users,
        limit=show_external_limit,
    )

    if verbose_diagnostics:
        diagnostics = (
            f"token_env={token_env}\n"
            f"org_members_loaded={len(org_members)}\n"
            f"membership_probe_attempted={membership_probe_attempted}\n"
            f"membership_probe_blocked={membership_probe_blocked}"
        )
        console.print(Panel(diagnostics, title="Diagnostics", border_style="magenta"))

    console.print(Panel(f"Analysis written to: [bold]{out_path}[/bold]", border_style="green"))
if __name__ == "__main__":
    app()