feat(git-export): add script for efficient directory export from git repos

Author: tmeckel

git-export.py

@@ -0,0 +1,328 @@
+#!/usr/bin/env python3
+"""
+Export a directory from a Git repository using treeless + sparse clone.
+
+Workflow:
+1. Clone repository with --filter=tree:0 --sparse --no-checkout.
+2. Configure sparse-checkout for the requested directory.
+3. Checkout the requested ref (or default branch).
+4. Copy only the requested directory contents to output.
+5. Always remove .git from the output.
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import time
+import urllib.parse
+from pathlib import Path
+
+
+class GitExportError(RuntimeError):
+    """Domain error for export failures."""
+
+
+def info(message: str) -> None:
+    print(f"[git-export] {message}", flush=True)
+
+
+def run_git(git_bin: str, args: list[str], cwd: Path | None = None, verbose: bool = False) -> None:
+    cmd = [git_bin, *args]
+    if verbose:
+        location = str(cwd) if cwd else os.getcwd()
+        print(f"+ (cwd={location}) {' '.join(cmd)}")
+    try:
+        subprocess.run(cmd, cwd=str(cwd) if cwd else None, check=True, text=True, capture_output=True)
+    except subprocess.CalledProcessError as e:
+        stderr = (e.stderr or "").strip()
+        stdout = (e.stdout or "").strip()
+        detail = stderr or stdout or str(e)
+        raise GitExportError(f"git command failed: {' '.join(cmd)}\n{detail}") from e
+
+
+def normalize_source_path(path: str) -> str:
+    source = path.strip().strip("/")
+    if not source:
+        raise GitExportError("Source path must not be empty")
+    parts = [p for p in source.split("/") if p not in ("", ".")]
+    if any(part == ".." for part in parts):
+        raise GitExportError("Source path must not contain '..'")
+    return "/".join(parts)
+
+
+def parse_github_directory_url(url: str) -> tuple[str, str, str | None]:
+    """
+    Parse a GitHub directory URL into (repo_url, source_path, ref).
+
+    Supported examples:
+    - https://github.com/org/repo/lang/ruby
+    - https://github.com/org/repo/tree/main/lang/ruby
+    - https://github.com/org/repo/blob/main/lang/ruby
+    """
+    parsed = urllib.parse.urlparse(url)
+    if parsed.scheme not in ("http", "https") or parsed.netloc not in ("github.com", "www.github.com"):
+        raise GitExportError(f"Not a supported GitHub URL: {url}")
+
+    parts = [p for p in parsed.path.split("/") if p]
+    if len(parts) < 3:
+        raise GitExportError(
+            "GitHub URL must include a directory path after owner/repo "
+            f"(got: {url})"
+        )
+
+    owner = parts[0]
+    repo = parts[1]
+    if repo.endswith(".git"):
+        repo = repo[:-4]
+
+    rest = parts[2:]
+    ref: str | None = None
+    source: str
+
+    if rest[0] in ("tree", "blob"):
+        if len(rest) < 3:
+            raise GitExportError(
+                "tree/blob URLs must include ref and directory path, "
+                f"got: {url}"
+            )
+        ref = rest[1]
+        source = "/".join(rest[2:])
+    else:
+        source = "/".join(rest)
+
+    repo_url = f"https://github.com/{owner}/{repo}.git"
+    return repo_url, normalize_source_path(source), ref
+
+
+def prepare_output_dir(output_dir: Path, force: bool) -> None:
+    if output_dir.exists():
+        if not force:
+            raise GitExportError(
+                f"Output path already exists: {output_dir} (use --force to overwrite)"
+            )
+        if output_dir.is_file() or output_dir.is_symlink():
+            output_dir.unlink()
+        else:
+            shutil.rmtree(output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+
+def copy_entry(src: Path, dst: Path) -> None:
+    if src.is_symlink():
+        target = os.readlink(src)
+        if dst.exists() or dst.is_symlink():
+            if dst.is_dir() and not dst.is_symlink():
+                shutil.rmtree(dst)
+            else:
+                dst.unlink()
+        os.symlink(target, dst)
+        return
+    if src.is_dir():
+        shutil.copytree(src, dst, symlinks=True, dirs_exist_ok=True)
+        return
+    shutil.copy2(src, dst, follow_symlinks=False)
+
+
+def export_directory(
+    repo_url: str,
+    source_path: str,
+    output_dir: Path,
+    ref: str | None,
+    depth: int,
+    force: bool,
+    git_bin: str,
+    verbose: bool,
+) -> None:
+    start_total = time.perf_counter()
+    source_path = normalize_source_path(source_path)
+    output_dir = output_dir.resolve()
+
+    info(f"Repository: {repo_url}")
+    info(f"Source path: {source_path}")
+    info(f"Ref: {ref or 'default branch'}")
+    info(f"Output: {output_dir}")
+
+    with tempfile.TemporaryDirectory(prefix="git-export-") as temp_dir:
+        work_dir = Path(temp_dir)
+        clone_dir = work_dir / "repo"
+
+        info("Step 1/6: cloning repository (treeless + sparse, no checkout)")
+        step_start = time.perf_counter()
+        run_git(
+            git_bin,
+            [
+                "clone",
+                "--depth",
+                str(depth),
+                "--filter=tree:0",
+                "--sparse",
+                "--no-checkout",
+                repo_url,
+                str(clone_dir),
+            ],
+            verbose=verbose,
+        )
+        info(f"Step 1/6 complete in {time.perf_counter() - step_start:.1f}s")
+
+        info("Step 2/6: configuring sparse checkout")
+        step_start = time.perf_counter()
+        run_git(git_bin, ["sparse-checkout", "init", "--cone"], cwd=clone_dir, verbose=verbose)
+        run_git(git_bin, ["sparse-checkout", "set", "--", source_path], cwd=clone_dir, verbose=verbose)
+        info(f"Step 2/6 complete in {time.perf_counter() - step_start:.1f}s")
+
+        info("Step 3/6: checking out requested ref/path")
+        step_start = time.perf_counter()
+        if ref:
+            run_git(
+                git_bin,
+                ["fetch", "--depth", str(depth), "origin", ref],
+                cwd=clone_dir,
+                verbose=verbose,
+            )
+            run_git(git_bin, ["checkout", "--detach", "FETCH_HEAD"], cwd=clone_dir, verbose=verbose)
+        else:
+            run_git(git_bin, ["checkout"], cwd=clone_dir, verbose=verbose)
+        info(f"Step 3/6 complete in {time.perf_counter() - step_start:.1f}s")
+
+        info("Step 4/6: validating source directory")
+        source_dir = clone_dir / source_path
+        if not source_dir.exists() or not source_dir.is_dir():
+            raise GitExportError(
+                f"Source directory not found after checkout: {source_path}\n"
+                f"Repository: {repo_url}\n"
+                f"Ref: {ref or 'default branch'}"
+            )
+        info("Step 4/6 complete")
+
+        info("Step 5/6: preparing output directory")
+        step_start = time.perf_counter()
+        prepare_output_dir(output_dir, force=force)
+        info(f"Step 5/6 complete in {time.perf_counter() - step_start:.1f}s")
+
+        info("Step 6/6: copying exported files")
+        step_start = time.perf_counter()
+        children = list(source_dir.iterdir())
+        total_children = len(children)
+        if total_children == 0:
+            info("Source directory is empty")
+        for idx, child in enumerate(children, start=1):
+            info(f"  - [{idx}/{total_children}] {child.name}")
+            copy_entry(child, output_dir / child.name)
+        info(f"Step 6/6 complete in {time.perf_counter() - step_start:.1f}s")
+
+    # Explicitly ensure .git is never left in output.
+    info("Finalizing export (removing .git if present)")
+    shutil.rmtree(output_dir / ".git", ignore_errors=True)
+    info(f"All done in {time.perf_counter() - start_total:.1f}s")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Export one directory from a huge Git repository using treeless + sparse clone."
+        )
+    )
+    parser.add_argument(
+        "input",
+        help=(
+            "Either a repository URL (legacy mode) or a full GitHub directory URL, "
+            "e.g. https://github.com/apache/avro/lang/ruby"
+        ),
+    )
+    parser.add_argument(
+        "arg2",
+        help=(
+            "In URL mode: destination output directory. "
+            "In legacy mode: source directory path."
+        ),
+    )
+    parser.add_argument(
+        "arg3",
+        nargs="?",
+        help="Legacy mode only: destination output directory.",
+    )
+    parser.add_argument(
+        "--source",
+        help=(
+            "Source directory path when using 2-arg mode with a repository URL input."
+        ),
+    )
+    parser.add_argument(
+        "--ref",
+        "-r",
+        help="Branch/tag/ref to export (default: repository default branch)",
+    )
+    parser.add_argument(
+        "--depth",
+        type=int,
+        default=1,
+        help="Fetch depth for clone/fetch (default: 1)",
+    )
+    parser.add_argument(
+        "--force",
+        "-f",
+        action="store_true",
+        help="Overwrite output directory if it already exists",
+    )
+    parser.add_argument(
+        "--git-bin",
+        default="git",
+        help="Git binary path/name (default: git)",
+    )
+    parser.add_argument(
+        "--verbose",
+        "-v",
+        action="store_true",
+        help="Print git commands while running",
+    )
+
+    args = parser.parse_args()
+
+    if args.depth < 1:
+        print("Error: --depth must be >= 1", file=sys.stderr)
+        return 2
+
+    try:
+        parsed_ref: str | None = None
+        if args.arg3 is not None:
+            # Legacy mode: repo source output
+            repo_url = args.input
+            source_path = args.arg2
+            output_path = args.arg3
+        else:
+            # URL mode: input output
+            output_path = args.arg2
+            if args.source:
+                repo_url = args.input
+                source_path = args.source
+            else:
+                repo_url, source_path, parsed_ref = parse_github_directory_url(args.input)
+
+        export_directory(
+            repo_url=repo_url,
+            source_path=source_path,
+            output_dir=Path(output_path),
+            ref=args.ref or parsed_ref,
+            depth=args.depth,
+            force=args.force,
+            git_bin=args.git_bin,
+            verbose=args.verbose,
+        )
+    except GitExportError as e:
+        print(f"Error: {e}", file=sys.stderr)
+        return 1
+    except FileNotFoundError as e:
+        print(f"Error: unable to execute git binary '{args.git_bin}': {e}", file=sys.stderr)
+        return 1
+
+    print(f"Export complete: {Path(output_path).resolve()}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())