Blame for refresh.py

# /// script
# requires-python = ">=3.11"
# dependencies = ["huggingface_hub>=0.30"]
# ///
"""Refresh an Issue Credit Space.

Reads config.toml and page.template.html from the Space, fetches issues from
GitHub, then commits index.html (data inlined) and data.json back to the Space.

Needs two tokens, both read from the environment and never written anywhere:
  GITHUB_TOKEN  any GitHub token; read-only access to public repos is enough
  HF_TOKEN      a Hugging Face token that can write to the Space
                (locally, your `hf auth login` token works instead)

Run on your machine:
  GITHUB_TOKEN=$(gh auth token) uv run refresh.py --space you/issue-credit
Run once as a Job:
  hf jobs uv run --secrets HF_TOKEN --secrets GITHUB_TOKEN refresh.py --space you/issue-credit
Run weekly:
  hf jobs scheduled uv run --secrets HF_TOKEN --secrets GITHUB_TOKEN @weekly refresh.py --space you/issue-credit
Try it without uploading:
  GITHUB_TOKEN=... uv run refresh.py --space you/issue-credit --out ./preview
"""

import argparse
import json
import os
import sys
import time
import tomllib
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from datetime import UTC, date, datetime, timedelta
from pathlib import Path

from huggingface_hub import CommitOperationAdd, HfApi, hf_hub_download

GRAPHQL_URL = "https://api.github.com/graphql"
TEAM_ASSOCIATIONS = {"COLLABORATOR": 1, "MEMBER": 2, "OWNER": 3}
OUTCOME_CODES = {"converted": 0, "self_fixed": 1, "completed": 2, "not_planned": 3, "open": 4}

ISSUES_QUERY = """
query($owner:String!, $name:String!, $cursor:String) {
  repository(owner:$owner, name:$name) {
    issues(first:50, after:$cursor, orderBy:{field:CREATED_AT, direction:DESC}) {
      pageInfo { hasNextPage endCursor }
      nodes {
        number title createdAt closedAt state stateReason
        author { __typename login }
        authorAssociation
        closedByPullRequestsReferences(first:5, includeClosedPrs:false) {
          nodes { number merged author { __typename login } authorAssociation }
        }
        timelineItems(itemTypes:[CLOSED_EVENT], last:1) {
          nodes {
            ... on ClosedEvent {
              closer {
                __typename
                ... on PullRequest { number merged author { __typename login } authorAssociation }
                ... on Commit { oid author { user { login } } }
              }
            }
          }
        }
      }
    }
  }
}
"""


# ---------- GitHub ----------

def run_graphql(query: str, variables: dict, token: str) -> dict:
    """POST one GraphQL query. Retries a few times on network or 5xx errors."""
    body = json.dumps({"query": query, "variables": variables}).encode()
    headers = {
        "Authorization": f"bearer {token}",
        "Content-Type": "application/json",
        "User-Agent": "issue-credit-space",
    }
    for attempt in range(4):
        request = urllib.request.Request(GRAPHQL_URL, data=body, headers=headers)
        try:
            with urllib.request.urlopen(request, timeout=90) as response:
                payload = json.loads(response.read())
        except urllib.error.HTTPError as error:
            if error.code in (401, 403):
                sys.exit(f"GitHub refused the request (HTTP {error.code}). Check that GITHUB_TOKEN is valid.")
            if attempt == 3:
                raise
            print(f"  GitHub HTTP {error.code}, retrying", file=sys.stderr)
            time.sleep(5 * (attempt + 1))
            continue
        except (urllib.error.URLError, TimeoutError) as error:
            if attempt == 3:
                raise
            print(f"  network error ({error.__class__.__name__}), retrying", file=sys.stderr)
            time.sleep(5 * (attempt + 1))
            continue
        if payload.get("errors") and not payload.get("data"):
            raise RuntimeError(f"GitHub GraphQL error: {payload['errors']}")
        return payload["data"]
    raise RuntimeError("unreachable")


def fetch_repo(repo: str, cutoff: str, token: str) -> list[dict]:
    """All issues in one repo created on or after `cutoff` (ISO date), newest first."""
    owner, name = repo.split("/")
    issues: list[dict] = []
    cursor = None
    while True:
        data = run_graphql(ISSUES_QUERY, {"owner": owner, "name": name, "cursor": cursor}, token)
        repository = data.get("repository")
        if repository is None:
            print(f"  skipping {repo}: not found or not visible", file=sys.stderr)
            return []
        page = repository["issues"]
        reached_cutoff = False
        for node in page["nodes"]:
            if node["createdAt"] < cutoff:
                reached_cutoff = True
                break
            node["repo"] = repo
            issues.append(node)
        if reached_cutoff or not page["pageInfo"]["hasNextPage"]:
            break
        cursor = page["pageInfo"]["endCursor"]
    print(f"  {repo}: {len(issues)} issues", file=sys.stderr)
    return issues


# ---------- classification ----------

def login_of(actor: dict | None) -> str | None:
    return (actor or {}).get("login")


def is_bot(actor: dict | None, bot_accounts: set[str]) -> bool:
    if not actor:
        return False
    login = actor.get("login") or ""
    return actor.get("__typename") == "Bot" or login in bot_accounts or login.endswith("[bot]")


def find_fix(issue: dict) -> tuple[dict | None, str]:
    """Return (fixer actor, reference) for the merged PR or commit that closed the issue."""
    for event in issue["timelineItems"]["nodes"]:
        closer = event.get("closer") or {}
        if closer.get("__typename") == "PullRequest" and closer.get("merged"):
            return closer.get("author"), f"#{closer['number']}"
        if closer.get("__typename") == "Commit":
            return (closer.get("author") or {}).get("user"), closer["oid"][:7]
    for pr in issue["closedByPullRequestsReferences"]["nodes"]:
        if pr.get("merged"):
            return pr.get("author"), f"#{pr['number']}"
    return None, ""


