github backup script

gh-backup-rolling.py

#!/usr/bin/env python3
"""
GitHub Rolling Backup Script

Creates mirror clones of all repositories from a GitHub organization or user
account, with rolling hardlinked snapshots for space-efficient historical backups.

How it works:
  1. Fetches the full repo list via `gh api` (handles pagination automatically)
  2. Mirror-clones each repo (or updates existing mirrors) into backups/current/
  3. Creates a timestamped hardlinked snapshot in backups/snapshots/
  4. Prunes snapshots beyond the retention limit

Mirror clones (git clone --mirror) preserve the entire git history: all branches,
tags, notes, and refs. Hardlinked snapshots share unchanged files on disk, so
each snapshot costs only the space of objects that changed since the last run.

Requirements:
  - git
  - gh (GitHub CLI, authenticated via `gh auth login`)

Usage:
  python gh-backup-rolling.py --org MY_ORG
  python gh-backup-rolling.py --user
  python gh-backup-rolling.py --org MY_ORG --backup-dir /mnt/backups --keep 30
"""

import argparse
import json
import logging
import os
import shutil
import subprocess
import sys
from datetime import datetime
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
log = logging.getLogger(__name__)


def check_prerequisites():
    """Verify that required tools are available and authenticated."""
    for tool in ("git", "gh"):
        if shutil.which(tool) is None:
            log.error(f"Required tool '{tool}' not found in PATH")
            sys.exit(1)

    result = subprocess.run(
        ["gh", "auth", "status"],
        capture_output=True, text=True,
    )
    if result.returncode != 0:
        log.error("GitHub CLI is not authenticated. Run 'gh auth login' first.")
        sys.exit(1)


def fetch_repos(org=None, include_forks=False, include_archived=False, repo_type="all"):
    """Fetch repository list using gh CLI with automatic pagination.

    Returns a list of dicts with keys: name, clone_url, ssh_url, fork, archived, full_name.
    Handles 200+ repos via GitHub API pagination (100 per page).
    """
    jq_filter = (
        ".[] | {name: .name, clone_url: .clone_url, ssh_url: .ssh_url, "
        "fork: .fork, archived: .archived, full_name: .full_name}"
    )

    if org:
        endpoint = f"/orgs/{org}/repos?per_page=100&type={repo_type}"
    else:
        # GitHub API does not allow both affiliation and type together;
        # use affiliation=owner with visibility to filter if needed
        if repo_type == "all":
            endpoint = "/user/repos?per_page=100&affiliation=owner"
        else:
            endpoint = f"/user/repos?per_page=100&affiliation=owner&visibility={repo_type}"

    cmd = ["gh", "api", "--paginate", endpoint, "--jq", jq_filter]

    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode != 0:
        log.error(f"Failed to fetch repos: {result.stderr}")
        sys.exit(1)

    repos = []
    for line in result.stdout.strip().split("\n"):
        if not line.strip():
            continue
        repo = json.loads(line)
        if not include_forks and repo.get("fork"):
            log.debug(f"  Skipping fork: {repo['full_name']}")
            continue
        if not include_archived and repo.get("archived"):
            log.debug(f"  Skipping archived: {repo['full_name']}")
            continue
        repos.append(repo)

    return repos


def mirror_clone_or_update(clone_url, mirror_dir):
    """Clone a repo as a mirror, or update an existing mirror/clone.

    Mirror clones are bare repos that include every ref on the remote.
    For existing mirrors, `git remote update --prune` fetches new objects
    and removes refs that no longer exist on the remote.

    Returns True on success, False on failure.
    """
    mirror_path = Path(mirror_dir)

    if mirror_path.exists() and (mirror_path / "HEAD").exists():
        # Bare/mirror repo (HEAD is at the top level, no .git subdirectory)
        log.info(f"  Updating mirror: {mirror_path.name}")
        result = subprocess.run(
            ["git", "remote", "update", "--prune"],
            cwd=str(mirror_path),
            capture_output=True, text=True,
        )
        if result.returncode != 0:
            log.warning(f"  Failed to update {mirror_path.name}: {result.stderr.strip()}")
            return False
    elif mirror_path.exists() and (mirror_path / ".git").is_dir():
        # Regular (non-bare) clone — fetch all branches and prune deleted ones
        log.info(f"  Updating clone: {mirror_path.name}")
        result = subprocess.run(
            ["git", "fetch", "--all", "--prune"],
            cwd=str(mirror_path),
            capture_output=True, text=True,
        )
        if result.returncode != 0:
            log.warning(f"  Failed to update {mirror_path.name}: {result.stderr.strip()}")
            return False
    else:
        # Fresh mirror clone — preserves full history, all branches, tags, refs
        log.info(f"  Cloning mirror: {clone_url}")
        if mirror_path.exists():
            shutil.rmtree(str(mirror_path))
        result = subprocess.run(
            ["git", "clone", "--mirror", clone_url, str(mirror_path)],
            capture_output=True, text=True,
        )
        if result.returncode != 0:
            log.warning(f"  Failed to clone {clone_url}: {result.stderr.strip()}")
            return False

    return True


def hardlink_tree(src, dst):
    """Recursively create a hardlinked copy of a directory tree.

    The directory structure is recreated at dst. Regular files are hardlinked
    (not copied), so the snapshot uses almost no additional disk space for
    files that haven't been modified.

    If a hardlink cannot be created (e.g., cross-filesystem), falls back to
    a regular copy for that file.
    """
    src = Path(src)
    dst = Path(dst)

    for dirpath, dirnames, filenames in os.walk(src):
        rel = Path(dirpath).relative_to(src)
        target_dir = dst / rel
        target_dir.mkdir(parents=True, exist_ok=True)

        for fname in filenames:
            src_file = Path(dirpath) / fname
            dst_file = target_dir / fname
            try:
                os.link(str(src_file), str(dst_file))
            except OSError:
                # Cross-device or unsupported — fall back to copy
                shutil.copy2(str(src_file), str(dst_file))


