479 lines
14 KiB
Python
Executable file
479 lines
14 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""
|
|
Migrate story points from Pagure custom fields to Forge labels.
|
|
|
|
This script reads the story_points custom field from Pagure issues and applies
|
|
corresponding point labels (points/01, points/02, etc.) to the matching Forge issues.
|
|
|
|
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
|
|
|
|
# Allowed story point values and their corresponding labels
|
|
ALLOWED_POINTS = [1, 2, 3, 5, 8, 13]
|
|
POINTS_TO_LABEL = {
|
|
1: "points/01",
|
|
2: "points/02",
|
|
3: "points/03",
|
|
5: "points/05",
|
|
8: "points/08",
|
|
13: "points/13",
|
|
}
|
|
|
|
|
|
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 parse_forge_url(forge_url: str) -> tuple[str, str]:
|
|
"""
|
|
Parse a Forge project URL into API base URL and owner/repo path.
|
|
|
|
Args:
|
|
forge_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(forge_url)
|
|
if not parsed.scheme or not parsed.netloc:
|
|
raise ValueError(f"Invalid Forge URL: {forge_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): {forge_url}")
|
|
|
|
return api_base, owner_repo
|
|
|
|
|
|
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_issue(api_base: str, project: str, issue_id: int) -> dict:
|
|
"""
|
|
Fetch a single issue from Pagure (no authentication required).
|
|
|
|
Args:
|
|
api_base: Pagure API base URL (e.g., 'https://pagure.io/api/0')
|
|
project: Pagure project path (e.g., 'fedora-qa/some-repo')
|
|
issue_id: Issue number
|
|
|
|
Returns:
|
|
Issue dictionary
|
|
|
|
Raises:
|
|
requests.exceptions.HTTPError: If the API call fails
|
|
"""
|
|
url = f"{api_base}/{project}/issue/{issue_id}"
|
|
response = requests.get(url, timeout=30)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def get_pagure_issues(
|
|
api_base: str, project: str, status: str = "all", delay: float = 1.0
|
|
) -> list[dict]:
|
|
"""
|
|
Fetch all issues from a Pagure project (no authentication required).
|
|
|
|
Args:
|
|
api_base: Pagure API base URL (e.g., 'https://pagure.io/api/0')
|
|
project: Pagure project path (e.g., 'fedora-qa/some-repo')
|
|
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"
|
|
params = {"status": status, "per_page": 100}
|
|
|
|
all_issues = []
|
|
page = 1
|
|
|
|
while True:
|
|
params["page"] = page
|
|
response = requests.get(url, 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 normalize_story_points(value: str) -> int | None:
|
|
"""
|
|
Parse and normalize a story points value.
|
|
|
|
Args:
|
|
value: The story points value as a string
|
|
|
|
Returns:
|
|
Normalized story points (1, 2, 3, 5, 8, or 13), or None if invalid
|
|
|
|
The value is rounded up to the nearest allowed value.
|
|
Values greater than 13 are capped at 13.
|
|
"""
|
|
try:
|
|
points = float(value)
|
|
if points != int(points):
|
|
# Non-integer floats are considered invalid
|
|
return None
|
|
points = int(points)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
if points <= 0:
|
|
return None
|
|
|
|
# Round up to the nearest allowed value
|
|
for allowed in ALLOWED_POINTS:
|
|
if points <= allowed:
|
|
return allowed
|
|
|
|
# Cap at maximum
|
|
return 13
|
|
|
|
|
|
def add_forge_label(
|
|
api_base: str, owner_repo: str, issue_id: int, label: str, api_token: str
|
|
) -> bool:
|
|
"""
|
|
Add a label to a Forge (Forgejo) issue.
|
|
|
|
Args:
|
|
api_base: Forge API base URL (e.g., 'https://forge.fedoraproject.org/api/v1')
|
|
owner_repo: Owner and repo path (e.g., 'quality/some-repo')
|
|
issue_id: Issue number
|
|
label: Label name to add
|
|
api_token: Forge API token for authentication
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
url = f"{api_base}/repos/{owner_repo}/issues/{issue_id}/labels"
|
|
headers = {
|
|
"Authorization": f"token {api_token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
data = {"labels": [label]}
|
|
|
|
response = requests.post(url, headers=headers, json=data, timeout=30)
|
|
|
|
if response.status_code in (200, 201):
|
|
return True
|
|
else:
|
|
print(f" Error adding label: {response.status_code} - {response.text}")
|
|
return False
|
|
|
|
|
|
def extract_story_points(issue: dict) -> str | None:
|
|
"""
|
|
Extract the story_points custom field value from a Pagure issue.
|
|
|
|
Args:
|
|
issue: Pagure issue dictionary
|
|
|
|
Returns:
|
|
The story_points value as a string, or None if empty or field doesn't exist
|
|
"""
|
|
custom_fields = issue.get("custom_fields", [])
|
|
|
|
for field in custom_fields:
|
|
if field.get("name") == "story_points":
|
|
value = field.get("value", "")
|
|
if value is None or str(value).strip() == "":
|
|
return None
|
|
return str(value).strip()
|
|
|
|
# Field doesn't exist (Pagure doesn't return empty custom fields)
|
|
return None
|
|
|
|
|
|
def process_issues(
|
|
pagure_api_base: str,
|
|
pagure_project: str,
|
|
forge_api_base: str,
|
|
forge_owner_repo: str,
|
|
forge_token: str,
|
|
issue_numbers: list[int] | None,
|
|
dry_run: bool,
|
|
delay: float,
|
|
) -> tuple[int, int, int]:
|
|
"""
|
|
Process issues and migrate story points to Forge labels.
|
|
|
|
Args:
|
|
pagure_api_base: Pagure API base URL
|
|
pagure_project: Pagure project path
|
|
forge_api_base: Forge API base URL
|
|
forge_owner_repo: Forge owner/repo path
|
|
forge_token: Forge API token
|
|
issue_numbers: Specific issue numbers to process, or None for all
|
|
dry_run: If True, don't actually add labels
|
|
delay: Delay between API calls in seconds
|
|
|
|
Returns:
|
|
Tuple of (processed, skipped, failed) counts
|
|
"""
|
|
processed = 0
|
|
skipped = 0
|
|
failed = 0
|
|
|
|
if issue_numbers:
|
|
# Fetch specific issues one by one
|
|
print(f"Fetching {len(issue_numbers)} specific issue(s) from Pagure...")
|
|
issues = []
|
|
for issue_id in issue_numbers:
|
|
try:
|
|
issue = get_pagure_issue(pagure_api_base, pagure_project, issue_id)
|
|
issues.append(issue)
|
|
except requests.exceptions.HTTPError as e:
|
|
print(f" Error fetching issue #{issue_id}: {e}")
|
|
failed += 1
|
|
if delay > 0:
|
|
time.sleep(delay)
|
|
else:
|
|
# Fetch all issues
|
|
print(f"Fetching all issues from {pagure_project}...")
|
|
issues = get_pagure_issues(pagure_api_base, pagure_project, delay=delay)
|
|
print(f"Found {len(issues)} issues")
|
|
|
|
for issue in issues:
|
|
issue_id = issue["id"]
|
|
print(f"\nProcessing issue #{issue_id}...")
|
|
|
|
# Extract story_points custom field
|
|
story_points_value = extract_story_points(issue)
|
|
|
|
# Check if story_points is missing or empty
|
|
if story_points_value is None:
|
|
print(" Skipping: story_points field doesn't exist or is empty")
|
|
skipped += 1
|
|
if delay > 0:
|
|
time.sleep(delay)
|
|
continue
|
|
|
|
# Normalize story points
|
|
normalized_points = normalize_story_points(story_points_value)
|
|
if normalized_points is None:
|
|
print(f" Failed: invalid story_points value '{story_points_value}' (not a number)")
|
|
failed += 1
|
|
if delay > 0:
|
|
time.sleep(delay)
|
|
continue
|
|
|
|
# Get the corresponding label
|
|
label = POINTS_TO_LABEL[normalized_points]
|
|
|
|
if story_points_value != str(normalized_points):
|
|
print(f" Story points: {story_points_value} -> normalized to {normalized_points}")
|
|
else:
|
|
print(f" Story points: {normalized_points}")
|
|
|
|
if dry_run:
|
|
print(f" [DRY RUN] Would add label '{label}' to Forge issue #{issue_id}")
|
|
processed += 1
|
|
else:
|
|
print(f" Adding label '{label}' to Forge issue #{issue_id}...")
|
|
if add_forge_label(forge_api_base, forge_owner_repo, issue_id, label, forge_token):
|
|
print(" Success!")
|
|
processed += 1
|
|
else:
|
|
failed += 1
|
|
|
|
if delay > 0:
|
|
time.sleep(delay)
|
|
|
|
return processed, skipped, failed
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Migrate story points from Pagure custom fields to Forge labels",
|
|
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 ~/.forge-token --dry-run
|
|
|
|
# Process specific issues
|
|
%(prog)s https://pagure.io/fedora-qa/some-repo https://forge.fedoraproject.org/quality/some-repo --token YOUR_TOKEN --issues 1 2 3
|
|
""",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"pagure_url",
|
|
help="Full URL to the Pagure project (e.g., 'https://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="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(
|
|
"--issues",
|
|
type=int,
|
|
nargs="+",
|
|
help="Specific issue numbers to process (default: all issues)",
|
|
)
|
|
parser.add_argument(
|
|
"--dry-run",
|
|
action="store_true",
|
|
help="Preview changes without actually adding labels",
|
|
)
|
|
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:
|
|
pagure_api_base, pagure_project = parse_pagure_url(args.pagure_url)
|
|
except ValueError as e:
|
|
print(f"Error: {e}")
|
|
sys.exit(1)
|
|
|
|
# Parse the Forge URL
|
|
try:
|
|
forge_api_base, forge_owner_repo = parse_forge_url(args.forge_url)
|
|
except ValueError as e:
|
|
print(f"Error: {e}")
|
|
sys.exit(1)
|
|
|
|
# Get the Forge API token
|
|
if args.token:
|
|
forge_token = args.token
|
|
else:
|
|
try:
|
|
forge_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 URL: {args.forge_url}")
|
|
if args.dry_run:
|
|
print("Mode: DRY RUN (no changes will be made)")
|
|
print()
|
|
|
|
try:
|
|
processed, skipped, failed = process_issues(
|
|
pagure_api_base=pagure_api_base,
|
|
pagure_project=pagure_project,
|
|
forge_api_base=forge_api_base,
|
|
forge_owner_repo=forge_owner_repo,
|
|
forge_token=forge_token,
|
|
issue_numbers=args.issues,
|
|
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 (no story_points): {skipped}")
|
|
print(f" Failed: {failed}")
|
|
|
|
if failed > 0:
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|