quality/forge-helpers
6
0
Fork 1
Code Issues Pull requests Releases Activity Actions

add pagure_add_migration_comment.py

Browse source
This commit is contained in:
Kamil Páral 2025-11-26 14:28:42 +01:00
commit 35977a8989
5 changed files with 549 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

11
.gitignore vendored Normal file
View file

@ -0,0 +1,11 @@
# Python virtual environment
.venv/
# Python cache
__pycache__/
*.pyc
*.pyo
# Environment variables
.env

57
README.md Normal file
View file

@ -0,0 +1,57 @@
# Forge Helpers
This repository contains various helpers used for migrating Fedora QA repositories from [Pagure](https://pagure.io/group/fedora-qa) into [Fedora Forge](https://forge.fedoraproject.org/quality) (based on [Forgejo](https://forgejo.org/)), and mass-changing the target tickets if there is some subsequent action needed.
## Setup
```bash
# Create and activate virtualenv
python3 -m venv .venv
source .venv/bin/activate
# Install dependencies
pip install -r requirements.txt
```
## Pagure API Token
Several scripts require a Pagure API token for authentication. To create one:
1. Log in to [Pagure](https://pagure.io/)
2. Go to your [API settings](https://pagure.io/settings#nav-api-tab)
3. Click "Create new API key"
4. Select the required ACLs (e.g., "Comment on a ticket" for adding migration comments)
5. Save the token to a file (e.g., `~/.pagure-token`) for use with `--token-file`
## Scripts
### pagure_add_migration_comment.py
Adds a migration notice comment to Pagure tickets, linking to their new Forge location.
**Usage:**
```bash
# Dry run on all issues (preview without making changes)
python pagure_add_migration_comment.py 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
python pagure_add_migration_comment.py https://stg.pagure.io/fedora-qa/some-repo \
https://forge.fedoraproject.org/quality/some-repo \
--token-file ~/.pagure-token --issues 1 2 3
# Process all issues with custom comment
python pagure_add_migration_comment.py https://pagure.io/fedora-qa/some-repo \
https://forge.fedoraproject.org/quality/some-repo \
--token-file ~/.pagure-token \
--comment "Moved to: {forge_url}"
```
Run with `--help` for all available options.
## Note
Most of the code in this repository is AI-generated using Claude (Anthropic) via Cursor IDE.

68
pagure-migration-comment.plan.md Normal file
View file

@ -0,0 +1,68 @@
# Pagure Migration Comment Helper
## Goal
Create a Python script that 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.
## Implementation
### 1. Project Setup
- Create a Python virtualenv in `.venv/` directory
- Create `requirements.txt` with `requests` library for API calls
- Install dependencies into the virtualenv
- Add `.venv/` to `.gitignore`
- Create main script `pagure_add_migration_comment.py`
### 2. Script Features
The script will:
- Accept command-line arguments for:
- Full Pagure project URL (e.g., `https://pagure.io/fedora-qa/some-repo` or `https://stg.pagure.io/fedora-qa/some-repo`)
- Forge project URL base (to construct target links)
- Pagure API token (via `--token` on command line or `--token-file` to read from file)
- Optional: specific issue number(s) or "all" to process all open issues
- Optional: dry-run mode to preview without making changes
- Optional: custom comment template
- Optional: delay between API calls (default: 1 second)
- Query Pagure API to get issue(s)
- Construct the corresponding Forge URL (same issue number, different base URL)
- Add a comment to each Pagure issue via API
### 3. Pagure API Integration
- Use Pagure API endpoint: `POST /api/0/<repo>/issue/<issue_id>/comment`
- Authentication via API token in headers
- Comment text will be configurable but default to something like:
> "This issue has been migrated to Fedora Forge: <link>"
### 4. Error Handling
- Handle rate limiting (configurable delay between requests)
- Log successes and failures
- Support resuming if interrupted (skip already-commented issues)
## Usage
```bash
# Activate virtualenv
source .venv/bin/activate
# Dry run on all issues (token from file)
python pagure_add_migration_comment.py 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
python pagure_add_migration_comment.py https://stg.pagure.io/fedora-qa/some-repo \
https://forge.fedoraproject.org/quality/some-repo \
--token YOUR_TOKEN --issues 1 2 3
# Process all issues with custom comment
python pagure_add_migration_comment.py https://pagure.io/fedora-qa/some-repo \
https://forge.fedoraproject.org/quality/some-repo \
--token-file ~/.pagure-token \
--comment "Moved to: {forge_url}"
```

411
pagure_add_migration_comment.py Executable file
View file

@ -0,0 +1,411 @@
#!/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.
"""
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:"
# 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_comments(
api_base: str, project: str, issue_id: int, api_token: str
) -> list[dict]:
"""
Fetch comments for a specific issue.
Args:
api_base: Pagure API base URL
project: Pagure project path
issue_id: Issue number
api_token: Pagure API token
Returns:
List of comment dictionaries
"""
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()
data = response.json()
return data.get("comments", [])
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 add_comment_to_issue(
api_base: str, project: str, issue_id: int, comment: str, api_token: str
) -> bool:
"""
Add a comment to a Pagure issue.
Args:
api_base: Pagure API base URL
project: Pagure project path
issue_id: Issue number
comment: Comment text to add
api_token: Pagure API token
Returns:
True if successful, False otherwise
"""
url = f"{api_base}/{project}/issue/{issue_id}/comment"
headers = {"Authorization": f"token {api_token}"}
data = {"comment": comment}
response = requests.post(url, headers=headers, data=data, timeout=30)
if response.status_code == 200:
return True
else:
print(f" Error adding comment: {response.status_code} - {response.text}")
return False
def construct_forge_url(forge_base: str, issue_id: int) -> str:
"""
Construct the Forge issue URL from the base URL and issue ID.
Args:
forge_base: Base URL for 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_base = forge_base.rstrip("/")
return f"{forge_base}/issues/{issue_id}"
def process_issues(
api_base: str,
project: str,
forge_base: str,
api_token: str,
issue_numbers: list[int] | None,
comment_template: str,
dry_run: bool,
delay: float,
) -> tuple[int, int, int]:
"""
Process issues and add migration comments.
Args:
api_base: Pagure API base URL
project: Pagure project path
forge_base: Base URL for the Forge project
api_token: Pagure API token
issue_numbers: Specific issue numbers to process, or None for all
comment_template: Template for the migration comment
dry_run: If True, don't actually add comments
delay: Delay between API calls in seconds
Returns:
Tuple of (processed, skipped, failed) counts
"""
processed = 0
skipped = 0
failed = 0
if issue_numbers:
# Process specific issues
issues = [{"id": num} for num in issue_numbers]
else:
# Fetch all issues
print(f"Fetching issues from {project}...")
issues = get_pagure_issues(api_base, project, api_token, delay=delay)
print(f"Found {len(issues)} issues")
for issue in issues:
issue_id = issue["id"]
forge_url = construct_forge_url(forge_base, issue_id)
comment = comment_template.format(forge_url=forge_url, issue_id=issue_id)
print(f"\nProcessing issue #{issue_id}...")
# Check for existing migration comment
try:
comments = get_issue_comments(api_base, project, issue_id, api_token)
except requests.exceptions.HTTPError as e:
print(f" Error fetching issue: {e}")
failed += 1
continue
if has_migration_comment(comments):
print(" Skipping: migration comment already exists")
skipped += 1
# Delay after API call to avoid rate limiting
if delay > 0:
time.sleep(delay)
continue
if dry_run:
print(f" [DRY RUN] Would add comment linking to: {forge_url}")
processed += 1
# Delay after API call to avoid rate limiting
if delay > 0:
time.sleep(delay)
else:
print(f" Adding migration comment linking to: {forge_url}")
if add_comment_to_issue(api_base, project, issue_id, comment, api_token):
print(" Success!")
processed += 1
else:
failed += 1
# Delay after API call to avoid rate limiting
if delay > 0:
time.sleep(delay)
return processed, skipped, failed
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
# Process all issues with custom comment
%(prog)s https://pagure.io/fedora-qa/some-repo https://forge.fedoraproject.org/quality/some-repo --token-file ~/.pagure-token --comment "Moved to: {forge_url}"
""",
)
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_base",
help="Base URL for 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(
"--comment",
default=DEFAULT_COMMENT_TEMPLATE,
help="Custom comment template. Use {forge_url} and {issue_id} as placeholders",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Preview changes without actually adding comments",
)
parser.add_argument(
"--delay",
type=float,
default=1.0,
help="Delay between API calls in seconds (default: 1.0)",
)
args = parser.parse_args()
# 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)
print(f"Pagure URL: {args.pagure_url}")
print(f"Forge base URL: {args.forge_base}")
if args.dry_run:
print("Mode: DRY RUN (no changes will be made)")
print()
try:
processed, skipped, failed = process_issues(
api_base=api_base,
project=project,
forge_base=args.forge_base,
api_token=api_token,
issue_numbers=args.issues,
comment_template=args.comment,
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 (already migrated): {skipped}")
print(f" Failed: {failed}")
if failed > 0:
sys.exit(1)
if __name__ == "__main__":
main()

2
requirements.txt Normal file
View file

@ -0,0 +1,2 @@
requests>=2.28.0