feat: PR backlog triage script + 9 duplicate PRs closed (#658)

scripts/pr-backlog-triage.py

@@ -0,0 +1,160 @@
+#!/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()