fetch-pr-reviews.py

gistfile1.txt

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["click", "tqdm"]
# ///
"""Fetch GitHub PR reviews + comments into a SQLite database via `gh api`."""

import json
import re
import sqlite3
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from urllib.parse import quote

import click
from tqdm import tqdm


SCHEMA = """
CREATE TABLE IF NOT EXISTS prs (
    repo TEXT NOT NULL,
    number INTEGER NOT NULL,
    title TEXT,
    state TEXT,
    author TEXT,
    body TEXT,
    url TEXT,
    created_at TEXT,
    updated_at TEXT,
    merged_at TEXT,
    closed_at TEXT,
    base_ref TEXT,
    head_ref TEXT,
    PRIMARY KEY (repo, number)
);

CREATE TABLE IF NOT EXISTS reviews (
    repo TEXT NOT NULL,
    id INTEGER NOT NULL,
    pr_number INTEGER NOT NULL,
    author TEXT,
    state TEXT,
    body TEXT,
    submitted_at TEXT,
    commit_id TEXT,
    PRIMARY KEY (repo, id)
);

CREATE TABLE IF NOT EXISTS review_comments (
    repo TEXT NOT NULL,
    id INTEGER NOT NULL,
    pr_number INTEGER NOT NULL,
    review_id INTEGER,
    in_reply_to_id INTEGER,
    author TEXT,
    body TEXT,
    path TEXT,
    line INTEGER,
    original_line INTEGER,
    start_line INTEGER,
    side TEXT,
    commit_id TEXT,
    diff_hunk TEXT,
    created_at TEXT,
    updated_at TEXT,
    PRIMARY KEY (repo, id)
);

CREATE TABLE IF NOT EXISTS issue_comments (
    repo TEXT NOT NULL,
    id INTEGER NOT NULL,
    pr_number INTEGER NOT NULL,
    author TEXT,
    body TEXT,
    created_at TEXT,
    updated_at TEXT,
    PRIMARY KEY (repo, id)
);

CREATE INDEX IF NOT EXISTS idx_reviews_pr ON reviews(repo, pr_number);
CREATE INDEX IF NOT EXISTS idx_review_comments_pr ON review_comments(repo, pr_number);
CREATE INDEX IF NOT EXISTS idx_issue_comments_pr ON issue_comments(repo, pr_number);
"""


UNIT_HOURS = {"h": 1, "d": 24, "w": 24 * 7, "m": 24 * 30, "y": 24 * 365}


def parse_since(s: str) -> datetime:
    m = re.fullmatch(r"\s*(\d+)\s*([hdwmy])\s*", s.lower())
    if not m:
        raise click.BadParameter(
            f"invalid --since: {s!r}; expected like 24h, 7d, 2w, 1m, 1y"
        )
    n, unit = int(m.group(1)), m.group(2)
    return datetime.now(timezone.utc) - timedelta(hours=n * UNIT_HOURS[unit])


def detect_repo() -> str:
    try:
        out = subprocess.check_output(
            ["gh", "repo", "view", "--json", "nameWithOwner", "-q", ".nameWithOwner"],
            text=True,
            stderr=subprocess.PIPE,
        ).strip()
        if out:
            return out
    except (subprocess.CalledProcessError, FileNotFoundError) as e:
        raise click.UsageError(
            f"could not detect repo (run inside a gh-linked repo or pass --repo OWNER/NAME): {e}"
        )
    raise click.UsageError("could not detect repo; pass --repo OWNER/NAME")


