codex
118 lines
4.0 KiB
Python
118 lines
4.0 KiB
Python
import os
|
|
import subprocess
|
|
import json
|
|
import urllib.request
|
|
import urllib.error
|
|
import base64
|
|
import shutil
|
|
import time
|
|
|
|
GITLAB_API = "https://gitlab.com/api/v4"
|
|
GITLAB_TOKEN = "glpat-KrN8svq4D9QeQD6-_p_r7GM6MQpvOjEKdTozN3R4cg8.01.171o6kz82"
|
|
FORGEJO_LOCAL_URL = "http://forgejo.local/api/v1"
|
|
AUTH = ("mrhavens", "Aok4y2k!")
|
|
BACKUP_DIR = "/home/antigravity/scratch/gitlab_backups"
|
|
|
|
os.makedirs(BACKUP_DIR, exist_ok=True)
|
|
|
|
def _auth_headers():
|
|
auth_str = f"{AUTH[0]}:{AUTH[1]}"
|
|
b64_auth = base64.b64encode(auth_str.encode('ascii')).decode('ascii')
|
|
return {
|
|
"Authorization": f"Basic {b64_auth}",
|
|
"Content-Type": "application/json",
|
|
"User-Agent": "curl/7.81.0"
|
|
}
|
|
|
|
def get_forgejo_repos():
|
|
repos = []
|
|
page = 1
|
|
while True:
|
|
req = urllib.request.Request(f"{FORGEJO_LOCAL_URL}/user/repos?limit=50&page={page}", headers=_auth_headers())
|
|
try:
|
|
with urllib.request.urlopen(req) as resp:
|
|
if resp.status != 200:
|
|
break
|
|
data = json.loads(resp.read().decode('utf-8'))
|
|
if not data:
|
|
break
|
|
repos.extend(data)
|
|
page += 1
|
|
except Exception as e:
|
|
break
|
|
return {r['name'] for r in repos}
|
|
|
|
def get_gitlab_repos():
|
|
repos = []
|
|
page = 1
|
|
while True:
|
|
req = urllib.request.Request(f"{GITLAB_API}/projects?owned=true&simple=true&per_page=100&page={page}")
|
|
req.add_header("PRIVATE-TOKEN", GITLAB_TOKEN)
|
|
try:
|
|
with urllib.request.urlopen(req) as resp:
|
|
data = json.loads(resp.read().decode('utf-8'))
|
|
if not data:
|
|
break
|
|
repos.extend(data)
|
|
page += 1
|
|
except Exception as e:
|
|
print(f"Failed to fetch GitLab repos: {e}")
|
|
break
|
|
return repos
|
|
|
|
print("Fetching existing repos from forgejo.local...")
|
|
local_repo_names = get_forgejo_repos()
|
|
print(f"Found {len(local_repo_names)} repos on forgejo.local.")
|
|
|
|
print("Fetching repos from GitLab cautiously...")
|
|
gitlab_repos = get_gitlab_repos()
|
|
print(f"Found {len(gitlab_repos)} repos on GitLab.")
|
|
|
|
for repo in gitlab_repos:
|
|
name = repo['path']
|
|
clone_url = repo['http_url_to_repo'].replace("https://", f"https://oauth2:{GITLAB_TOKEN}@")
|
|
|
|
print(f"\n--- Processing GitLab Repo: {name} ---")
|
|
|
|
# Sleep to avoid triggering rate limits / security
|
|
time.sleep(10)
|
|
|
|
# 1. Create on forgejo.local if it doesn't exist
|
|
if name not in local_repo_names:
|
|
print(f"Creating {name} on forgejo.local...")
|
|
req = urllib.request.Request(f"{FORGEJO_LOCAL_URL}/user/repos", data=json.dumps({"name": name, "private": repo.get('visibility') == 'private'}).encode('utf-8'), headers=_auth_headers(), method="POST")
|
|
try:
|
|
urllib.request.urlopen(req)
|
|
local_repo_names.add(name)
|
|
except urllib.error.HTTPError as e:
|
|
if e.code != 409:
|
|
print(f"Failed to create repo {name}: {e.code} {e.reason}")
|
|
continue
|
|
else:
|
|
print(f"{name} already exists on forgejo.local.")
|
|
|
|
# 2. Clone and Push
|
|
repo_dir = os.path.join(BACKUP_DIR, name)
|
|
if os.path.exists(repo_dir):
|
|
shutil.rmtree(repo_dir)
|
|
|
|
print(f"Cloning {name} from GitLab...")
|
|
try:
|
|
subprocess.run(["git", "clone", "--mirror", clone_url, repo_dir], check=True, capture_output=True)
|
|
|
|
print(f"Pushing {name} to forgejo.local...")
|
|
push_url = f"http://{AUTH[0]}:{AUTH[1]}@forgejo.local/mrhavens/{name}.git"
|
|
subprocess.run(["git", "push", "--mirror", push_url], cwd=repo_dir, check=True, capture_output=True)
|
|
print(f"Successfully mirrored {name}!")
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"Error processing {name}: {e}")
|
|
if e.stderr:
|
|
print(e.stderr.decode('utf-8', errors='ignore'))
|
|
|
|
# Clean up to save space
|
|
if os.path.exists(repo_dir):
|
|
shutil.rmtree(repo_dir)
|
|
|
|
print("\nGitLab synchronization complete.")
|