Spaces:
Running
Running
| # /// 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() | |