feat(github-all-prs): add needs-review+needs-work commands

Author: okurz

github-all-prs

@@ -0,0 +1,132 @@
+#!/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()