def create_snapshot(current_dir, snapshots_dir, timestamp):
    """Create a timestamped hardlinked snapshot of the current backup state."""
    snapshot_name = timestamp.strftime("%Y-%m-%dT%H%M%S")
    snapshot_path = Path(snapshots_dir) / snapshot_name

    if snapshot_path.exists():
        log.warning(f"Snapshot {snapshot_name} already exists, skipping")
        return snapshot_path

    log.info(f"Creating snapshot: {snapshot_name}")
    hardlink_tree(current_dir, snapshot_path)

    return snapshot_path


def prune_snapshots(snapshots_dir, keep):
    """Remove old snapshots, keeping the most recent `keep` snapshots."""
    snapshots_dir = Path(snapshots_dir)
    if not snapshots_dir.exists():
        return

    snapshots = sorted(
        [d for d in snapshots_dir.iterdir() if d.is_dir()],
        key=lambda d: d.name,
        reverse=True,
    )

    to_remove = snapshots[keep:]
    for old in to_remove:
        log.info(f"Pruning old snapshot: {old.name}")
        shutil.rmtree(str(old))

    if to_remove:
        log.info(f"Pruned {len(to_remove)} old snapshot(s)")


def main():
    parser = argparse.ArgumentParser(
        description="GitHub rolling backup with hardlinked snapshots",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --org my-company
  %(prog)s --org my-company --backup-dir /mnt/backups --keep 60
  %(prog)s --user --ssh --include-forks
  %(prog)s --org my-company --dry-run
  %(prog)s --snapshot-only --backup-dir ./backups
        """,
    )

    source = parser.add_mutually_exclusive_group(required=True)
    source.add_argument("--org", help="GitHub organization name to backup")
    source.add_argument(
        "--user", action="store_true",
        help="Backup the authenticated user's own repositories",
    )

    parser.add_argument(
        "--backup-dir", default="./backups",
        help="Base directory for backups (default: ./backups)",
    )
    parser.add_argument(
        "--keep", type=int, default=30,
        help="Number of snapshots to retain (default: 30)",
    )
    parser.add_argument(
        "--include-forks", action="store_true",
        help="Include forked repositories",
    )
    parser.add_argument(
        "--include-archived", action="store_true",
        help="Include archived repositories",
    )
    parser.add_argument(
        "--type", choices=["all", "public", "private"], default="all",
        dest="repo_type",
        help="Repository type filter (default: all)",
    )
    parser.add_argument(
        "--ssh", action="store_true",
        help="Use SSH URLs instead of HTTPS for cloning",
    )
    parser.add_argument(
        "--dry-run", action="store_true",
        help="List repos that would be backed up without doing anything",
    )
    parser.add_argument(
        "--snapshot-only", action="store_true",
        help="Only create a snapshot of existing backups (skip fetching)",
    )
    parser.add_argument(
        "--no-snapshot", action="store_true",
        help="Only fetch/update repos without creating a snapshot",
    )
    parser.add_argument(
        "-v", "--verbose", action="store_true",
        help="Enable debug logging",
    )

    args = parser.parse_args()

    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)

    check_prerequisites()

    backup_dir = Path(args.backup_dir).resolve()
    current_dir = backup_dir / "current"
    snapshots_dir = backup_dir / "snapshots"

    current_dir.mkdir(parents=True, exist_ok=True)
    snapshots_dir.mkdir(parents=True, exist_ok=True)

    timestamp = datetime.now()

    # ── Fetch and clone/update ──────────────────────────────────────────
    if not args.snapshot_only:
        log.info("Fetching repository list...")
        repos = fetch_repos(
            org=args.org,
            include_forks=args.include_forks,
            include_archived=args.include_archived,
            repo_type=args.repo_type,
        )
        log.info(f"Found {len(repos)} repositories")

        if args.dry_run:
            for repo in repos:
                url = repo["ssh_url"] if args.ssh else repo["clone_url"]
                status = "update" if (current_dir / f"{repo['name']}.git").exists() else "clone"
                print(f"  [{status}] {repo['full_name']}  ({url})")
            return

        success = 0
        failed = 0
        failed_repos = []

        for i, repo in enumerate(repos, 1):
            name = repo["name"]
            url = repo["ssh_url"] if args.ssh else repo["clone_url"]
            mirror_dir = current_dir / f"{name}.git"

            log.info(f"[{i}/{len(repos)}] {repo['full_name']}")
            if mirror_clone_or_update(url, mirror_dir):
                success += 1
            else:
                failed += 1
                failed_repos.append(repo["full_name"])

        log.info(f"Backup complete: {success} succeeded, {failed} failed out of {len(repos)}")
        if failed_repos:
            log.warning("Failed repos:")
            for name in failed_repos:
                log.warning(f"  - {name}")

    # ── Snapshot ────────────────────────────────────────────────────────
    if not args.no_snapshot:
        # Only snapshot if there's something in current/
        if any(current_dir.iterdir()):
            create_snapshot(current_dir, snapshots_dir, timestamp)
            prune_snapshots(snapshots_dir, args.keep)
        else:
            log.warning("No repos in current/ — skipping snapshot")

    log.info("Done.")


if __name__ == "__main__":
    main()