the-nexus/scripts/check_existing_prs.py

#!/usr/bin/env python3
"""
Check if PRs already exist for an issue before creating a new one.

This script prevents duplicate PRs by checking if there are already
open PRs for a given issue.

Usage:
    python3 scripts/check_existing_prs.py <issue_number>

Exit codes:
    0 - No existing PRs found (safe to create new PR)
    1 - Existing PRs found (do not create new PR)
    2 - Error (API failure, missing parameters, etc.)

Designed for issue #1474: Prevent duplicate PRs
"""

import json
import os
import sys
import urllib.request
import urllib.error
from datetime import datetime


def check_existing_prs(issue_number: int, repo: str = None, token: str = None) -> int:
    """
    Check if PRs already exist for an issue.
    
    Args:
        issue_number: The issue number to check
        repo: Repository in format "owner/repo" (default: from env or "Timmy_Foundation/the-nexus")
        token: Gitea API token (default: from GITEA_TOKEN env var)
    
    Returns:
        0: No existing PRs found (safe to create new PR)
        1: Existing PRs found (do not create new PR)
        2: Error (API failure, missing parameters, etc.)
    """
    # Get configuration from environment
    gitea_url = os.environ.get('GITEA_URL', 'https://forge.alexanderwhitestone.com')
    token = token or os.environ.get('GITEA_TOKEN')
    repo = repo or os.environ.get('REPO', 'Timmy_Foundation/the-nexus')
    
    if not token:
        print("ERROR: GITEA_TOKEN environment variable not set", file=sys.stderr)
        return 2
    
    # Validate issue number
    if not isinstance(issue_number, int) or issue_number <= 0:
        print("ERROR: Issue number must be a positive integer", file=sys.stderr)
        return 2
    
    # Build API URL
    api_url = f"{gitea_url}/api/v1/repos/{repo}/pulls?state=open&limit=100"
    
    # Make API request
    try:
        req = urllib.request.Request(api_url, headers={
            'Authorization': f'token {token}',
            'Content-Type': 'application/json'
        })
        
        with urllib.request.urlopen(req, timeout=30) as resp:
            prs = json.loads(resp.read())
    
    except urllib.error.URLError as e:
        print(f"ERROR: Failed to fetch PRs: {e}", file=sys.stderr)
        return 2
    except json.JSONDecodeError as e:
        print(f"ERROR: Failed to parse API response: {e}", file=sys.stderr)
        return 2
    except Exception as e:
        print(f"ERROR: Unexpected error: {e}", file=sys.stderr)
        return 2
    
    # Check for PRs referencing this issue
    issue_ref = f"#{issue_number}"
    matching_prs = []
    
    for pr in prs:
        title = pr.get('title', '')
        body = pr.get('body', '') or ''
        
        # Check if issue is referenced in title or body
        if issue_ref in title or issue_ref in body:
            matching_prs.append(pr)
    
    # Report results
    timestamp = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
    
    if not matching_prs:
        print(f"[{timestamp}] ✅ No existing PRs found for issue #{issue_number}")
        print("Safe to create a new PR")
        return 0
    
    # Found existing PRs
    print(f"[{timestamp}] ⚠️  Found existing PRs for issue #{issue_number}:")
    print()
    
    for pr in matching_prs:
        pr_number = pr.get('number')
        pr_title = pr.get('title')
        pr_branch = pr.get('head', {}).get('ref', 'unknown')
        pr_created = pr.get('created_at', 'unknown')
        pr_url = pr.get('html_url', 'unknown')
        
        print(f"  PR #{pr_number}: {pr_title}")
        print(f"    Branch: {pr_branch}")
        print(f"    Created: {pr_created}")
        print(f"    URL: {pr_url}")
        print()
    
    print("❌ Do not create a new PR. Review existing PRs first.")
    print()
    print("Options:")
    print("  1. Review and merge an existing PR")
    print("  2. Close duplicates and keep the best one")
    print("  3. Add comments to existing PRs instead of creating new ones")
    print()
    print("To see details of existing PRs:")
    print(f'  curl -H "Authorization: token $GITEA_TOKEN" "{gitea_url}/api/v1/repos/{repo}/pulls?state=open" | jq \'.[] | select(.title | test("#{issue_number}"; "i"))\'')
    
    return 1


def main():
    """Main entry point."""
    if len(sys.argv) < 2:
        print("Usage: python3 check_existing_prs.py <issue_number>", file=sys.stderr)
        print("       python3 check_existing_prs.py <issue_number> [repo] [token]", file=sys.stderr)
        return 2
    
    try:
        issue_number = int(sys.argv[1])
    except ValueError:
        print("ERROR: Issue number must be an integer", file=sys.stderr)
        return 2
    
    repo = sys.argv[2] if len(sys.argv) > 2 else None
    token = sys.argv[3] if len(sys.argv) > 3 else None
    
    return check_existing_prs(issue_number, repo, token)


if __name__ == '__main__':
    sys.exit(main())