MQL5-Google-Onedrive/scripts/review_pull_requests.py

#!/usr/bin/env python3
"""
Pull Request Review Script
Reviews all pull requests and creates a comprehensive summary
"""

import subprocess
import sys
import json
from pathlib import Path
from datetime import datetime
from collections import defaultdict

REPO_ROOT = Path(__file__).resolve().parents[1]

# ⚡ Bolt: Global cache for branch metadata to avoid redundant git calls.
BRANCH_METADATA = {}


def run_command(cmd, capture_output=True):
    """Run a command and return the result."""
    try:
        result = subprocess.run(
            cmd,
            cwd=REPO_ROOT,
            capture_output=capture_output,
            text=True,
            timeout=30,
            encoding='utf-8',
            errors='replace'
        )
        return result
    except FileNotFoundError:
        # ⚡ Bolt: Only silence 'gh' missing. Other missing commands (like 'git') should raise.
        if cmd and cmd[0] == "gh":
            return None
        raise
    except Exception as e:
        print(f"Error running command {cmd}: {e}", file=sys.stderr)
        return None


def populate_branch_metadata():
    """⚡ Bolt: Fetch all branch metadata in a single call using git for-each-ref."""
    global BRANCH_METADATA

    # ⚡ Bolt: Use %(ahead-behind:main) to get commit counts relative to main.
    # This requires Git 2.41+. Format: branch_name | ahead behind | date | subject
    fmt = "%(refname:short)|%(ahead-behind:main)|%(committerdate:iso8601)|%(subject)"
    result = run_command(["git", "for-each-ref", f"--format={fmt}", "refs/remotes/origin"])

    if not result or result.returncode != 0:
        return

    # Check if ahead-behind is supported by the installed Git version
    use_fallback = "%(ahead-behind:main)" in result.stdout

    if use_fallback:
        # Fallback for older Git versions: fetch basic info, ahead/behind will be fetched lazily
        fmt = "%(refname:short)|0 0|%(committerdate:iso8601)|%(subject)"
        result = run_command(["git", "for-each-ref", f"--format={fmt}", "refs/remotes/origin"])
        if not result or result.returncode != 0:
            return

    for line in result.stdout.strip().split("\n"):
        if not line:
            continue
        try:
            # ⚡ Bolt: Subject might contain pipes, so we limit the split to 3 (4 parts).
            parts = line.split("|", 3)
            if len(parts) < 4:
                continue

            ref, ab, date, subject = parts

            # ⚡ Bolt: Key by short name (e.g. 'feat/foo') for consistent lookup.
            short_ref = ref.replace("origin/", "")

            # ⚡ Bolt: Filter out administrative refs. 'origin' represents origin/HEAD.
            if short_ref in ["HEAD", "main"] or ref == "origin":
                continue

            if use_fallback:
                ahead, behind = None, None
            else:
                try:
                    ahead, behind = map(int, ab.split())
                except (ValueError, IndexError):
                    ahead, behind = None, None

            BRANCH_METADATA[short_ref] = {
                "ahead": ahead,
                "behind": behind,
                "date": date,
                "subject": subject,
                "full_name": ref if ref.startswith("origin/") else f"origin/{ref}"
            }
        except Exception:
            continue


def get_prs_via_gh_cli():
    """Get PRs using GitHub CLI."""
    result = run_command(["gh", "pr", "list", "--state", "all", "--json", "number,title,state,author,createdAt,updatedAt,headRefName,baseRefName,isDraft,labels"])
    if result and result.returncode == 0:
        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            return []
    return None


