Create pr_summary.py

scripts/pr_summary.py

@@ -0,0 +1,218 @@
+import asyncio
+import json
+import re
+from pathlib import Path
+from typing import Optional, List, Dict, Any
+import typer
+from rich.console import Console
+from InquirerPy import inquirer
+from InquirerPy.base.control import Choice
+
+app = typer.Typer(
+    help="FastAPI/Python Dev Tool: Auto-summarize GitHub PR diffs grouped by module.",
+    add_completion=False,
+)
+console = Console(stderr=True)  # Route log status messages to stderr
+
+# Default ignore patterns (lockfiles & autogenerated migrations)
+IGNORE_PATTERNS = [
+    re.compile(r"(^|/)(poetry\.lock|package-lock\.json|pnpm-lock\.yaml|Pipfile\.lock|requirements\.txt)$"),
+    re.compile(r"^alembic/versions/.*\.py$"),
+]
+
+
+def is_ignored_file(path: str, include_all: bool) -> bool:
+    if include_all:
+        return False
+    return any(pattern.search(path) for pattern in IGNORE_PATTERNS)
+
+
+def extract_module(file_path: str) -> str:
+    parts = Path(file_path).parts
+    if len(parts) == 1:
+        return "root files"
+    return parts[0]
+
+
+async def fetch_open_prs(limit: int = 15) -> List[Dict[str, Any]]:
+    """Asynchronously fetches recent open PRs using gh CLI."""
+    cmd = [
+        "gh", "pr", "list",
+        "--state", "open",
+        "--limit", str(limit),
+        "--json", "number,title,author,headRefName"
+    ]
+
+    proc = await asyncio.create_subprocess_exec(
+        *cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE
+    )
+    stdout, stderr = await proc.communicate()
+
+    if proc.returncode != 0:
+        console.print(f"[bold red]Error listing open PRs:[/bold red] {stderr.decode().strip()}")
+        raise typer.Exit(code=1)
+
+    return json.loads(stdout.decode())
+
+
+def select_pr_interactively() -> str:
+    """Displays an interactive selection menu using InquirerPy."""
+    console.print("[dim]Fetching open Pull Requests...[/dim]")
+    open_prs = asyncio.run(fetch_open_prs())
+
+    if not open_prs:
+        console.print("[bold yellow]No open Pull Requests found in this repository.[/bold yellow]")
+        raise typer.Exit(code=0)
+
+    # Format choices for InquirerPy
+    choices = [
+        Choice(
+            value=str(pr["number"]),
+            name=f"#{pr['number']} - {pr['title']} (@{pr['author']['login']}) [{pr['headRefName']}]"
+        )
+        for pr in open_prs
+    ]
+
+    selected_pr_number = inquirer.select(
+        message="Select an open PR to summarize:",
+        choices=choices,
+        default=choices[0].value,
+    ).execute()
+
+    return selected_pr_number
+
+
+async def fetch_pr_json(pr_id: str) -> Dict[str, Any]:
+    """Asynchronously calls `gh pr view` to retrieve raw JSON metadata."""
+    cmd = [
+        "gh", "pr", "view", pr_id,
+        "--json", "number,title,author,additions,deletions,changedFiles,files"
+    ]
+
+    proc = await asyncio.create_subprocess_exec(
+        *cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE
+    )
+    stdout, stderr = await proc.communicate()
+
+    if proc.returncode != 0:
+        console.print(f"[bold red]Error running gh CLI:[/bold red] {stderr.decode().strip()}")
+        raise typer.Exit(code=1)
+
+    return json.loads(stdout.decode())
+
+
+def process_pr_data(
+    raw_data: Dict[str, Any],
+    extension: Optional[str] = None,
+    include_all: bool = False
+) -> Dict[str, Any]:
+    """Filters, aggregates, and groups PR file changes by module."""
+    raw_files = raw_data.get("files", [])
+
+    normalized_ext = extension if not extension else (f".{extension}" if not extension.startswith(".") else extension)
+
+    filtered_files = []
+    for f in raw_files:
+        path = f["path"]
+        if is_ignored_file(path, include_all):
+            continue
+        if normalized_ext and not path.endswith(normalized_ext):
+            continue
+        filtered_files.append(f)
+
+    tot_additions = sum(f["additions"] for f in filtered_files)
+    tot_deletions = sum(f["deletions"] for f in filtered_files)
+
+    modules_map: Dict[str, Dict[str, Any]] = {}
+    for f in filtered_files:
+        mod = extract_module(f["path"])
+        if mod not in modules_map:
+            modules_map[mod] = {
+                "module": mod,
+                "count": 0,
+                "additions": 0,
+                "deletions": 0,
+                "files": []
+            }
+        modules_map[mod]["count"] += 1
+        modules_map[mod]["additions"] += f["additions"]
+        modules_map[mod]["deletions"] += f["deletions"]
+        modules_map[mod]["files"].append(f["path"])
+
+    return {
+        "number": raw_data["number"],
+        "title": raw_data["title"],
+        "author": raw_data["author"]["login"],
+        "stats": {
+            "changed_files": len(filtered_files),
+            "additions": tot_additions,
+            "deletions": tot_deletions,
+        },
+        "filters": {
+            "extension": normalized_ext,
+            "include_all": include_all,
+        },
+        "modules": list(modules_map.values())
+    }
+
+
+def format_markdown(summary: Dict[str, Any]) -> str:
+    """Formats processed PR summary into GitHub-Flavored Markdown."""
+    if summary["stats"]["changed_files"] == 0:
+        return (
+            f"### PR #{summary['number']}: {summary['title']}\n"
+            f"**Author:** @{summary['author']}\n\n"
+            f"⚠️ No relevant files matching constraints were found in this PR."
+        )
+
+    ext_info = f"[Filtered by `{summary['filters']['extension']}`] " if summary['filters']['extension'] else ""
+    inc_info = "[Including lockfiles/migrations]" if summary['filters']['include_all'] else "[Lockfiles & Alembic migrations ignored]"
+
+    lines = [
+        f"### PR #{summary['number']}: {summary['title']}",
+        f"**Author:** @{summary['author']} | **Stats:** {summary['stats']['changed_files']} files changed (`+{summary['stats']['additions']}` / `-{summary['stats']['deletions']}`) {ext_info}{inc_info}",
+        "",
+        "#### 📂 Changes by Module",
+        ""
+    ]
+
+    for mod in summary["modules"]:
+        lines.append(f"- **`{mod['module']}/`** ({mod['count']} files) [`+{mod['additions']}` / `-{mod['deletions']}`]")
+        for filepath in mod["files"]:
+            lines.append(f"  - `{filepath}`")
+
+    return "\n".join(lines)
+
+
+@app.command()
+def main(
+    pr_id: Optional[str] = typer.Argument(None, help="Pull Request number or branch name. Interactive menu opens if omitted."),
+    extension: Optional[str] = typer.Option(None, "--ext", "-e", help="Filter strictly by file extension (e.g. py, sql)"),
+    include_all: bool = typer.Option(False, "--include-all", "-a", help="Include lockfiles and Alembic migrations"),
+    json_out: bool = typer.Option(False, "--json", "-j", help="Output raw JSON instead of Markdown")
+):
+    """
+    Summarize GitHub PR changes by top-level module/directory.
+    """
+    # If no PR ID is passed, open the interactive InquirerPy selection menu
+    if not pr_id:
+        pr_id = select_pr_interactively()
+
+    console.print(f"[dim]Fetching PR #{pr_id} stats via gh CLI...[/dim]")
+
+    raw_data = asyncio.run(fetch_pr_json(pr_id))
+    summary = process_pr_data(raw_data, extension=extension, include_all=include_all)
+
+    if json_out:
+        print(json.dumps(summary, indent=2))
+    else:
+        md_text = format_markdown(summary)
+        print(md_text)
+
+
+if __name__ == "__main__":
+    app()