the-nexus/scripts/pr-backlog-analyzer.py

#!/usr/bin/env python3
"""
PR Backlog Analyzer for timmy-config

Analyzes open PRs and provides recommendations for cleanup.
"""

import json
import subprocess
import sys
from datetime import datetime, timedelta
from pathlib import Path


def get_open_prs(repo: str, token: str) -> list:
    """Get all open PRs from a repository."""
    result = subprocess.run([
        "curl", "-s", "-H", f"Authorization: token {token}",
        f"https://forge.alexanderwhitestone.com/api/v1/repos/{repo}/pulls?state=open&limit=100"
    ], capture_output=True, text=True)
    
    if result.returncode != 0:
        print(f"Error fetching PRs: {result.stderr}")
        return []
    
    return json.loads(result.stdout)


def analyze_pr(pr: dict) -> dict:
    """Analyze a single PR."""
    created = datetime.fromisoformat(pr['created_at'].replace('Z', '+00:00'))
    age_days = (datetime.now(created.tzinfo) - created).days
    
    # Check for reviews
    reviews = pr.get('reviews', [])
    has_approvals = any(r.get('state') == 'APPROVED' for r in reviews)
    has_changes_requested = any(r.get('state') == 'CHANGES_REQUESTED' for r in reviews)
    
    # Check labels
    labels = [l['name'] for l in pr.get('labels', [])]
    
    return {
        'number': pr['number'],
        'title': pr['title'],
        'branch': pr['head']['ref'],
        'created': pr['created_at'],
        'age_days': age_days,
        'user': pr['user']['login'],
        'has_approvals': has_approvals,
        'has_changes_requested': has_changes_requested,
        'labels': labels,
        'url': pr['html_url'],
    }


def categorize_prs(prs: list) -> dict:
    """Categorize PRs by status."""
    categories = {
        'stale': [],  # > 30 days old
        'needs_review': [],  # No reviews
        'approved': [],  # Approved but not merged
        'changes_requested': [],  # Changes requested
        'recent': [],  # < 7 days old
    }
    
    for pr in prs:
        if pr['age_days'] > 30:
            categories['stale'].append(pr)
        elif pr['has_approvals']:
            categories['approved'].append(pr)
        elif pr['has_changes_requested']:
            categories['changes_requested'].append(pr)
        elif pr['age_days'] < 7:
            categories['recent'].append(pr)
        else:
            categories['needs_review'].append(pr)
    
    return categories


def generate_report(repo: str, prs: list, categories: dict) -> str:
    """Generate a markdown report."""
    report = f"""# PR Backlog Report — {repo}

Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

## Summary

- **Total Open PRs**: {len(prs)}
- **Stale (>30 days)**: {len(categories['stale'])}
- **Needs Review**: {len(categories['needs_review'])}
- **Approved**: {len(categories['approved'])}
- **Changes Requested**: {len(categories['changes_requested'])}
- **Recent (<7 days)**: {len(categories['recent'])}

## Recommendations

### Immediate Actions
1. **Merge approved PRs**: {len(categories['approved'])} PRs are ready to merge
2. **Review stale PRs**: {len(categories['stale'])} PRs are >30 days old
3. **Address changes requested**: {len(categories['changes_requested'])} PRs need updates

### Process Improvements
1. **Assign reviewers**: Ensure each PR has a reviewer within 24 hours
2. **Set SLAs**: 
   - Review within 48 hours
   - Merge within 7 days of approval
   - Close stale PRs after 30 days
3. **Automate**: Add CI checks to prevent backlog

## Detailed Analysis

### Stale PRs (>30 days)
"""
    
    for pr in categories['stale']:
        report += f"- **#{pr['number']}**: {pr['title']}\n"
        report += f"  - Age: {pr['age_days']} days\n"
        report += f"  - Author: {pr['user']}\n"
        report += f"  - URL: {pr['url']}\n\n"
    
    report += "\n### Approved PRs (Ready to Merge)\n"
    for pr in categories['approved']:
        report += f"- **#{pr['number']}**: {pr['title']}\n"
        report += f"  - Age: {pr['age_days']} days\n"
        report += f"  - Author: {pr['user']}\n"
        report += f"  - URL: {pr['url']}\n\n"
    
    report += "\n### Needs Review\n"
    for pr in categories['needs_review']:
        report += f"- **#{pr['number']}**: {pr['title']}\n"
        report += f"  - Age: {pr['age_days']} days\n"
        report += f"  - Author: {pr['user']}\n"
        report += f"  - URL: {pr['url']}\n\n"
    
    return report


def main():
    """Main function."""
    token = Path.home() / '.config' / 'gitea' / 'token'
    if not token.exists():
        print("Error: Gitea token not found")
        sys.exit(1)
    
    token_str = token.read_text().strip()
    repo = "Timmy_Foundation/timmy-config"
    
    print(f"Fetching PRs for {repo}...")
    prs = get_open_prs(repo, token_str)
    
    if not prs:
        print("No open PRs found")
        return
    
    print(f"Found {len(prs)} open PRs")
    
    # Analyze PRs
    analyzed = [analyze_pr(pr) for pr in prs]
    categories = categorize_prs(analyzed)
    
    # Generate report
    report = generate_report(repo, analyzed, categories)
    
    # Save report
    output_dir = Path("reports")
    output_dir.mkdir(exist_ok=True)
    
    report_file = output_dir / f"pr-backlog-{datetime.now().strftime('%Y%m%d')}.md"
    report_file.write_text(report)
    
    print(f"\nReport saved to: {report_file}")
    print(f"\nSummary:")
    print(f"  Total PRs: {len(prs)}")
    print(f"  Stale: {len(categories['stale'])}")
    print(f"  Approved: {len(categories['approved'])}")
    print(f"  Needs Review: {len(categories['needs_review'])}")


if __name__ == "__main__":
    main()