Script to pull all the issues and their comments from a GitHub Repository for analysis

pull_issues.py

#!/usr/bin/env python3
"""Pull all issues and comments from a public GitHub repo into JSON files.

Usage:
    python3 pull_issues.py [--repo OWNER/NAME] [--out DIR]

Defaults to swicg/activitypub-e2ee -> ./e2ee-data/.

Output layout:
    <out>/issues.json       Flat array of all issues (PRs filtered out).
    <out>/comments.json     Flat array of every issue comment in the repo.
    <out>/issues/<n>.json   Per-issue bundle: {"issue": {...}, "comments": [...]}.

If the environment variable GITHUB_TOKEN is set, requests are authenticated
(5000 req/hr); otherwise unauthenticated (60 req/hr). Total cost is roughly
5 requests regardless of issue count, because we use the repo-wide
/issues/comments endpoint rather than fetching comments per issue.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any


def _auth_headers() -> dict[str, str]:
    headers = {"Accept": "application/vnd.github+json"}
    token = os.environ.get("GITHUB_TOKEN")
    if token:
        headers["Authorization"] = f"Bearer {token}"
    return headers


def fetch_paginated(url: str) -> list[dict[str, Any]]:
    """Fetch every page of a GitHub list endpoint and return a flat list."""
    sep = "&" if "?" in url else "?"
    results: list[dict[str, Any]] = []
    page = 1
    while True:
        page_url = f"{url}{sep}per_page=100&page={page}"
        req = urllib.request.Request(page_url, headers=_auth_headers())
        try:
            with urllib.request.urlopen(req) as resp:
                batch = json.load(resp)
        except urllib.error.HTTPError as e:
            body = e.read().decode("utf-8", errors="replace")
            sys.exit(f"GitHub API error {e.code} on {page_url}: {body}")

        if not batch:
            break
        results.extend(batch)
        if len(batch) < 100:
            break
        page += 1
    return results


def comment_issue_number(comment: dict[str, Any]) -> int:
    # `issue_url` ends with /issues/<n>
    return int(comment["issue_url"].rsplit("/", 1)[-1])


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Pull all issues and comments from a public GitHub repo."
    )
    parser.add_argument(
        "--repo",
        default="swicg/activitypub-e2ee",
        help="GitHub repo as OWNER/NAME (default: swicg/activitypub-e2ee)",
    )
    parser.add_argument(
        "--out",
        type=Path,
        default=Path("./e2ee-data"),
        help="output directory (default: ./e2ee-data)",
    )
    args = parser.parse_args()

    api = f"https://api.github.com/repos/{args.repo}"
    out = args.out
    (out / "issues").mkdir(parents=True, exist_ok=True)

    auth_state = "authenticated" if os.environ.get("GITHUB_TOKEN") else "unauthenticated"
    print(f"==> Fetching issues from {args.repo} ({auth_state})")
    issues = [i for i in fetch_paginated(f"{api}/issues?state=all")
              if "pull_request" not in i]
    print(f"    {len(issues)} issues (PRs excluded)")

    print("==> Fetching all comments")
    comments = fetch_paginated(f"{api}/issues/comments")
    print(f"    {len(comments)} comments")

    (out / "issues.json").write_text(json.dumps(issues, indent=2))
    (out / "comments.json").write_text(json.dumps(comments, indent=2))

    print("==> Writing per-issue bundles")
    by_issue: dict[int, list[dict[str, Any]]] = {}
    for c in comments:
        by_issue.setdefault(comment_issue_number(c), []).append(c)
    for issue in issues:
        n = issue["number"]
        bundle = {"issue": issue, "comments": by_issue.get(n, [])}
        (out / "issues" / f"{n}.json").write_text(json.dumps(bundle, indent=2))

    print()
    print("Done. Output:")
    print(f"  {out}/issues.json")
    print(f"  {out}/comments.json")
    print(f"  {out}/issues/<n>.json  (one bundle per issue)")


if __name__ == "__main__":
    main()