lukeonline - scripts/export-repos.py
#!/usr/bin/env python3
import base64
import json
import shlex
import shutil
import subprocess
import sys
from pathlib import Path
OUTPUT_ROOT = Path("src/content/repos")
def run_ssh(remote, command):
"""Run a shell command on the remote server."""
result = subprocess.run(
["ssh", remote, command],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
if result.returncode != 0:
print(
f"SSH command failed: {command}",
file=sys.stderr,
)
print(result.stderr, file=sys.stderr)
raise RuntimeError("Remote command failed")
return result.stdout
def git(remote, repo, *args):
"""Run git --git-dir=<repo> remotely."""
command = " ".join(
[
"git",
"--git-dir",
shlex.quote(repo),
*[shlex.quote(str(arg)) for arg in args],
]
)
return run_ssh(remote, command)
def get_repo_metadata(remote, repo):
"""Get basic repository metadata."""
head = git(
remote,
repo,
"rev-parse",
"HEAD",
).strip()
default_branch = git(
remote,
repo,
"symbolic-ref",
"HEAD",
).strip().removeprefix("refs/heads/")
branches = git(
remote,
repo,
"for-each-ref",
"--format=%(refname:short)",
"refs/heads/",
).splitlines()
tags = git(
remote,
repo,
"for-each-ref",
"--format=%(refname:short)",
"refs/tags/",
).splitlines()
return {
"head": head,
"defaultBranch": default_branch,
"branches": branches,
"tags": tags,
}
def get_commits(remote, repo):
"""Get all commits in chronological order."""
output = git(
remote,
repo,
"log",
"--reverse",
"--format=%H%x00%P%x00%an%x00%ae%x00%aI%x00%s%x00",
)
commits = []
for record in output.split("\n"):
if not record:
continue
fields = record.split("\x00")
if len(fields) < 6:
continue
sha, parents, author, email, date, message = fields[:6]
commits.append(
{
"sha": sha,
"shortSha": sha[:7],
"parents": parents.split() if parents else [],
"author": {
"name": author,
"email": email,
},
"date": date,
"message": message,
}
)
return commits
def get_commit_diff(remote, repo, commit):
"""Get the complete diff for a commit."""
return git(
remote,
repo,
"show",
"--format=",
"--binary",
commit,
)
def get_tree(remote, repo, commit):
"""Get the complete file tree for a commit."""
result = subprocess.run(
[
"ssh",
remote,
"git",
"--git-dir",
repo,
"ls-tree",
"-r",
"-z",
commit,
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
if result.returncode != 0:
print(result.stderr.decode(), file=sys.stderr)
raise RuntimeError("Could not read repository tree")
files = []
for entry in result.stdout.split(b"\0"):
if not entry:
continue
metadata, path = entry.split(b"\t", 1)
mode, object_type, sha = metadata.decode().split()
files.append(
{
"path": path.decode("utf-8", errors="replace"),
"mode": mode,
"type": object_type,
"sha": sha,
}
)
return files
def get_file_content(remote, repo, commit, path):
"""Get a file's contents from a specific commit."""
command = (
f"git --git-dir {shlex.quote(repo)} "
f"show {shlex.quote(f'{commit}:{path}')}"
)
result = subprocess.run(
["ssh", remote, command],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
if result.returncode != 0:
return None
return result.stdout
def write_json(path, data):
"""Write formatted JSON."""
with open(path, "w", encoding="utf-8") as f:
json.dump(
data,
f,
ensure_ascii=False,
indent=2,
)
f.write("\n")
def export_repository(remote, repo):
"""Export a single repository."""
name = Path(repo).name
output = OUTPUT_ROOT / name
print(f"Exporting {name}...")
output.mkdir(parents=True, exist_ok=True)
(output / "commits").mkdir(exist_ok=True)
(output / "files").mkdir(exist_ok=True)
metadata = get_repo_metadata(remote, repo)
print(f" HEAD: {metadata['head']}")
print(f" Branch: {metadata['defaultBranch']}")
print(f" Branches: {len(metadata['branches'])}")
print(f" Tags: {len(metadata['tags'])}")
commits = get_commits(remote, repo)
print(f" Commits: {len(commits)}")
write_json(
output / "repo.json",
{
"name": name,
**metadata,
"commitCount": len(commits),
},
)
write_json(
output / "commits.json",
commits,
)
for number, commit in enumerate(commits, 1):
sha = commit["sha"]
print(
f" Commit {number}/{len(commits)}: "
f"{commit['shortSha']} {commit['message']}"
)
commit_data = {
**commit,
"diff": get_commit_diff(
remote,
repo,
sha,
),
}
write_json(
output / "commits" / f"{sha}.json",
commit_data,
)
print(" Exporting file tree...")
files = get_tree(
remote,
repo,
metadata["head"],
)
write_json(
output / "tree.json",
files,
)
print(f" Exporting {len(files)} files...")
for number, file_info in enumerate(files, 1):
path = file_info["path"]
print(
f" File {number}/{len(files)}: {path}"
)
content = get_file_content(
remote,
repo,
metadata["head"],
path,
)
if content is None:
continue
try:
text = content.decode("utf-8")
file_data = {
**file_info,
"binary": False,
"content": text,
}
except UnicodeDecodeError:
file_data = {
**file_info,
"binary": True,
"content": base64.b64encode(
content
).decode("ascii"),
}
filename = (
base64.urlsafe_b64encode(
path.encode("utf-8")
)
.decode("ascii")
.rstrip("=")
)
write_json(
output / "files" / f"{filename}.json",
file_data,
)
print(f" Finished {name}")
def main():
if len(sys.argv) != 2:
print(
f"Usage: {sys.argv[0]} user@host:/path/to/repositories",
file=sys.stderr,
)
sys.exit(1)
remote = sys.argv[1].split(':')[0]
remote_root = sys.argv[1].split(':')[1]
print("Checking SSH connection...")
run_ssh(remote, "true")
print("Finding repositories...")
repos = run_ssh(
remote,
(
f"find {shlex.quote(remote_root)} "
"-mindepth 1 "
"-maxdepth 1 "
"-type d "
"-print"
),
).splitlines()
repos = [
repo.strip()
for repo in repos
if repo.strip()
]
if not repos:
print(
"No repositories found.",
file=sys.stderr,
)
sys.exit(1)
print(f"Found {len(repos)} repositories.")
OUTPUT_ROOT.mkdir(
parents=True,
exist_ok=True,
)
current_names = {
Path(repo).name
for repo in repos
}
for existing in OUTPUT_ROOT.iterdir():
if (
existing.is_dir()
and existing.name not in current_names
):
print(
f"Removing old repository: "
f"{existing.name}"
)
shutil.rmtree(existing)
for repo in sorted(repos):
export_repository(
remote,
repo,
)
print()
print("Repository export complete.")
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print(
"\nCancelled.",
file=sys.stderr,
)
sys.exit(130)
except Exception as e:
print(
f"\nERROR: {e}",
file=sys.stderr,
)
sys.exit(1)