[2026-09-30] Compile a list of merged GitHub PRs

list_merged_prs.py

import csv
import datetime
import enum
import pathlib
import subprocess

import pydantic

_HERE = pathlib.Path(__file__).resolve().parent
# GitHub search returns at most 1000 results per query
_SEARCH_CAP = 1000
_FIELDS = "number,title,author,mergedAt,url,reviews"
_ORG = "kaiju"
_REPO = "hedgehog"
_RANGES_ISO8601 = (
    "2026-07-01T00:00:00Z..2026-08-01T00:00:00Z",
    "2026-08-01T00:00:00Z..2026-09-01T00:00:00Z",
    "2026-09-01T00:00:00Z..2026-10-01T00:00:00Z",
)


class _ForbidExtra(pydantic.BaseModel):
    model_config = pydantic.ConfigDict(
        extra="forbid", validate_by_name=True, validate_by_alias=True
    )


class _ReviewState(enum.Enum):
    APPROVED = "APPROVED"
    CHANGES_REQUESTED = "CHANGES_REQUESTED"
    COMMENTED = "COMMENTED"
    DISMISSED = "DISMISSED"
    PENDING = "PENDING"


class _AuthorAssociation(enum.Enum):
    COLLABORATOR = "COLLABORATOR"
    CONTRIBUTOR = "CONTRIBUTOR"
    MEMBER = "MEMBER"
    NONE = "NONE"


class _Reaction(enum.Enum):
    EYES = "EYES"
    HEART = "HEART"
    HOORAY = "HOORAY"
    LAUGH = "LAUGH"
    THUMBS_UP = "THUMBS_UP"
    ROCKET = "ROCKET"


class _ReviewCommit(_ForbidExtra):
    oid: str


class _ReactionUsers(_ForbidExtra):
    total_count: int = pydantic.Field(alias="totalCount")


class _ReactionGroup(_ForbidExtra):
    content: _Reaction
    users: _ReactionUsers


class _ReviewAuthor(_ForbidExtra):
    login: str


class _PRReview(_ForbidExtra):
    id_: str = pydantic.Field(alias="id", default=None)
    author: _ReviewAuthor
    author_association: _AuthorAssociation = pydantic.Field(alias="authorAssociation")
    body: str
    submitted_at: pydantic.AwareDatetime | None = pydantic.Field(alias="submittedAt")
    includes_created_edit: bool = pydantic.Field(alias="includesCreatedEdit")
    reaction_groups: list[_ReactionGroup] = pydantic.Field(alias="reactionGroups")
    state: _ReviewState
    commit: _ReviewCommit


class _PRAuthor(_ForbidExtra):
    id_: str | None = pydantic.Field(alias="id", default=None)
    is_bot: bool
    login: str
    name: str | None = pydantic.Field(default=None)


class _ParsedPR(_ForbidExtra):
    author: _PRAuthor
    merged_at: pydantic.AwareDatetime = pydantic.Field(alias="mergedAt")
    number: int
    title: str
    url: str
    reviews: list[_PRReview]


class _ParsedPRs(pydantic.RootModel[list[_ParsedPR]]):
    pass


def _fetch_range(repo: str, range_str: str) -> list[_ParsedPR]:
    cmd = (
        "gh",
        "pr",
        "list",
        "--repo",
        repo,
        "--state",
        "merged",
        "--search",
        f"merged:{range_str}",
        "--json",
        _FIELDS,
        "--limit",
        str(_SEARCH_CAP),
    )

    result_json = subprocess.check_output(cmd, encoding="utf-8")
    parsed_prs_root = _ParsedPRs.model_validate_json(result_json)
    parsed_prs = parsed_prs_root.root

    if len(parsed_prs) >= _SEARCH_CAP:
        raise RuntimeError("Range has too many PRs", len(parsed_prs), range_str)

    return parsed_prs


def _get_approval_count(reviews: list[_PRReview]) -> int:
    latest: dict[str, _PRReview] = {}
    for review in reviews:
        if review.state not in (
            _ReviewState.APPROVED,
            _ReviewState.CHANGES_REQUESTED,
            _ReviewState.DISMISSED,
        ):
            continue

        login = review.author.login
        existing = latest.get(login)
        if existing is None:
            latest[login] = review
            continue

        if existing.submitted_at is None or review.submitted_at is None:
            # NOTE: `submitted_at` can only be none on `PENDING` reviews
            raise NotImplementedError

        if existing.submitted_at < review.submitted_at:
            latest[login] = review

    return sum(1 for review in latest.values() if review.state == _ReviewState.APPROVED)


def _sort_func(parsed_pr: _ParsedPR) -> tuple[datetime.datetime, int]:
    return parsed_pr.merged_at, parsed_pr.number


def main() -> None:
    by_number: dict[int, _ParsedPR] = {}
    for range_str in _RANGES_ISO8601:
        parsed_prs = _fetch_range(f"{_ORG}/{_REPO}", range_str)
        for parsed_pr in parsed_prs:
            if parsed_pr.number in by_number:
                raise NotImplementedError(parsed_pr.number)

            by_number[parsed_pr.number] = parsed_pr

    as_list = sorted(by_number.values(), key=_sort_func)

    csv_rows: list[dict] = []
    for parsed_pr in as_list:
        pr_number = parsed_pr.number
        author_login = parsed_pr.author.login
        approval_count = _get_approval_count(parsed_pr.reviews)
        csv_rows.append({
            "repo": _REPO,
            "number": pr_number,
            "title": parsed_pr.title,
            "author": author_login,
            "merged_at": parsed_pr.merged_at.isoformat(),
            "approval_count": approval_count,
        })

    with open(_HERE / f"merged_prs.{_REPO}.csv", "w") as file_obj:
        writer = csv.DictWriter(
            file_obj,
            fieldnames=(
                "repo",
                "number",
                "title",
                "author",
                "merged_at",
                "approval_count",
            ),
        )
        writer.writeheader()
        writer.writerows(csv_rows)


if __name__ == "__main__":
    main()