122 lines
4.4 KiB
Python
Executable file
122 lines
4.4 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
import requests
|
|
import json
|
|
from dotenv import load_dotenv
|
|
import os
|
|
|
|
# ------------------ CONFIG ------------------
|
|
load_dotenv()
|
|
|
|
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
|
|
GITHUB_USERNAME = os.getenv("GITHUB_USERNAME")
|
|
|
|
FORGEJO_TOKEN = os.getenv("FORGEJO_TOKEN")
|
|
FORGEJO_USERNAME = os.getenv("FORGEJO_USERNAME") # only for repo path
|
|
FORGEJO_URL = os.getenv("FORGEJO_URL")
|
|
|
|
# Migration options
|
|
def str2bool(v):
|
|
return str(v).lower() in ("yes", "true", "1")
|
|
|
|
MIRROR = str2bool(os.getenv("MIRROR", "False"))
|
|
IMPORT_ISSUES = str2bool(os.getenv("IMPORT_ISSUES", "True"))
|
|
IMPORT_LABELS = str2bool(os.getenv("IMPORT_LABELS", "True"))
|
|
IMPORT_MILESTONES = str2bool(os.getenv("IMPORT_MILESTONES", "True"))
|
|
IMPORT_PR = str2bool(os.getenv("IMPORT_PR", "True"))
|
|
IMPORT_WIKI = str2bool(os.getenv("IMPORT_WIKI", "True"))
|
|
# --------------------------------------------
|
|
|
|
# Get Forgejo user ID using personal token
|
|
def get_forgejo_uid():
|
|
headers = {"Authorization": f"token {FORGEJO_TOKEN}"}
|
|
resp = requests.get(f"{FORGEJO_URL}/api/v1/user", headers=headers)
|
|
resp.raise_for_status()
|
|
return resp.json()["id"]
|
|
|
|
# Get GitHub repos including archived status and privacy
|
|
def get_github_repos():
|
|
headers = {"Authorization": f"token {GITHUB_TOKEN}"}
|
|
repos = []
|
|
page = 1
|
|
while True:
|
|
# Use /user/repos for authenticated user and visibility=all
|
|
url = f"https://api.github.com/user/repos?per_page=100&page={page}&visibility=all&affiliation=owner"
|
|
resp = requests.get(url, headers=headers)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
if not data:
|
|
break
|
|
for r in data:
|
|
repos.append({
|
|
"name": r["name"],
|
|
"archived": r["archived"],
|
|
"private": r["private"]
|
|
})
|
|
page += 1
|
|
return repos
|
|
|
|
# Check if repo exists in Forgejo
|
|
def get_forgejo_repo(repo_name):
|
|
headers = {"Authorization": f"token {FORGEJO_TOKEN}"}
|
|
resp = requests.get(f"{FORGEJO_URL}/api/v1/repos/{FORGEJO_USERNAME}/{repo_name}", headers=headers)
|
|
if resp.status_code == 200:
|
|
return resp.json()
|
|
return None
|
|
|
|
# Migrate a repo to Forgejo using authenticated clone if private
|
|
def migrate_repo(repo_name, uid, private):
|
|
# Use HTTPS clone URL for GitHub
|
|
clone_url = f"https://{GITHUB_USERNAME}:{GITHUB_TOKEN}@github.com/{GITHUB_USERNAME}/{repo_name}.git" if private else f"https://github.com/{GITHUB_USERNAME}/{repo_name}.git"
|
|
headers = {"Authorization": f"token {FORGEJO_TOKEN}", "Content-Type": "application/json"}
|
|
payload = {
|
|
"clone_addr": clone_url,
|
|
"uid": uid,
|
|
"repo_name": repo_name,
|
|
"mirror": MIRROR,
|
|
"private": private,
|
|
"auth_username": GITHUB_USERNAME if private else "",
|
|
"auth_password": GITHUB_TOKEN if private else "",
|
|
"issues": IMPORT_ISSUES,
|
|
"labels": IMPORT_LABELS,
|
|
"milestones": IMPORT_MILESTONES,
|
|
"pull_requests": IMPORT_PR,
|
|
"wiki": IMPORT_WIKI
|
|
}
|
|
resp = requests.post(f"{FORGEJO_URL}/api/v1/repos/migrate", headers=headers, data=json.dumps(payload))
|
|
if resp.status_code == 201:
|
|
print(f"[SUCCESS] Migrated {repo_name}")
|
|
else:
|
|
print(f"[ERROR] Failed to migrate {repo_name}: {resp.status_code} {resp.text}")
|
|
|
|
# Update existing repo visibility or archive
|
|
def update_repo(repo_name, private, archived):
|
|
headers = {"Authorization": f"token {FORGEJO_TOKEN}", "Content-Type": "application/json"}
|
|
payload = {"private": private, "archived": archived}
|
|
resp = requests.patch(f"{FORGEJO_URL}/api/v1/repos/{FORGEJO_USERNAME}/{repo_name}", headers=headers, data=json.dumps(payload))
|
|
if resp.status_code == 200:
|
|
print(f"[UPDATED] {repo_name}: private={private}, archived={archived}")
|
|
else:
|
|
print(f"[ERROR] Failed to update {repo_name}: {resp.status_code} {resp.text}")
|
|
|
|
def main():
|
|
uid = get_forgejo_uid()
|
|
repos = get_github_repos()
|
|
print(f"Found {len(repos)} repos on GitHub.")
|
|
|
|
for repo in repos:
|
|
name = repo["name"]
|
|
private = repo["private"]
|
|
archived = repo["archived"]
|
|
|
|
existing = get_forgejo_repo(name)
|
|
if existing:
|
|
# Update visibility and archive status if needed
|
|
update_repo(name, private, archived)
|
|
continue
|
|
|
|
migrate_repo(name, uid, private)
|
|
if archived:
|
|
update_repo(name, private, archived) # archive after migration
|
|
|
|
if __name__ == "__main__":
|
|
main()
|