def get_prs_via_git():
    """Get PR information via git branches."""
    if not BRANCH_METADATA:
        populate_branch_metadata()

    # ⚡ Bolt: Check if ahead-behind info is available (Git 2.41+)
    has_ab = any(m.get("ahead") is not None for m in BRANCH_METADATA.values())

    if has_ab:
        # Use cache for open branches (those with commits ahead of main)
        open_branches = [m["full_name"] for m in BRANCH_METADATA.values() if m["ahead"] and m["ahead"] > 0]

        # For merged branches, the cache alone isn't enough to distinguish from stale branches.
        # Use the specialized git command (it's fast).
        merged_branches = []
        result_merged = run_command(["git", "branch", "-r", "--merged", "main"])
        if result_merged and result_merged.returncode == 0:
            merged_branches = [b.strip() for b in result_merged.stdout.strip().split("\n")
                              if b.strip() and "origin/main" not in b and "HEAD" not in b]

        return {
            "open": open_branches,
            "merged": merged_branches
        }

    # ⚡ Bolt: Full fallback to original method for older Git versions.
    result = run_command(["git", "branch", "-r", "--no-merged", "main"])
    branches = []
    if result and result.returncode == 0:
        branches = [b.strip() for b in result.stdout.strip().split("\n") 
                    if b.strip() and "origin/main" not in b and "HEAD" not in b]
    
    result_merged = run_command(["git", "branch", "-r", "--merged", "main"])
    merged_branches = []
    if result_merged and result_merged.returncode == 0:
        merged_branches = [b.strip() for b in result_merged.stdout.strip().split("\n") 
                          if b.strip() and "origin/main" not in b and "HEAD" not in b]
    
    return {
        "open": branches,
        "merged": merged_branches
    }


def analyze_branch_name(branch_name):
    """Analyze branch name to extract PR information."""
    branch = branch_name.replace("origin/", "")
    
    info = {
        "type": "unknown",
        "category": "other",
        "description": branch
    }
    
    # Categorize branches
    if branch.startswith("Cursor/"):
        info["type"] = "cursor"
        info["category"] = "ai-generated"
        info["description"] = branch.replace("Cursor/A6-9V/", "")
    elif branch.startswith("copilot/"):
        info["type"] = "copilot"
        info["category"] = "ai-generated"
        info["description"] = branch.replace("copilot/", "")
    elif branch.startswith("bolt-"):
        info["type"] = "bolt"
        info["category"] = "optimization"
        info["description"] = branch.replace("bolt-", "")
    elif branch.startswith("feat/"):
        info["type"] = "feature"
        info["category"] = "feature"
        info["description"] = branch.replace("feat/", "")
    elif branch.startswith("feature/"):
        info["type"] = "feature"
        info["category"] = "feature"
        info["description"] = branch.replace("feature/", "")
    
    return info


def get_branch_info(branch_name):
    """Get detailed information about a branch."""
    if not BRANCH_METADATA:
        populate_branch_metadata()

    short_name = branch_name.replace("origin/", "")
    meta = BRANCH_METADATA.get(short_name)

    # ⚡ Bolt: Use cached metadata if it contains ahead/behind info.
    if meta and meta.get("ahead") is not None:
        return {
            "branch": short_name,
            "full_name": meta["full_name"],
            "commit_count": meta["ahead"],
            "commits": [meta["subject"]],
            "last_commit_date": meta["date"]
        }

    # ⚡ Bolt: Fallback to manual fetching if not in cache or missing info (old Git).
    full_name = branch_name if branch_name.startswith("origin/") else f"origin/{branch_name}"
    
    # Get commit count relative to main
    result = run_command(["git", "log", "--oneline", f"main..{full_name}"])
    commit_count = 0
    commits = []
    if result and result.returncode == 0:
        commits = [c.strip() for c in result.stdout.strip().split("\n") if c.strip()]
        commit_count = len(commits)
    
    # Get last commit date
    last_commit = meta["date"] if meta else None
    if not last_commit:
        result_date = run_command(["git", "log", "-1", "--format=%ci", full_name])
        if result_date and result_date.returncode == 0:
            last_commit = result_date.stdout.strip()
    
    return {
        "branch": short_name,
        "full_name": full_name,
        "commit_count": commit_count,
        "commits": commits[:5],
        "last_commit_date": last_commit
    }


