push_to_mirror.py

import json
import os
import subprocess
import sys
from pathlib import Path
from urllib.parse import urlparse, urlunparse

import requests
import jsonschema
import yaml

CONFIG_FILE = Path("config.yaml")
SCHEMA_FILE = Path("schema.json")
CLONE_BASE_DIR = Path("repos")
CODEBERG_API = "https://codeberg.org/api/v1"


def load_config(config_path: Path) -> dict:
    with config_path.open() as f:
        return yaml.safe_load(f)


def load_schema(schema_path: Path) -> dict:
    with schema_path.open() as f:
        return json.load(f)


def validate_config(config: dict, schema: dict) -> None:
    jsonschema.validate(instance=config, schema=schema)


def clone_from_github(url: str, dest_dir: Path) -> None:
    token = os.environ.get("GITHUB_TOKEN")
    if token:
        parsed = urlparse(url)
        clone_url = urlunparse(
            parsed._replace(netloc=f"oauth2:{token}@{parsed.hostname}")
        )
    else:
        print("  GITHUB_TOKEN not set, attempting unauthenticated clone")
        clone_url = url
    print(f"  cloning {url} -> {dest_dir}")
    subprocess.run(["git", "clone", clone_url, str(dest_dir)], check=True)


def ensure_codeberg_repo(destination: str) -> None:
    token = os.environ.get("CODEBERG_TOKEN")
    if not token:
        raise RuntimeError("CODEBERG_TOKEN environment variable is not set")

    _, owner, repo = urlparse(destination).path.rstrip("/").split("/", 2)
    headers = {"Authorization": f"token {token}", "Content-Type": "application/json"}

    r = requests.get(f"{CODEBERG_API}/repos/{owner}/{repo}", headers=headers)
    if r.status_code == 200:
        return

    payload = {"name": repo, "private": False, "auto_init": False}
    r = requests.post(f"{CODEBERG_API}/orgs/{owner}/repos", json=payload, headers=headers)
    r.raise_for_status()
    print(f"  created {destination}")


def push_to_codeberg(repo_dir: Path, destination: str) -> None:
    token = os.environ.get("CODEBERG_TOKEN")
    if not token:
        raise RuntimeError("CODEBERG_TOKEN environment variable is not set")
    parsed = urlparse(destination)
    authed_url = urlunparse(parsed._replace(netloc=f"oauth2:{token}@{parsed.hostname}"))
    print(f"  pushing to {destination}")
    subprocess.run(
        ["git", "-C", str(repo_dir), "push", "--mirror", authed_url], check=True
    )


def clone_repository(source: str, dest_dir: Path) -> None:
    if dest_dir.exists():
        print(f"  already cloned, skipping: {dest_dir}")
        return
    clone_from_github(source, dest_dir)


def main():
    schema = load_schema(SCHEMA_FILE)
    config = load_config(CONFIG_FILE)

    try:
        validate_config(config, schema)
    except jsonschema.ValidationError as e:
        print(f"Invalid configuration: {e.message}", file=sys.stderr)
        sys.exit(1)

    repositories = config["repositories"]
    CLONE_BASE_DIR.mkdir(exist_ok=True)
    for repo in repositories:
        source = repo["source"]
        destination = repo["destination"]
        dest_dir = CLONE_BASE_DIR / urlparse(source).path.strip("/").replace("/", "_")

        print(f"Syncing {source} -> {destination}")
        ensure_codeberg_repo(destination)
        clone_repository(source, dest_dir)
        push_to_codeberg(dest_dir, destination)


if __name__ == "__main__":
    main()