forge-helpers/forge_convert_labels.py

#!/usr/bin/python3
"""
Switch issues from using per-repo priority (and other) labels created by the Pagure
import process to using the instance-wide, 'templated' priority labels
for consistency.

Note: this script was initally based on the others in the repository, which were
mostly AI-generated using Claude (Anthropic) via Cursor IDE. All changes were done
manually.
"""

import argparse
import json
from pathlib import Path
import sys
import time
from urllib.parse import urlparse

import requests

def read_token_from_file(token_file: Path) -> str:
    """
    Read API token from a file.

    Args:
        token_file: Path to the file containing the token

    Returns:
        The token string (stripped of whitespace)

    Raises:
        FileNotFoundError: If the file doesn't exist
        ValueError: If the file is empty
    """
    content = token_file.read_text().strip()
    if not content:
        raise ValueError(f"Token file is empty: {token_file}")
    return content

def read_replacements_from_file(replacements_file: str) -> dict:
    """
    Read label replacements dict from a JSON file.

    Args:
        replacements_file: Path to the file containing the dict

    Returns:
        The dict

    Raises:
        FileNotFoundError: If the file doesn't exist
        json.decoder.JSONDecodeError: If the file is not valid JSON
    """
    with open(replacements_file, "r", encoding="utf-8") as rfh:
        return json.load(rfh)

def parse_url(url: str) -> tuple[str, str]:
    """
    Parse a Forge project URL into API base URL and owner/repo path.

    Args:
        url: Full URL to the Forge project
                   (e.g., 'https://forge.fedoraproject.org/quality/some-repo')

    Returns:
        Tuple of (api_base, owner_repo)
        e.g., ('https://forge.fedoraproject.org/api/v1', 'quality/some-repo')

    Raises:
        ValueError: If the URL is invalid
    """
    parsed = urlparse(url)
    if not parsed.scheme or not parsed.netloc:
        raise ValueError(f"Invalid Forge URL: {url}")

    api_base = f"{parsed.scheme}://{parsed.netloc}/api/v1"
    # Remove leading slash from path
    owner_repo = parsed.path.lstrip("/")

    if not owner_repo or "/" not in owner_repo:
        raise ValueError(f"Invalid Forge project path (expected owner/repo): {url}")

    return api_base, owner_repo

def process_issues(
    api_base: str,
    owner_repo: str,
    token: str,
    dry_run: bool,
    delay: float,
    replacements: dict,
) -> tuple[int, int, int]:
    """
    Process issues and swap migrated labels to templated ones.

    Args:
        api_base: Forge API base URL
        owner_repo: Forge owner/repo path
        token: Forge API token
        dry_run: If True, don't actually add labels
        delay: Delay between issues in seconds

    Returns:
        Failed count
    """
    failed = 0
    i = 0
    while True:
        # this is sloppy, it breaks if any issue/PR has been deleted,
        # but it's easier than dealing with pagination...
        i += 1
        url = f"{api_base}/repos/{owner_repo}/issues/{i}/labels"
        headers = {
            "Authorization": f"token {token}",
            "Content-Type": "application/json",
        }
        response = requests.get(url, headers=headers, timeout=30)
        if response.ok:
            orig_labels = response.json()
        elif response.status_code == 404:
            # ran out of issues
            break
        else:
            print(f"Unexpected response for issue {i}: {response.status_code}")
            failed += 1
            continue
        orig_labels = requests.get(url, headers=headers, timeout=30).json()
        new_labels = []
        changed = 0
        for label in orig_labels:
            lname = label["name"].lower()
            if lname in replacements and replacements[lname]["color"] == label["color"]:
                new_labels.append(replacements[lname]["id"])
                changed = 1
                if dry_run:
                    print(f"Issue {i}: Would replace {label['name']} {label['id']} with {replacements[lname]['id']}")
            else:
                new_labels.append(label["id"])
        if changed and not dry_run:
            data = {"labels": new_labels}
            print(f"Updating labels on issue {i}...")
            response = requests.put(url, headers=headers, json=data, timeout=30)
            if not response.ok:
                print(f"Unexpected response for issue {i}: {response.status_code}")
                failed += 1
        time.sleep(delay)
    return failed

def main():
    parser = argparse.ArgumentParser(
        description="Port priority labels from per-repo to per-org",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Dry run on all issues (token from file)
  %(prog)s https://forge.fedoraproject.org/quality/some-repo --token-file ~/.forge-token --dry-run

  # Process specific issues
  %(prog)s https://forge.fedoraproject.org/quality/some-repo --token YOUR_TOKEN --issues 1 2 3

Configuration:
  The default mapping is in labels-default.json. For each issue in the repo, any label with
  a name and color that matches an entry in the mapping will be replaced with the label that
  has the id specified in the same mapping
        """,
    )

    parser.add_argument(
        "url",
        help="Full URL to the Forge project (e.g., 'https://forge.fedoraproject.org/quality/some-repo')",
    )

    token_group = parser.add_mutually_exclusive_group(required=True)
    token_group.add_argument(
        "--token",
        help="Forge API token for authentication",
    )
    token_group.add_argument(
        "--token-file",
        type=Path,
        help="Path to file containing the Forge API token",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Preview changes without actually adding labels",
    )
    parser.add_argument(
        "--delay",
        type=float,
        default=0.5,
        help="Delay between API calls in seconds (default: 0.5)",
    )
    parser.add_argument(
        "--replacements-file",
        type=str,
        default="labels-default.json",
        help="File containing map of labels to replace",
    )

    args = parser.parse_args()

    # Parse the Forge URL
    try:
        api_base, owner_repo = parse_url(args.url)
    except ValueError as e:
        print(f"Error: {e}")
        sys.exit(1)

    # Get the Forge API token
    if args.token:
        token = args.token
    else:
        try:
            token = read_token_from_file(args.token_file)
        except FileNotFoundError:
            print(f"Error: Token file not found: {args.token_file}")
            sys.exit(1)
        except ValueError as e:
            print(f"Error: {e}")
            sys.exit(1)

    try:
        replacements = read_replacements_from_file(args.replacements_file)
    except FileNotFoundError:
        sys.exit(f"Error: Replacements file not found: {args.replacements_file}")
    except json.decoder.JSONDecodeError:
        sys.exit(f"Error: Replacements file not valid JSON: {args.replacements_file}")

    print(f"Forge URL: {args.url}")
    if args.dry_run:
        print("Mode: DRY RUN (no changes will be made)")
    print()

    try:
        failed = process_issues(
            api_base=api_base,
            owner_repo=owner_repo,
            token=token,
            dry_run=args.dry_run,
            delay=args.delay,
            replacements=replacements,
        )
    except requests.exceptions.HTTPError as e:
        print(f"\nAPI Error: {e}")
        sys.exit(1)
    except KeyboardInterrupt:
        print("\n\nInterrupted by user")
        sys.exit(130)

    print(f"\n{'=' * 40}")
    print("Summary:")
    print(f"  Failed: {failed}")

    if failed > 0:
        sys.exit(1)


if __name__ == "__main__":
    main()