github-all-prs

#!/usr/bin/env python3
import json
import subprocess
import concurrent.futures
import typer
from typing import Optional
from datetime import datetime, timezone
from dateutil.parser import isoparse

app = typer.Typer(help="List GitHub PRs needing review or work in Slack-compatible format.", no_args_is_help=False)

def run_gh(args: list[str]) -> Optional[str]:
    """Execute a gh CLI command and return its stdout."""
    try:
        res = subprocess.run(["gh"] + args, capture_output=True, text=True, check=True)
        return res.stdout.strip()
    except subprocess.CalledProcessError:
        return None

def get_me() -> str:
    """Return the current user's GitHub login."""
    return run_gh(["api", "user", "--jq", ".login"]) or "@me"

def parse_date(date_str: str) -> datetime:
    """Parse ISO 8601 date string from GitHub."""
    if not date_str:
        return datetime.fromtimestamp(0, tz=timezone.utc)
    d = isoparse(date_str)
    if d.tzinfo is None:
        d = d.replace(tzinfo=timezone.utc)
    return d

def get_pr_details(url: str) -> dict:
    """Fetch detailed PR status."""
    fields = "url,title,reviewDecision,statusCheckRollup,latestReviews,comments,commits"
    data_raw = run_gh(["pr", "view", url, "--json", fields])
    return json.loads(data_raw) if data_raw else {}

def needs_review_filter(pr: dict) -> Optional[str]:
    """PR is ready for review by others (no changes requested)."""
    if pr.get("reviewDecision") != "CHANGES_REQUESTED":
        return f"* {pr['url']} {pr['title']}"
    return None

def needs_work_filter(pr: dict, me: str) -> Optional[str]:
    """PR needs work by the author (CI failure or unanswered changes requested)."""
    reasons = []
    # Check CI failures
    checks = pr.get("statusCheckRollup", [])
    if any(c.get("conclusion") in ["FAILURE", "ACTION_REQUIRED", "TIMED_OUT", "CANCELLED", "ERROR"] for c in checks):
        reasons.append("CI failing")
    # Check Review Decision
    if pr.get("reviewDecision") == "CHANGES_REQUESTED":
        cr_reviews = [r for r in pr.get("latestReviews", []) if r.get("state") == "CHANGES_REQUESTED"]
        if cr_reviews:
            # Find latest Changes Requested review
            dates = [parse_date(r.get("updatedAt")) for r in cr_reviews if r.get("updatedAt")]
            if dates:
                latest_cr = max(dates)
                # Answered by comment?
                answered = False
                for c in pr.get("comments", []):
                    if c.get("author", {}).get("login") == me:
                        if parse_date(c.get("createdAt")) > latest_cr:
                            answered = True
                            break
                # Answered by commit?
                if not answered:
                    for c in pr.get("commits", []):
                        if parse_date(c.get("committedDate")) > latest_cr:
                            answered = True
                            break
                if not answered:
                    reasons.append("Changes requested")
            else:
                reasons.append("Changes requested")
    if reasons:
        return f"* {pr['url']} {pr['title']} ({', '.join(reasons)})"
    return None


def list_needs_review(limit: int, workers: int):
    """Core logic for needs-review."""
    query = f"search prs --author @me --state open --checks success --sort updated --limit {limit} --json url,title,isDraft"
    raw_prs = run_gh(query.split())
    if not raw_prs:
        return
    prs = [p for p in json.loads(raw_prs) if not p["isDraft"]]
    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
        details = list(executor.map(lambda p: get_pr_details(p["url"]), prs))
    output = "\n".join(filter(None, [needs_review_filter(d) for d in details]))
    if output:
        typer.echo(output)

def list_needs_work(limit: int, workers: int):
    """Core logic for needs-work."""
    me = get_me()
    query = f"search prs --author @me --state open --sort updated --limit {limit} --json url,title,isDraft"
    raw_prs = run_gh(query.split())
    if not raw_prs:
        return
    prs = [p for p in json.loads(raw_prs) if not p["isDraft"]]
    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
        details = list(executor.map(lambda p: get_pr_details(p["url"]), prs))
    output = "\n".join(filter(None, [needs_work_filter(d, me) for d in details]))
    if output:
        typer.echo(output)

@app.command()
def needs_review(
    limit: int = typer.Option(100, help="Maximum number of PRs to search."),
    workers: int = typer.Option(10, help="Number of parallel workers for status checks."),
):
    """Fetch and display open PRs by current user that are ready for review (CI success, no changes requested)."""
    list_needs_review(limit, workers)

@app.command()
def needs_work(
    limit: int = typer.Option(100, help="Maximum number of PRs to search."),
    workers: int = typer.Option(10, help="Number of parallel workers for status checks."),
):
    """Fetch and display open PRs by current user that need work (CI failure or unanswered changes requested)."""
    list_needs_work(limit, workers)

@app.callback(invoke_without_command=True)
def main(ctx: typer.Context):
    """Default to needs-review if no command provided."""
    if ctx.invoked_subcommand is None:
        list_needs_review(100, 10)

if __name__ == "__main__":
    app()