def main():
    """Main review function."""
    print("=" * 80)
    print("PULL REQUEST REVIEW")
    print(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 80)
    print()
    
    # Try GitHub CLI first
    prs = get_prs_via_gh_cli()
    
    if prs is not None:
        print(f"Found {len(prs)} pull requests via GitHub CLI")
        print()
        
        # Group by state
        by_state = defaultdict(list)
        for pr in prs:
            by_state[pr.get("state", "unknown")].append(pr)
        
        print("Pull Requests by State:")
        for state, pr_list in sorted(by_state.items()):
            print(f"  {state.upper()}: {len(pr_list)}")
        print()
        
        # Show open PRs
        open_prs = by_state.get("OPEN", [])
        if open_prs:
            print("=" * 80)
            print("OPEN PULL REQUESTS")
            print("=" * 80)
            for pr in open_prs:
                print(f"\nPR #{pr.get('number', 'N/A')}: {pr.get('title', 'No title')}")
                print(f"  Author: {pr.get('author', {}).get('login', 'Unknown')}")
                print(f"  Branch: {pr.get('headRefName', 'N/A')} -> {pr.get('baseRefName', 'main')}")
                print(f"  Created: {pr.get('createdAt', 'N/A')}")
                print(f"  Updated: {pr.get('updatedAt', 'N/A')}")
                print(f"  Draft: {'Yes' if pr.get('isDraft') else 'No'}")
                labels = [l.get('name') for l in pr.get('labels', [])]
                if labels:
                    print(f"  Labels: {', '.join(labels)}")
        
        # Show merged PRs
        merged_prs = by_state.get("MERGED", [])
        if merged_prs:
            print("\n" + "=" * 80)
            print(f"MERGED PULL REQUESTS ({len(merged_prs)} total)")
            print("=" * 80)
            print(f"\nShowing last 10 merged PRs:")
            for pr in merged_prs[-10:]:
                print(f"  PR #{pr.get('number', 'N/A')}: {pr.get('title', 'No title')}")
    
    else:
        # Fallback to git branch analysis
        print("GitHub CLI not available, analyzing branches...")
        print()
        
        branch_info = get_prs_via_git()
        
        open_branches = branch_info["open"]
        merged_branches = branch_info["merged"]
        
        print(f"Open branches (potential PRs): {len(open_branches)}")
        print(f"Merged branches (completed PRs): {len(merged_branches)}")
        print()
        
        # Categorize open branches
        categories = defaultdict(list)
        for branch in open_branches:
            info = analyze_branch_name(branch)
            categories[info["category"]].append((branch, info))
        
        print("=" * 80)
        print("OPEN BRANCHES (Potential Pull Requests)")
        print("=" * 80)
        print()
        
        for category, branches in sorted(categories.items()):
            print(f"{category.upper()}: {len(branches)} branches")
            for branch, info in branches[:10]:  # Show first 10
                branch_details = get_branch_info(branch)
                print(f"  - {info['description']}")
                print(f"    Branch: {branch_details['branch']}")
                print(f"    Commits: {branch_details['commit_count']}")
                if branch_details['last_commit_date']:
                    print(f"    Last commit: {branch_details['last_commit_date']}")
            if len(branches) > 10:
                print(f"    ... and {len(branches) - 10} more")
            print()
        
        print("=" * 80)
        print("MERGED BRANCHES (Completed Pull Requests)")
        print("=" * 80)
        print(f"\nTotal merged: {len(merged_branches)}")
        print("\nRecent merged branches:")
        # Show only first 20 to keep output clean
        for branch in merged_branches[:20]:
            info = analyze_branch_name(branch)
            print(f"  - {info['description']}")
    
    print("\n" + "=" * 80)
    print("REVIEW COMPLETE")
    print("=" * 80)
    print("\nNote: GitHub doesn't support 'pinning' pull requests directly.")
    print("Consider:")
    print("1. Creating a tracking issue for important PRs")
    print("2. Using labels to categorize PRs")
    print("3. Adding PRs to project boards")
    print("4. Creating a PR summary document")


if __name__ == "__main__":
    main()