Some checks failed
Smoke Test / smoke (pull_request) Failing after 26s
Automates the recommendation from #662: bulk close issues whose acceptance criteria are met and PRs are merged. Features: - Scans open issues for linked merged PRs - Skips issues with assignees or active labels (epic, in-progress) - Dry-run by default, --close to execute - Adds closing comment for audit trail Usage: python scripts/backlog_cleanup.py --dry-run python scripts/backlog_cleanup.py --close --limit 50 Refs #662
111 lines
3.5 KiB
Python
Executable File
111 lines
3.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Backlog Cleanup — Bulk close issues whose PRs are merged.
|
|
|
|
Usage:
|
|
python backlog_cleanup.py --repo Timmy_Foundation/timmy-home --dry-run
|
|
python backlog_cleanup.py --repo Timmy_Foundation/timmy-home --close
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import argparse
|
|
import urllib.request
|
|
import urllib.error
|
|
import time
|
|
from pathlib import Path
|
|
|
|
|
|
def get_token():
|
|
f = Path.home() / ".config" / "gitea" / "token"
|
|
if f.exists():
|
|
return f.read_text().strip()
|
|
return os.environ.get("GITEA_TOKEN", "")
|
|
|
|
|
|
def api(base, token, path, method="GET", data=None):
|
|
url = f"{base}/api/v1{path}"
|
|
headers = {"Authorization": f"token {token}"}
|
|
body = json.dumps(data).encode() if data else None
|
|
if data:
|
|
headers["Content-Type"] = "application/json"
|
|
req = urllib.request.Request(url, data=body, headers=headers, method=method)
|
|
try:
|
|
return json.loads(urllib.request.urlopen(req, timeout=15).read())
|
|
except Exception as e:
|
|
print(f" API error: {e}", file=sys.stderr)
|
|
return None
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument("--repo", default="Timmy_Foundation/timmy-home")
|
|
p.add_argument("--base", default="https://forge.alexanderwhitestone.com")
|
|
p.add_argument("--dry-run", action="store_true", default=True)
|
|
p.add_argument("--close", action="store_true")
|
|
p.add_argument("--limit", type=int, default=20)
|
|
args = p.parse_args()
|
|
if args.close:
|
|
args.dry_run = False
|
|
|
|
token = get_token()
|
|
issues = api(args.base, token, f"/repos/{args.repo}/issues?state=open&limit={args.limit}")
|
|
if not issues:
|
|
return 1
|
|
|
|
issues = [i for i in issues if not i.get("pull_request")]
|
|
print(f"Scanning {len(issues)} issues...")
|
|
|
|
closable = []
|
|
for issue in issues:
|
|
if issue.get("assignees"):
|
|
continue
|
|
labels = {l.get("name", "").lower() for l in issue.get("labels", [])}
|
|
if labels & {"epic", "in-progress", "claw-code-in-progress", "blocked"}:
|
|
continue
|
|
|
|
# Check for merged PRs referencing this issue
|
|
ref = f"#{issue['number']}"
|
|
prs = api(args.base, token, f"/repos/{args.repo}/pulls?state=all&limit=20")
|
|
time.sleep(0.1) # Rate limit
|
|
|
|
linked_merged = [
|
|
pr for pr in (prs or [])
|
|
if ref in (pr.get("body", "") + pr.get("title", ""))
|
|
and (pr.get("state") == "merged" or pr.get("merged"))
|
|
]
|
|
|
|
if linked_merged:
|
|
reason = f"merged PR #{linked_merged[0]['number']}"
|
|
closable.append((issue, reason))
|
|
tag = "WOULD CLOSE" if args.dry_run else "CLOSING"
|
|
print(f" {tag} #{issue['number']}: {issue['title'][:50]} — {reason}")
|
|
|
|
if not closable:
|
|
print("No issues to close.")
|
|
return 0
|
|
|
|
print(f"\n{'Would close' if args.dry_run else 'Closing'} {len(closable)} issues")
|
|
if args.dry_run:
|
|
print("(use --close to execute)")
|
|
return 0
|
|
|
|
closed = 0
|
|
for issue, reason in closable:
|
|
api(args.base, token, f"/repos/{args.repo}/issues/{issue['number']}/comments",
|
|
method="POST", data={"body": f"Closing — {reason}.\nAutomated by backlog_cleanup.py"})
|
|
r = api(args.base, token, f"/repos/{args.repo}/issues/{issue['number']}",
|
|
method="POST", data={"state": "closed"})
|
|
if r:
|
|
closed += 1
|
|
print(f" Closed #{issue['number']}")
|
|
time.sleep(0.2)
|
|
|
|
print(f"\nClosed {closed}/{len(closable)}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|