Github Mass Clone Script

gistfile1.txt

#!/usr/bin/env -S uv run --script
# /// script
# dependencies = [
#   "PyGitHub",
#   "GitPython"
# ]
# ///

import argparse
import subprocess
import sys
import logging
from pathlib import Path
import getpass

import git

from github import Github
from github import Auth

logger = logging.getLogger()

# gh_cmd="$1"
# shift
# orgname="$1"

# if [ -z "$orgname" ]; then
#     fatal "Must supply a gh command to use"
# fi

# if [ -z "$orgname" ]; then
#     fatal "Must supply an organization name"
# fi

# $gh_cmd repo list "$orgname" --limit 1000 | while read -r repo _; do
#     log "$repo"
#     $gh_cmd repo clone "$repo" "$repo" -- -q 2>/dev/null || (
#         log "Updating $repo"
#         if ! cd "$repo" ; then
#             log "Error: could not change directory to $repo"
#             exit 1
#         fi
#         # Handle case where local checkout is on a non-main/master branch
#         # - ignore checkout errors because some repos may have zero commits, 
#         # so no main or master
#         git checkout -q main 2>/dev/null || true
#         git checkout -q master 2>/dev/null || true
#         git pull -q
#     )
# done

def _token_for_host(host):
  # Hilariously this is easier then trying to talk to secret manager directly.
  output = subprocess.check_output(["secret-tool", "search", "server", host, "user", getpass.getuser()], 
                                 stderr=subprocess.DEVNULL,
                                 encoding="utf8")
  token = [line for line in output.split("\n") if line.startswith("secret = ")][0].split("secret = ")[1]
  return token

parser = argparse.ArgumentParser("Github Mass Clone Tool")

parser.add_argument("--log-level", default="info", help="Log Level")

parser.add_argument("--host", required=True, help="Github host to clone from")
parser.add_argument("--token", help="Authentication token")
parser.add_argument("--anonymous", action="store_true", help="Run clone anonymously (--token is ignored)")
parser.add_argument("--organization", help="Organization or user to clone")

def main(argv):
  args = parser.parse_args(argv)
  logging.basicConfig(
    stream=sys.stderr,
    level=getattr(logging, args.log_level.upper()),

  )

  host = args.host

  if args.token is None and not args.anonymous:
    token = _token_for_host(args.host)
  else:
    token = args.token
  
  orgname = args.organization
  if orgname is None:
    raise ValueError("must specify an organization")
  
  if not args.anonymous:
    logger.info("Authenticated Checkout From %s: %s", host, orgname)
    auth = Auth.Token(token) if not args.anonymous else None
  else:
    logger.info("Anonymous Checkout From %s: %s", host, orgname)
    auth = None

  if host == "github":
    g = Github(auth=auth)
  else:
    g = Github(base_url=f"https://{host}/api/v3", auth=auth)

  logger.debug("Get organization: %s", orgname)
  org = g.get_organization(orgname)

  update_candidates = []

  for repo in org.get_repos():
    logger.info("Processing: %s", repo.full_name)
    
    repo_dir = Path(repo.full_name)
    repo_parent = repo_dir.parent

    logger.debug("Directory will be: %s", repo_dir.as_posix())
    logger.debug("Clone directory will be: %s", repo_parent.as_posix())

    if not repo_parent.exists():
      logger.debug("Making directory: %s", repo_parent.as_posix())
      repo_parent.mkdir(exist_ok=True)

    try:
      subprocess.run(["git", "clone", repo.clone_url], cwd=repo_parent.absolute().as_posix(),check=True,
                     stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
      logger.info("Cloned: %s", repo.full_name)
    except subprocess.CalledProcessError:
      logger.debug("Deferring: %s", repo.full_name)
      update_candidates.append(repo)

  logger.info("Updating existing repos")
  for repo in update_candidates:
    logger.info("Updating: %s", repo.full_name)
      
    repo_obj = git.Repo(repo_dir.absolute())
    main_branch = "main" if "main" in repo_obj.branches else "master"
    repo_obj.branches[main_branch].checkout()
    
    repo_obj.remotes["origin"].pull()

if __name__ == "__main__":
  main(sys.argv[1:])