def outcome_for(issue: dict, author: str, fixer: str | None) -> str:
    if issue["state"] == "OPEN":
        return "open"
    if fixer and fixer != author:
        return "converted"
    if fixer and fixer == author:
        return "self_fixed"
    if issue["stateReason"] == "COMPLETED":
        return "completed"
    return "not_planned"


def collect_team(raw: list[dict]) -> dict[str, str]:
    """People GitHub marks as Member/Owner/Collaborator anywhere, with their strongest label."""
    team: dict[str, str] = {}

    def note(actor: dict | None, association: str | None) -> None:
        login = login_of(actor)
        if not login or association not in TEAM_ASSOCIATIONS:
            return
        current = team.get(login)
        if current is None or TEAM_ASSOCIATIONS[association] > TEAM_ASSOCIATIONS[current]:
            team[login] = association

    for issue in raw:
        note(issue.get("author"), issue["authorAssociation"])
        for event in issue["timelineItems"]["nodes"]:
            closer = event.get("closer") or {}
            note(closer.get("author"), closer.get("authorAssociation"))
        for pr in issue["closedByPullRequestsReferences"]["nodes"]:
            note(pr.get("author"), pr.get("authorAssociation"))
    return {login: association.title() for login, association in sorted(team.items())}


def build_payload(raw: list[dict], config: dict, repos: list[str], space: str, today, since) -> dict:
    bot_accounts = set(config.get("bot_accounts", []))
    team = collect_team(raw)
    rows = []
    for issue in raw:
        author_actor = issue.get("author")
        if is_bot(author_actor, bot_accounts):
            continue
        author = login_of(author_actor) or "ghost"
        fixer_actor, reference = find_fix(issue)
        fixer = login_of(fixer_actor)
        if is_bot(fixer_actor, bot_accounts) or (fixer or "").lower().startswith("copilot"):
            fixer_kind = 2
        elif fixer in team:
            fixer_kind = 1
        else:
            fixer_kind = 0
        rows.append([
            repos.index(issue["repo"]),
            issue["number"],
            issue["title"][:110],
            author,
            OUTCOME_CODES[outcome_for(issue, author, fixer)],
            fixer or "",
            reference,
            fixer_kind,
            issue["createdAt"][:10],
            (issue["closedAt"] or "")[:10],
        ])
    return {
        "title": config.get("title", "Issue credit"),
        "team_label": config.get("team_label", "Team"),
        "days": config.get("days", 365),
        "space": space,
        "generated": today.isoformat(),
        "since": since.isoformat(),
        "repos": repos,
        "team": team,
        "issues": rows,
    }


# ---------- Space files ----------

def read_space_file(space: str, filename: str, local_override: str | None) -> str:
    if local_override:
        return Path(local_override).read_text()
    return Path(hf_hub_download(space, filename, repo_type="space")).read_text()


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--space", required=True, help="Space to read config from and write to, e.g. you/issue-credit")
    parser.add_argument("--config", help="Use a local config.toml instead of the Space's copy")
    parser.add_argument("--template", help="Use a local page.template.html instead of the Space's copy")
    parser.add_argument("--out", help="Write index.html and data.json to this folder instead of uploading")
    args = parser.parse_args()

    token = os.environ.get("GITHUB_TOKEN", "").strip()
    if not token:
        sys.exit("Set GITHUB_TOKEN. GitHub's GraphQL API needs a token, even for public repos.")

    config = tomllib.loads(read_space_file(args.space, "config.toml", args.config))
    template = read_space_file(args.space, "page.template.html", args.template)
    repos = list(dict.fromkeys(config["repos"]))
    today = datetime.now(UTC).date()
    if "since" in config:
        since = date.fromisoformat(str(config["since"]))
    else:
        since = today - timedelta(days=int(config.get("days", 365)))

    print(f"Fetching {len(repos)} repos, issues opened since {since}", file=sys.stderr)
    with ThreadPoolExecutor(max_workers=4) as pool:
        per_repo = list(pool.map(lambda repo: fetch_repo(repo, since.isoformat(), token), repos))
    found = [repo for repo, issues in zip(repos, per_repo) if issues]
    raw = [issue for issues in per_repo for issue in issues]

    payload = build_payload(raw, config, found, args.space, today, since)
    data_json = json.dumps(payload, separators=(",", ":"), ensure_ascii=False)
    index_html = template.replace("/*__DATA__*/null", data_json.replace("</", "<\\/"))
    if "/*__DATA__*/null" in index_html or index_html == template:
        sys.exit("The template has no /*__DATA__*/null placeholder.")

    # Belt and braces: never upload anything that contains the token.
    if token in data_json or token in index_html:
        sys.exit("Refusing to upload: the GitHub token appeared in the output.")

    summary = f"{len(payload['issues'])} issues from {len(found)} repos"
    if args.out:
        out = Path(args.out)
        out.mkdir(parents=True, exist_ok=True)
        (out / "index.html").write_text(index_html)
        (out / "data.json").write_text(data_json)
        print(f"Wrote {summary} to {out}/", file=sys.stderr)
        return

    HfApi().create_commit(
        repo_id=args.space,
        repo_type="space",
        operations=[
            CommitOperationAdd("index.html", index_html.encode()),
            CommitOperationAdd("data.json", data_json.encode()),
        ],
        commit_message=f"Refresh data {today}: {summary}",
    )
    print(f"Uploaded {summary} to https://huggingface.co/spaces/{args.space}", file=sys.stderr)


if __name__ == "__main__":
    main()