145 lines
5.0 KiB
Python
145 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
pr_triage.py — Triage PR backlog for timmy-config.
|
|
|
|
Identifies duplicate PRs for the same issue, unassigned PRs,
|
|
and recommends which to close/merge.
|
|
|
|
Usage:
|
|
python3 scripts/pr_triage.py --repo Timmy_Foundation/timmy-config
|
|
python3 scripts/pr_triage.py --repo Timmy_Foundation/timmy-config --close-duplicates --dry-run
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import urllib.request
|
|
from collections import defaultdict
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
|
|
GITEA_URL = "https://forge.alexanderwhitestone.com"
|
|
|
|
|
|
def get_token():
|
|
return (Path.home() / ".config" / "gitea" / "token").read_text().strip()
|
|
|
|
|
|
def fetch_open_prs(repo, headers):
|
|
all_prs = []
|
|
page = 1
|
|
while True:
|
|
url = f"{GITEA_URL}/api/v1/repos/{repo}/pulls?state=open&limit=100&page={page}"
|
|
req = urllib.request.Request(url, headers=headers)
|
|
resp = urllib.request.urlopen(req, timeout=15)
|
|
data = json.loads(resp.read())
|
|
if not data:
|
|
break
|
|
all_prs.extend(data)
|
|
if len(data) < 100:
|
|
break
|
|
page += 1
|
|
return all_prs
|
|
|
|
|
|
def find_duplicate_groups(prs):
|
|
issue_prs = defaultdict(list)
|
|
for pr in prs:
|
|
text = (pr.get("body") or "") + " " + (pr.get("title") or "")
|
|
issues = set(re.findall(r"#(\d+)", text))
|
|
for iss in issues:
|
|
issue_prs[iss].append(pr)
|
|
return {k: v for k, v in issue_prs.items() if len(v) > 1}
|
|
|
|
|
|
def generate_report(repo, prs):
|
|
now = datetime.now(timezone.utc)
|
|
lines = [f"# PR Triage Report — {repo}",
|
|
f"\nGenerated: {now.strftime('%Y-%m-%d %H:%M UTC')}",
|
|
f"Total open PRs: {len(prs)}", ""]
|
|
|
|
duplicates = find_duplicate_groups(prs)
|
|
unassigned = [p for p in prs if not p.get("assignee")]
|
|
|
|
lines.append("## Duplicate PR Groups")
|
|
if duplicates:
|
|
total_dupes = sum(len(v) - 1 for v in duplicates.values())
|
|
lines.append(f"**{len(duplicates)} issues with duplicate PRs ({total_dupes} excess PRs)**")
|
|
for issue, pr_group in sorted(duplicates.items(), key=lambda x: -len(x[1])):
|
|
keep = max(pr_group, key=lambda p: p["number"])
|
|
close = [p for p in pr_group if p["number"] != keep["number"]]
|
|
lines.append(f"\n### Issue #{issue} ({len(pr_group)} PRs)")
|
|
lines.append(f"- **KEEP:** #{keep['number']} — {keep['title'][:60]}")
|
|
for p in close:
|
|
lines.append(f"- CLOSE: #{p['number']} — {p['title'][:60]}")
|
|
else:
|
|
lines.append("No duplicate PR groups found.")
|
|
lines.append("")
|
|
|
|
lines.append(f"## Unassigned PRs: {len(unassigned)}")
|
|
for p in unassigned[:10]:
|
|
lines.append(f"- #{p['number']}: {p['title'][:70]}")
|
|
if len(unassigned) > 10:
|
|
lines.append(f"- ... and {len(unassigned) - 10} more")
|
|
lines.append("")
|
|
|
|
lines.append("## Recommendations")
|
|
excess = sum(len(v) - 1 for v in duplicates.values())
|
|
lines.append(f"1. Close {excess} duplicate PRs (keep newest for each issue)")
|
|
lines.append(f"2. Assign reviewers to {len(unassigned)} unassigned PRs")
|
|
lines.append(f"3. Consider adding duplicate-PR prevention to CI")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def close_duplicate_prs(repo, prs, headers, dry_run=True):
|
|
duplicates = find_duplicate_groups(prs)
|
|
closed = 0
|
|
for issue, pr_group in duplicates.items():
|
|
keep = max(pr_group, key=lambda p: p["number"])
|
|
for pr in pr_group:
|
|
if pr["number"] == keep["number"]:
|
|
continue
|
|
if dry_run:
|
|
print(f"Would close PR #{pr['number']}: {pr['title'][:60]}")
|
|
else:
|
|
url = f"{GITEA_URL}/api/v1/repos/{repo}/pulls/{pr['number']}"
|
|
data = json.dumps({"state": "closed"}).encode()
|
|
req = urllib.request.Request(url, data=data, headers={**headers, "Content-Type": "application/json"}, method="PATCH")
|
|
try:
|
|
urllib.request.urlopen(req)
|
|
print(f"Closed PR #{pr['number']}")
|
|
closed += 1
|
|
except Exception as e:
|
|
print(f"Failed to close #{pr['number']}: {e}")
|
|
return closed
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--repo", default="Timmy_Foundation/timmy-config")
|
|
parser.add_argument("--close-duplicates", action="store_true")
|
|
parser.add_argument("--dry-run", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
token = get_token()
|
|
headers = {"Authorization": f"token {token}"}
|
|
prs = fetch_open_prs(args.repo, headers)
|
|
|
|
if args.close_duplicates:
|
|
closed = close_duplicate_prs(args.repo, prs, headers, args.dry_run)
|
|
print(f"\n{'Would close' if args.dry_run else 'Closed'} {closed} duplicate PRs")
|
|
else:
|
|
report = generate_report(args.repo, prs)
|
|
print(report)
|
|
docs_dir = Path(__file__).resolve().parent.parent / "docs"
|
|
docs_dir.mkdir(exist_ok=True)
|
|
(docs_dir / "pr-triage-report.md").write_text(report)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|