Compare commits
1 Commits
fix/1543
...
fix/issue-
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d8173c4bb1 |
95
scripts/check_duplicate_pr.py
Normal file
95
scripts/check_duplicate_pr.py
Normal file
@@ -0,0 +1,95 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
check_duplicate_pr.py — Pre-flight check before creating a PR.
|
||||
|
||||
Checks if there's already an open PR for this issue on any branch.
|
||||
Prevents the duplicate PR problem described in issue #1460.
|
||||
|
||||
Usage:
|
||||
python3 scripts/check_duplicate_pr.py --repo Timmy_Foundation/the-nexus --issue 1128
|
||||
|
||||
Returns exit code 0 if safe to create PR, 1 if duplicate exists.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
GITEA_URL = "https://forge.alexanderwhitestone.com"
|
||||
|
||||
|
||||
def get_token():
|
||||
token_path = Path.home() / ".config" / "gitea" / "token"
|
||||
return token_path.read_text().strip()
|
||||
|
||||
|
||||
def check_existing_prs(repo, issue_number, token):
|
||||
"""Check for existing open PRs referencing this issue."""
|
||||
headers = {"Authorization": f"token {token}"}
|
||||
|
||||
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)
|
||||
data = json.loads(resp.read())
|
||||
if not data:
|
||||
break
|
||||
all_prs.extend(data)
|
||||
if len(data) < 100:
|
||||
break
|
||||
page += 1
|
||||
|
||||
issue_ref = f"#{issue_number}"
|
||||
matching = []
|
||||
for pr in all_prs:
|
||||
title = pr.get("title", "")
|
||||
body = pr.get("body", "")
|
||||
branch = pr.get("head", {}).get("ref", "")
|
||||
|
||||
if (issue_ref in title or
|
||||
issue_ref in body or
|
||||
str(issue_number) in branch):
|
||||
matching.append(pr)
|
||||
|
||||
return matching
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Check for duplicate PRs before creating")
|
||||
parser.add_argument("--repo", required=True, help="Repo (e.g., Timmy_Foundation/the-nexus)")
|
||||
parser.add_argument("--issue", required=True, type=int, help="Issue number")
|
||||
parser.add_argument("--branch", default="", help="Branch name (for display)")
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
token = get_token()
|
||||
except FileNotFoundError:
|
||||
print("ERROR: Gitea token not found at ~/.config/gitea/token")
|
||||
sys.exit(2)
|
||||
|
||||
existing = check_existing_prs(args.repo, args.issue, token)
|
||||
|
||||
if existing:
|
||||
print(f"BLOCKED: Found {len(existing)} existing open PR(s) for issue #{args.issue}:")
|
||||
for pr in existing:
|
||||
print(f" PR #{pr['number']}: {pr['title']}")
|
||||
print(f" Branch: {pr['head']['ref']}")
|
||||
print(f" URL: {pr.get('html_url', 'N/A')}")
|
||||
print(f"\nDo NOT create another PR. Use the existing one or close it first.")
|
||||
print(f"If you need to update, push to the existing branch.")
|
||||
sys.exit(1)
|
||||
else:
|
||||
print(f"OK: No existing open PRs for issue #{args.repo}#{args.issue}")
|
||||
if args.branch:
|
||||
print(f"Safe to create PR from branch: {args.branch}")
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
106
scripts/cleanup_duplicate_prs.py
Normal file
106
scripts/cleanup_duplicate_prs.py
Normal file
@@ -0,0 +1,106 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
cleanup_duplicate_prs.py — Close duplicate PRs for an issue.
|
||||
|
||||
Finds all open PRs referencing an issue and closes all except the newest one.
|
||||
|
||||
Usage:
|
||||
python3 scripts/cleanup_duplicate_prs.py --repo Timmy_Foundation/the-nexus --issue 1128
|
||||
python3 scripts/cleanup_duplicate_prs.py --repo Timmy_Foundation/the-nexus --issue 1128 --dry-run
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
GITEA_URL = "https://forge.alexanderwhitestone.com"
|
||||
|
||||
|
||||
def get_token():
|
||||
token_path = Path.home() / ".config" / "gitea" / "token"
|
||||
return token_path.read_text().strip()
|
||||
|
||||
|
||||
def find_duplicate_prs(repo, issue_number, token):
|
||||
headers = {"Authorization": f"token {token}"}
|
||||
|
||||
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)
|
||||
data = json.loads(resp.read())
|
||||
if not data:
|
||||
break
|
||||
all_prs.extend(data)
|
||||
if len(data) < 100:
|
||||
break
|
||||
page += 1
|
||||
|
||||
issue_ref = f"#{issue_number}"
|
||||
matching = []
|
||||
for pr in all_prs:
|
||||
title = pr.get("title", "")
|
||||
body = pr.get("body", "")
|
||||
branch = pr.get("head", {}).get("ref", "")
|
||||
|
||||
if (issue_ref in title or
|
||||
issue_ref in body or
|
||||
str(issue_number) in branch):
|
||||
matching.append(pr)
|
||||
|
||||
# Sort by PR number (newest last)
|
||||
matching.sort(key=lambda p: p["number"])
|
||||
return matching
|
||||
|
||||
|
||||
def close_pr(repo, pr_number, token):
|
||||
headers = {"Authorization": f"token {token}", "Content-Type": "application/json"}
|
||||
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, method="PATCH")
|
||||
resp = urllib.request.urlopen(req)
|
||||
return json.loads(resp.read())
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Close duplicate PRs for an issue")
|
||||
parser.add_argument("--repo", required=True)
|
||||
parser.add_argument("--issue", required=True, type=int)
|
||||
parser.add_argument("--dry-run", action="store_true")
|
||||
args = parser.parse_args()
|
||||
|
||||
token = get_token()
|
||||
duplicates = find_duplicate_prs(args.repo, args.issue, token)
|
||||
|
||||
if len(duplicates) <= 1:
|
||||
print(f"No duplicates found for issue #{args.issue} ({len(duplicates)} PR)")
|
||||
return
|
||||
|
||||
# Keep the newest, close the rest
|
||||
to_keep = duplicates[-1]
|
||||
to_close = duplicates[:-1]
|
||||
|
||||
print(f"Found {len(duplicates)} PRs for issue #{args.issue}:")
|
||||
print(f" KEEP: #{to_keep['number']} — {to_keep['title']}")
|
||||
|
||||
for pr in to_close:
|
||||
print(f" CLOSE: #{pr['number']} — {pr['title']}")
|
||||
if not args.dry_run:
|
||||
try:
|
||||
close_pr(args.repo, pr["number"], token)
|
||||
print(f" Closed.")
|
||||
except Exception as e:
|
||||
print(f" Failed: {e}")
|
||||
|
||||
if args.dry_run:
|
||||
print(f"\nDry run — no changes made. Run without --dry-run to close {len(to_close)} PRs.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user