timmy-config/scripts/pr-backlog-triage.py

#!/usr/bin/env python3
"""
pr-backlog-triage.py — Analyze and triage open PR backlog.

Identifies duplicate PRs (same issue number), stale PRs (old with no activity),
and generates a triage report.

Usage:
    python3 scripts/pr-backlog-triage.py --report      # Print report
    python3 scripts/pr-backlog-triage.py --close-dupes # Close duplicate PRs (keep newest)
    python3 scripts/pr-backlog-triage.py --dry-run     # Show what would be closed
"""

import argparse
import json
import re
import sys
import urllib.request
from collections import defaultdict
from datetime import datetime, timezone

GITEA_URL = "https://forge.alexanderwhitestone.com"
TOKEN_PATH = "/Users/apayne/.config/gitea/token"
REPO = "Timmy_Foundation/timmy-config"


def load_token():
    with open(TOKEN_PATH) as f:
        return f.read().strip()


def api_get(path, token):
    req = urllib.request.Request(
        f"{GITEA_URL}/api/v1/repos/{REPO}{path}",
        headers={"Authorization": f"token {token}"}
    )
    return json.loads(urllib.request.urlopen(req, timeout=30).read())


def api_patch(path, token, data):
    req = urllib.request.Request(
        f"{GITEA_URL}/api/v1/repos/{REPO}{path}",
        data=json.dumps(data).encode(),
        headers={"Authorization": f"token {token}", "Content-Type": "application/json"},
        method="PATCH"
    )
    return json.loads(urllib.request.urlopen(req, timeout=15).read())


def api_post(path, token, data):
    req = urllib.request.Request(
        f"{GITEA_URL}/api/v1/repos/{REPO}{path}",
        data=json.dumps(data).encode(),
        headers={"Authorization": f"token {token}", "Content-Type": "application/json"},
        method="POST"
    )
    return json.loads(urllib.request.urlopen(req, timeout=15).read())


def extract_issue_refs(title, body):
    """Extract issue numbers referenced in title or body."""
    text = f"{title} {body or ''}"
    # Match #123 or (fixes #123) or (closes #123)
    refs = set(int(m) for m in re.findall(r'#(\d{2,5})', text))
    return refs


def main():
    parser = argparse.ArgumentParser(description="Triage open PR backlog")
    parser.add_argument("--report", action="store_true", help="Print triage report")
    parser.add_argument("--close-dupes", action="store_true", help="Close duplicate PRs (keep newest)")
    parser.add_argument("--dry-run", action="store_true", help="Show what would be closed")
    args = parser.parse_args()

    if not args.report and not args.close_dupes:
        args.report = True

    token = load_token()
    prs = api_get("/pulls?state=open&limit=100", token)

    print(f"Found {len(prs)} open PRs\n")

    # Build issue → PR mapping
    issue_to_prs = defaultdict(list)
    for pr in prs:
        refs = extract_issue_refs(pr["title"], pr.get("body", ""))
        for ref in refs:
            issue_to_prs[ref].append(pr)

    # Find duplicates (same issue referenced by multiple PRs)
    duplicates = {}
    for issue_num, pr_list in issue_to_prs.items():
        if len(pr_list) > 1:
            # Sort by number (newest first)
            sorted_prs = sorted(pr_list, key=lambda p: -p["number"])
            duplicates[issue_num] = sorted_prs

    if args.report:
        print(f"{'='*60}")
        print(f"DUPLICATE PRs ({len(duplicates)} issues with multiple PRs)")
        print(f"{'='*60}")
        for issue_num, pr_list in sorted(duplicates.items()):
            print(f"\nIssue #{issue_num}: {len(pr_list)} PRs")
            for i, pr in enumerate(pr_list):
                marker = "KEEP" if i == 0 else "CLOSE"
                print(f"  [{marker}] PR #{pr['number']}: {pr['title'][:70]}")
                print(f"       branch={pr['head']['ref']} created={pr['created_at'][:10]}")

        total_dupes = sum(len(v) - 1 for v in duplicates.values())
        print(f"\nTotal duplicate PRs that could be closed: {total_dupes}")

        # Check for PRs referencing closed issues
        print(f"\n{'='*60}")
        print("PRs referencing CLOSED issues:")
        print(f"{'='*60}")
        closed_issue_prs = []
        for issue_num in issue_to_prs:
            try:
                issue = api_get(f"/../../issues/{issue_num}", token)
                if issue.get("state") == "closed":
                    for pr in issue_to_prs[issue_num]:
                        closed_issue_prs.append((issue_num, pr))
            except Exception:
                pass

        for issue_num, pr in sorted(closed_issue_prs, key=lambda x: -x[1]["number"]):
            print(f"  PR #{pr['number']}: {pr['title'][:70]} (issue #{issue_num} is CLOSED)")

    if args.close_dupes:
        closed = 0
        for issue_num, pr_list in duplicates.items():
            # Keep the newest (first in list), close the rest
            keep = pr_list[0]
            close_list = pr_list[1:]

            for pr in close_list:
                if args.dry_run:
                    print(f"DRY RUN: Would close PR #{pr['number']} (duplicate of #{keep['number']} for issue #{issue_num})")
                else:
                    # Add comment
                    try:
                        api_post(f"/issues/{pr['number']}/comments", token, {
                            "body": f"Closing as duplicate. PR #{keep['number']} is newer and addresses the same issue (#{issue_num})."
                        })
                    except Exception:
                        pass

                    # Close the PR
                    try:
                        api_patch(f"/pulls/{pr['number']}", token, {"state": "closed"})
                        print(f"Closed PR #{pr['number']} (duplicate of #{keep['number']})")
                        closed += 1
                    except Exception as e:
                        print(f"Error closing PR #{pr['number']}: {e}")

        print(f"\nClosed {closed} duplicate PRs")


if __name__ == "__main__":
    main()