Update scripts/sync_forgejo_issues.py [skip ci]

scripts/sync_forgejo_issues.py

@@ -0,0 +1,204 @@
+#!/usr/bin/env python3
+"""
+Sync issues from Forgejo to GitHub
+
+This script fetches issues from a Forgejo repository and creates corresponding
+issues in a GitHub repository, avoiding duplicates.
+
+Usage:
+    python sync_forgejo_issues.py
+
+Environment Variables:
+    FORGEJO_URL: Forgejo API URL (e.g., https://git.suyu.dev/api/v1/repos/suyu/suyu/issues)
+    FORGEJO_TOKEN: Forgejo access token (optional for public repos)
+    GITHUB_REPOSITORY: Target GitHub repository (e.g., owner/repo)
+    GITHUB_TOKEN: GitHub access token
+"""
+
+import os
+import requests
+import json
+import time
+import sys
+from urllib.parse import urlparse
+
+
+def get_existing_github_issues(github_repo, github_headers):
+    """Get all existing GitHub issues to avoid duplicates"""
+    existing_issues = {}
+    page = 1
+
+    print("📋 Fetching existing GitHub issues...")
+
+    while True:
+        url = f"https://api.github.com/repos/{github_repo}/issues"
+        params = {"state": "all", "per_page": 100, "page": page}
+
+        try:
+            response = requests.get(url, headers=github_headers, params=params)
+            response.raise_for_status()
+            issues = response.json()
+
+            if not issues:
+                break
+
+            for issue in issues:
+                # Use title as key for duplicate detection
+                existing_issues[issue["title"]] = issue["number"]
+
+            page += 1
+
+        except requests.exceptions.RequestException as e:
+            print(f"❌ Error fetching existing GitHub issues: {e}")
+            break
+
+    return existing_issues
+
+
+def fetch_forgejo_issues(forgejo_api_url, forgejo_headers):
+    """Fetch issues from Forgejo API"""
+    all_issues = []
+    page = 1
+
+    print("🔍 Fetching issues from Forgejo...")
+
+    while True:
+        params = {"state": "open", "per_page": 50, "page": page}
+
+        try:
+            response = requests.get(forgejo_api_url, headers=forgejo_headers, params=params)
+
+            if response.status_code == 403:
+                print("⚠️  403 Forbidden: Trying without authentication...")
+                # Try without authentication for public repos
+                response = requests.get(forgejo_api_url, params=params)
+
+            response.raise_for_status()
+            issues = response.json()
+
+            if not issues:
+                break
+
+            all_issues.extend(issues)
+            page += 1
+
+        except requests.exceptions.RequestException as e:
+            print(f"❌ Error fetching issues from Forgejo: {e}")
+            if page == 1:  # If first page fails, exit
+                return []
+            break
+
+    return all_issues
+
+
+def create_github_issue(title, body, labels, github_repo, github_headers, forgejo_api_url):
+    """Create a new issue on GitHub"""
+    create_issue_url = f"https://api.github.com/repos/{github_repo}/issues"
+
+    # Prepare issue body with source attribution
+    source_url = forgejo_api_url.replace('/api/v1/repos/', '/').replace('/issues', '')
+    attributed_body = f"**Imported from Forgejo**: {source_url}\n\n---\n\n{body}"
+
+    new_issue = {
+        "title": f"[Forgejo] {title}",
+        "body": attributed_body,
+    }
+
+    if labels:
+        new_issue["labels"] = labels + ["imported-from-forgejo"]
+    else:
+        new_issue["labels"] = ["imported-from-forgejo"]
+
+    try:
+        create_response = requests.post(create_issue_url, json=new_issue, headers=github_headers)
+
+        if create_response.status_code == 201:
+            issue_data = create_response.json()
+            print(f"✅ Issue '{title}' created successfully (#{issue_data['number']})")
+            return True
+        else:
+            print(f"❌ Failed to create issue '{title}': {create_response.status_code}")
+            print(f"Response: {create_response.text}")
+            return False
+
+    except requests.exceptions.RequestException as e:
+        print(f"❌ Error creating issue '{title}' on GitHub: {e}")
+        return False
+
+
+def main():
+    # Environment variables
+    forgejo_api_url = os.getenv("FORGEJO_URL")
+    forgejo_token = os.getenv("FORGEJO_TOKEN")
+    github_repo = os.getenv("GITHUB_REPOSITORY")
+    github_token = os.getenv("GITHUB_TOKEN")
+
+    # Validate required environment variables
+    if not forgejo_api_url:
+        print("❌ FORGEJO_URL environment variable is required")
+        sys.exit(1)
+    if not github_repo:
+        print("❌ GITHUB_REPOSITORY environment variable is required")
+        sys.exit(1)
+    if not github_token:
+        print("❌ GITHUB_TOKEN environment variable is required")
+        sys.exit(1)
+
+    # Headers for API requests
+    forgejo_headers = {}
+    if forgejo_token:
+        forgejo_headers["Authorization"] = f"token {forgejo_token}"
+
+    github_headers = {"Authorization": f"Bearer {github_token}"}
+
+    print("🔄 Starting Forgejo to GitHub issue sync...")
+
+    # Get existing GitHub issues
+    existing_issues = get_existing_github_issues(github_repo, github_headers)
+    print(f"Found {len(existing_issues)} existing GitHub issues")
+
+    # Fetch Forgejo issues
+    forgejo_issues = fetch_forgejo_issues(forgejo_api_url, forgejo_headers)
+
+    if not forgejo_issues:
+        print("❌ No issues found on Forgejo or failed to fetch issues")
+        sys.exit(1)
+
+    print(f"Found {len(forgejo_issues)} issues on Forgejo")
+
+    # Process each Forgejo issue
+    created_count = 0
+    skipped_count = 0
+
+    for issue in forgejo_issues:
+        title = issue.get("title", "Untitled Issue")
+        body = issue.get("body", "")
+        labels = [label["name"] for label in issue.get("labels", [])]
+
+        # Check for duplicates (with and without [Forgejo] prefix)
+        prefixed_title = f"[Forgejo] {title}"
+
+        if title in existing_issues or prefixed_title in existing_issues:
+            print(f"⏭️  Issue '{title}' already exists, skipping...")
+            skipped_count += 1
+            continue
+
+        # Create the issue
+        if create_github_issue(title, body, labels, github_repo, github_headers, forgejo_api_url):
+            created_count += 1
+            # Add small delay to avoid rate limiting
+            time.sleep(1)
+
+        # Rate limiting protection
+        if created_count > 0 and created_count % 10 == 0:
+            print("⏸️  Pausing for rate limit protection...")
+            time.sleep(5)
+
+    print(f"\n📊 Sync completed:")
+    print(f"   ✅ Created: {created_count} issues")
+    print(f"   ⏭️  Skipped: {skipped_count} issues")
+    print(f"   📋 Total processed: {len(forgejo_issues)} issues")
+
+
+if __name__ == "__main__":
+    main()