From afbc9d0194f23f0810f5422a67d04f609b76305c Mon Sep 17 00:00:00 2001 From: Zyntro Date: Sat, 29 Aug 2026 21:45:11 +0700 Subject: [PATCH] Create pr_summary.py Signed-off-by: Zyntro --- scripts/pr_summary.py | 218 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 218 insertions(+) create mode 100644 scripts/pr_summary.py diff --git a/scripts/pr_summary.py b/scripts/pr_summary.py new file mode 100644 index 0000000..414f508 --- /dev/null +++ b/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()