def gh_api(path: str) -> object:
    result = subprocess.run(
        ["gh", "api", "-H", "Accept: application/vnd.github+json", path],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        raise RuntimeError(f"gh api {path} failed:\n{result.stderr}")
    body = result.stdout.strip()
    return json.loads(body) if body else []


def gh_paginate(path: str, *, stop_when=None):
    """Yield items across pages. If stop_when(item) is truthy, stop after that item."""
    page = 1
    sep = "&" if "?" in path else "?"
    while True:
        items = gh_api(f"{path}{sep}page={page}&per_page=100")
        if not items:
            return
        for item in items:
            if stop_when and stop_when(item):
                return
            yield item
        if len(items) < 100:
            return
        page += 1


def upsert(conn, table, row):
    placeholders = ",".join("?" * len(row))
    conn.execute(f"INSERT OR REPLACE INTO {table} VALUES ({placeholders})", row)


@click.command()
@click.option("--repo", default=None, help="OWNER/NAME (defaults to current gh repo)")
@click.option(
    "--db",
    default="pr_reviews.db",
    show_default=True,
    type=click.Path(dir_okay=False),
    help="SQLite database path",
)
@click.option(
    "--since",
    default="30d",
    show_default=True,
    help="time window: 24h, 7d, 30d, 2w, 1m, 1y",
)
@click.option(
    "--state",
    default="all",
    show_default=True,
    type=click.Choice(["all", "open", "closed"]),
)
@click.option(
    "--full",
    is_flag=True,
    help="disable incremental skip; refetch reviews/comments for every PR in window",
)
@click.option("-v", "--verbose", is_flag=True)
def main(repo, db, since, state, full, verbose):
    """Fetch PRs, reviews, inline review comments, and issue comments into SQLite."""
    cutoff = parse_since(since)
    repo = repo or detect_repo()
    cutoff_iso = cutoff.strftime("%Y-%m-%dT%H:%M:%SZ")
    click.echo(f"repo: {repo}  since: {cutoff_iso}  db: {db}", err=True)

    conn = sqlite3.connect(db)
    conn.executescript(SCHEMA)

    known = {} if full else {
        n: u for n, u in conn.execute(
            "SELECT number, updated_at FROM prs WHERE repo = ?", (repo,)
        )
    }

    pulls = f"/repos/{repo}/pulls?state={state}&sort=updated&direction=desc"
    n_prs = n_skipped = n_reviews = n_rcomments = n_icomments = 0

    total = None
    try:
        state_q = "" if state == "all" else f" is:{state}"
        q = f"repo:{repo} is:pr{state_q} updated:>={cutoff_iso}"
        search = gh_api(f"/search/issues?q={quote(q)}&per_page=1")
        total = search.get("total_count") if isinstance(search, dict) else None
    except Exception:
        pass

    bar = tqdm(total=total, unit="pr", desc="PRs", disable=verbose)

    for pr in gh_paginate(pulls, stop_when=lambda p: (p.get("updated_at") or "") < cutoff_iso):
        num = pr["number"]
        api_updated = pr.get("updated_at")
        unchanged = known.get(num) == api_updated and api_updated is not None
        if verbose:
            tag = "skip" if unchanged else " new"
            click.echo(f"  {tag} #{num} {pr.get('title', '')[:70]}", err=True)
        upsert(conn, "prs", (
            repo, num, pr.get("title"), pr.get("state"),
            (pr.get("user") or {}).get("login"),
            pr.get("body"), pr.get("html_url"),
            pr.get("created_at"), api_updated,
            pr.get("merged_at"), pr.get("closed_at"),
            (pr.get("base") or {}).get("ref"),
            (pr.get("head") or {}).get("ref"),
        ))
        n_prs += 1
        if unchanged:
            n_skipped += 1
            conn.commit()
            continue

        for r in gh_paginate(f"/repos/{repo}/pulls/{num}/reviews"):
            upsert(conn, "reviews", (
                repo, r["id"], num,
                (r.get("user") or {}).get("login"),
                r.get("state"), r.get("body"),
                r.get("submitted_at"), r.get("commit_id"),
            ))
            n_reviews += 1

        for c in gh_paginate(f"/repos/{repo}/pulls/{num}/comments"):
            upsert(conn, "review_comments", (
                repo, c["id"], num,
                c.get("pull_request_review_id"),
                c.get("in_reply_to_id"),
                (c.get("user") or {}).get("login"),
                c.get("body"), c.get("path"),
                c.get("line"), c.get("original_line"),
                c.get("start_line"), c.get("side"),
                c.get("commit_id"), c.get("diff_hunk"),
                c.get("created_at"), c.get("updated_at"),
            ))
            n_rcomments += 1

        for ic in gh_paginate(f"/repos/{repo}/issues/{num}/comments"):
            upsert(conn, "issue_comments", (
                repo, ic["id"], num,
                (ic.get("user") or {}).get("login"),
                ic.get("body"),
                ic.get("created_at"), ic.get("updated_at"),
            ))
            n_icomments += 1

        conn.commit()
        bar.set_postfix(reviews=n_reviews, comments=n_rcomments + n_icomments, skipped=n_skipped)
        bar.update(1)

    bar.close()
    conn.close()
    click.echo(
        f"saved {n_prs} PRs ({n_skipped} unchanged, skipped), "
        f"{n_reviews} reviews, {n_rcomments} inline comments, "
        f"{n_icomments} issue comments",
        err=True,
    )


if __name__ == "__main__":
    main()