forge-helpers/pagure_add_migration_comment.py

#!/usr/bin/env python3
"""
Add migration notice comments to Pagure tickets, linking to their new Forge locations.

This script adds a comment to old Pagure tickets notifying users that the ticket
has been migrated to Fedora Forge, with a hyperlink to the new location.

Note: Most of the code is AI-generated using Claude (Anthropic) via Cursor IDE.
"""

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

import requests

DEFAULT_COMMENT_TEMPLATE = (
    "This issue has been migrated to Fedora Forge:\n{forge_url}\n\n"
    "Please continue any further discussion there."
)
MIGRATION_MARKER = "This issue has been migrated to Fedora Forge:"

PAGURE_STATUS = {"open": "Open", "closed": "Closed", "all": "all"}

# Ensure the marker is present in the default template to detect already-migrated issues
assert MIGRATION_MARKER in DEFAULT_COMMENT_TEMPLATE, (
    "MIGRATION_MARKER must be contained in DEFAULT_COMMENT_TEMPLATE"
)


def parse_pagure_url(pagure_url: str) -> tuple[str, str]:
    """
    Parse a Pagure project URL into API base URL and project path.

    Args:
        pagure_url: Full URL to the Pagure project
                    (e.g., 'https://pagure.io/fedora-qa/some-repo' or
                     'https://stg.pagure.io/fedora-qa/some-repo')

    Returns:
        Tuple of (api_base, project_path)
        e.g., ('https://pagure.io/api/0', 'fedora-qa/some-repo')

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

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

    if not project_path:
        raise ValueError(f"No project path found in URL: {pagure_url}")

    return api_base, project_path


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 get_pagure_issues(
    api_base: str, project: str, api_token: str, status: str = "all", delay: float = 1.0
) -> list[dict]:
    """
    Fetch issues from a Pagure project.

    Args:
        api_base: Pagure API base URL (e.g., 'https://pagure.io/api/0')
        project: Pagure project path (e.g., 'fedora-qa/some-repo')
        api_token: Pagure API token for authentication
        status: Issue status filter ('all', 'Open', 'Closed')
        delay: Delay between API calls in seconds (for pagination)

    Returns:
        List of issue dictionaries
    """
    url = f"{api_base}/{project}/issues"
    headers = {"Authorization": f"token {api_token}"}
    params = {"status": status, "per_page": 100}

    all_issues = []
    page = 1

    while True:
        params["page"] = page
        response = requests.get(url, headers=headers, params=params, timeout=30)
        response.raise_for_status()
        data = response.json()

        issues = data.get("issues", [])
        if not issues:
            break

        all_issues.extend(issues)
        page += 1

        # Check if we've fetched all pages
        if len(issues) < params["per_page"]:
            break

        # Delay between page fetches to avoid rate limiting
        if delay > 0:
            time.sleep(delay)

    return all_issues


def get_issue(
    api_base: str, project: str, issue_id: int, api_token: str
) -> dict:
    """
    Fetch a single issue from Pagure.

    Args:
        api_base: Pagure API base URL
        project: Pagure project path
        issue_id: Issue number
        api_token: Pagure API token

    Returns:
        Issue dictionary

    Raises:
        requests.exceptions.HTTPError: If the API call fails
    """
    url = f"{api_base}/{project}/issue/{issue_id}"
    headers = {"Authorization": f"token {api_token}"}

    response = requests.get(url, headers=headers, timeout=30)
    response.raise_for_status()
    return response.json()


def has_migration_comment(comments: list[dict]) -> bool:
    """
    Check if any comment already contains the migration notice.

    Args:
        comments: List of comment dictionaries

    Returns:
        True if migration comment exists, False otherwise
    """
    for comment in comments:
        if MIGRATION_MARKER in comment.get("comment", ""):
            return True
    return False


def update_issue(
    api_base: str,
    project: str,
    issue_id: int,
    api_token: str,
    *,
    comment: str | None = None,
    close: bool = False,
    issue_status: str | None = None,
) -> bool:
    """
    Update a Pagure issue by adding a comment and/or closing it.

    Args:
        api_base: Pagure API base URL
        project: Pagure project path
        issue_id: Issue number
        api_token: Pagure API token
        comment: Comment text to add, or None to skip
        close: If True, close the issue when it is Open
        issue_status: Current issue status ('Open' or 'Closed')

    Returns:
        True if all requested operations succeed, False otherwise
    """
    headers = {"Authorization": f"token {api_token}"}

    if comment is not None:
        url = f"{api_base}/{project}/issue/{issue_id}/comment"
        response = requests.post(url, headers=headers, data={"comment": comment}, timeout=30)
        if response.status_code != 200:
            print(f"  Error adding comment: {response.status_code} - {response.text}")
            return False

    if close and issue_status == "Open":
        url = f"{api_base}/{project}/issue/{issue_id}/status"
        response = requests.post(url, headers=headers, data={"status": "Closed"}, timeout=30)
        if response.status_code != 200:
            print(f"  Error closing issue: {response.status_code} - {response.text}")
            return False

    return True


def construct_forge_issue_url(forge_url: str, issue_id: int) -> str:
    """
    Construct the Forge issue URL from the project URL and issue ID.

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

    Returns:
        Full Forge issue URL
    """
    # Remove trailing slash if present
    forge_url = forge_url.rstrip("/")
    return f"{forge_url}/issues/{issue_id}"


def process_issues(
    api_base: str,
    project: str,
    forge_url: str,
    api_token: str,
    issue_numbers: list[int] | None,
    issue_status: str,
    skip_comment: bool,
    skip_close: bool,
    dry_run: bool,
    delay: float,
) -> tuple[int, int, list[int]]:
    """
    Process issues and add migration comments.

    Args:
        api_base: Pagure API base URL
        project: Pagure project path
        forge_url: Full URL to the Forge project
        api_token: Pagure API token
        issue_numbers: Specific issue numbers to process, or None for all
        issue_status: Pagure status filter ('all', 'Open', 'Closed')
        skip_comment: If True, do not post migration comments
        skip_close: If True, do not close open issues
        dry_run: If True, don't actually modify issues
        delay: Delay between API calls in seconds

    Returns:
        Tuple of (processed, skipped, failed_issue_ids)
    """
    processed = 0
    skipped = 0
    failed_ids: list[int] = []

    # *** Retrieve issues ***
    if issue_numbers:
        print(f"Fetching {len(issue_numbers)} specific issue(s) from {project}...")
        issues = []
        for issue_id in issue_numbers:
            try:
                issue = get_issue(api_base, project, issue_id, api_token)
            except requests.exceptions.HTTPError as e:
                print(f"  Error fetching issue #{issue_id}: {e}")
                failed_ids.append(issue_id)
                if delay > 0:
                    time.sleep(delay)
                continue

            if issue_status != "all" and issue["status"] != issue_status:
                print(
                    f"  Skipping #{issue_id}: status is {issue['status']} "
                    f"(filter: {issue_status})"
                )
                skipped += 1
            else:
                issues.append(issue)

            if delay > 0:
                time.sleep(delay)
    else:
        print(f"Fetching issues from {project}...")
        issues = get_pagure_issues(
            api_base, project, api_token, status=issue_status, delay=delay
        )
        print(f"Found {len(issues)} issues")

    # *** Update issues ***
    for issue in issues:
        issue_id = issue["id"]
        status = issue["status"]
        comments = issue.get("comments", [])
        forge_issue_url = construct_forge_issue_url(forge_url, issue_id)
        comment_text = DEFAULT_COMMENT_TEMPLATE.format(
            forge_url=forge_issue_url, issue_id=issue_id
        )

        print(f"\nProcessing issue #{issue_id}...")

        requested_comment = not skip_comment
        requested_close = not skip_close
        needs_comment = requested_comment and not has_migration_comment(comments)
        needs_close = not skip_close and status == "Open"

        if requested_comment and has_migration_comment(comments):
            print("  Migration comment already exists")

        if requested_close and status == "Closed":
            print("  Issue is already closed")

        if not needs_comment and not needs_close:
            print("  Skipping: nothing to do")
            skipped += 1
            continue

        if dry_run:
            if needs_comment:
                print(f"  [DRY RUN] Would add comment linking to: {forge_issue_url}")
            if needs_close:
                print("  [DRY RUN] Would close issue")
            processed += 1
        else:
            if needs_comment:
                print(f"  Adding migration comment linking to: {forge_issue_url}")
            if needs_close:
                print("  Closing issue...")

            if update_issue(
                api_base,
                project,
                issue_id,
                api_token,
                comment=comment_text if needs_comment else None,
                close=needs_close,
                issue_status=status,
            ):
                print("  Success!")
                processed += 1
            else:
                failed_ids.append(issue_id)

            if delay > 0:
                time.sleep(delay)

    return processed, skipped, failed_ids


def main():
    parser = argparse.ArgumentParser(
        description="Add migration notice comments to Pagure tickets",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Dry run on all issues (token from file)
  %(prog)s https://pagure.io/fedora-qa/some-repo https://forge.fedoraproject.org/quality/some-repo --token-file ~/.pagure-token --dry-run

  # Process specific issues on staging Pagure
  %(prog)s https://stg.pagure.io/fedora-qa/some-repo https://forge.fedoraproject.org/quality/some-repo --token YOUR_TOKEN --issues 1 2 3

  # Comment only, without closing issues
  %(prog)s https://pagure.io/fedora-qa/some-repo https://forge.fedoraproject.org/quality/some-repo --token-file ~/.pagure-token --skip-close
        """,
    )

    parser.add_argument(
        "pagure_url",
        help="Full URL to the Pagure project (e.g., 'https://pagure.io/fedora-qa/some-repo' or 'https://stg.pagure.io/fedora-qa/some-repo')",
    )
    parser.add_argument(
        "forge_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="Pagure API token for authentication",
    )
    token_group.add_argument(
        "--token-file",
        type=Path,
        help="Path to file containing the Pagure API token",
    )
    parser.add_argument(
        "--issues",
        type=int,
        nargs="+",
        help="Specific issue numbers to process (default: all issues)",
    )
    parser.add_argument(
        "--issue-status",
        choices=["open", "closed", "all"],
        default="all",
        help="Only process issues with this status (default: all)",
    )
    parser.add_argument(
        "--skip-close",
        action="store_true",
        help="Do not close open issues (by default, open issues are closed)",
    )
    parser.add_argument(
        "--skip-comment",
        action="store_true",
        help="Do not post migration comment (by default, migration comment is posted)",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Preview changes without actually modifying issues",
    )
    parser.add_argument(
        "--delay",
        type=float,
        default=1.0,
        help="Delay between API calls in seconds (default: 1.0)",
    )

    args = parser.parse_args()

    if args.skip_comment and args.skip_close:
        print("Error: --skip-comment and --skip-close together would do nothing")
        sys.exit(1)

    # Parse the Pagure URL
    try:
        api_base, project = parse_pagure_url(args.pagure_url)
    except ValueError as e:
        print(f"Error: {e}")
        sys.exit(1)

    # Get the API token
    if args.token:
        api_token = args.token
    else:
        try:
            api_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)

    pagure_status = PAGURE_STATUS[args.issue_status]

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

    try:
        processed, skipped, failed_ids = process_issues(
            api_base=api_base,
            project=project,
            forge_url=args.forge_url,
            api_token=api_token,
            issue_numbers=args.issues,
            issue_status=pagure_status,
            skip_comment=args.skip_comment,
            skip_close=args.skip_close,
            dry_run=args.dry_run,
            delay=args.delay,
        )
    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"  Processed: {processed}")
    print(f"  Skipped: {skipped}")
    if failed_ids:
        failed_list = " ".join(str(issue_id) for issue_id in sorted(failed_ids))
        print(f"  Failed: {len(failed_ids)} (Issue IDs: {failed_list})")
    else:
        print("  Failed: 0")

    if failed_ids:
        sys.exit(1)


if __name__ == "__main__":
    main()