499 lines
18 KiB
Python
499 lines
18 KiB
Python
"""SVN→Git synchronization engine.
|
||
|
||
Builds Git commits from SVN revisions using `git ls-tree`, `git hash-object`,
|
||
`git mktree`, and `git commit-tree` via subprocess. No pygit2 required.
|
||
"""
|
||
|
||
import logging
|
||
import os
|
||
import subprocess
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Tuple
|
||
|
||
from .config import MirrorConfig
|
||
from .mirror import Mirror
|
||
from .gitea import push_to_gitea, GiteaError
|
||
from .git_to_svn import sync_git_to_svn
|
||
from .svn import (
|
||
SVNPathChange,
|
||
get_file as svn_get_file,
|
||
get_latest_revision as svn_get_latest_revision,
|
||
get_log as svn_get_log,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Well-known Git empty tree hash (4b825dc642cb6eb9a060e54bf899d153036d1e3c).
|
||
# Exists in every Git repository implicitly.
|
||
_EMPTY_TREE = "4b825dc642cb6eb9a060e54bf899d153036d1e3c"
|
||
|
||
|
||
class SyncError(Exception):
|
||
pass
|
||
|
||
|
||
# ─── SVN path parsing ────────────────────────────────────────
|
||
|
||
|
||
def parse_svn_path(full_path: str, svn_cfg) -> Tuple[Optional[str], str]:
|
||
"""Split an absolute SVN path into (branch, relative_path).
|
||
|
||
Standard layout examples:
|
||
/trunk/src/main.c → ("trunk", "src/main.c")
|
||
/branches/feat/x.c → ("branches/feat", "x.c")
|
||
/tags/v1.0/README → ("tags/v1.0", "README")
|
||
"""
|
||
path = full_path.lstrip("/")
|
||
|
||
t = svn_cfg.trunk
|
||
b = svn_cfg.branches
|
||
tg = svn_cfg.tags
|
||
|
||
if svn_cfg.layout == "std":
|
||
if path == t:
|
||
return (t, "")
|
||
if path.startswith(t + "/"):
|
||
return (t, path[len(t) + 1:])
|
||
|
||
for prefix in (b, tg):
|
||
if path.startswith(prefix + "/"):
|
||
parts = path.split("/", 2)
|
||
branch = f"{parts[0]}/{parts[1]}"
|
||
rel = parts[2] if len(parts) > 2 else ""
|
||
return (branch, rel)
|
||
|
||
return (None, path)
|
||
|
||
|
||
def group_changes_by_branch(
|
||
changes: List[SVNPathChange], svn_cfg,
|
||
) -> Dict[str, List[Tuple]]:
|
||
"""Group a list of SVN path changes by affected branch.
|
||
|
||
Returns {branch: [(action, kind, rel_path, full_svn_path, ...)]}
|
||
"""
|
||
groups: Dict[str, List[Tuple]] = {}
|
||
for c in changes:
|
||
branch, rel = parse_svn_path(c.path, svn_cfg)
|
||
if branch is None:
|
||
logger.warning("Cannot determine branch for path %s", c.path)
|
||
continue
|
||
groups.setdefault(branch, []).append((
|
||
c.action, c.kind, rel, c.path,
|
||
c.copyfrom_path, c.copyfrom_rev,
|
||
))
|
||
return groups
|
||
|
||
|
||
# ─── Ref / author helpers ────────────────────────────────────
|
||
|
||
|
||
def branch_to_git_ref(branch: str, svn_cfg) -> str:
|
||
"""Map SVN branch name to Git ref.
|
||
|
||
trunk → refs/heads/master
|
||
branches/X → refs/heads/X
|
||
tags/X → refs/tags/X
|
||
"""
|
||
if branch == svn_cfg.trunk:
|
||
return "refs/heads/master"
|
||
if branch.startswith(svn_cfg.tags + "/") and len(branch) > len(svn_cfg.tags) + 1:
|
||
return "refs/tags/" + branch[len(svn_cfg.tags) + 1:]
|
||
if branch.startswith(svn_cfg.branches + "/") and len(branch) > len(svn_cfg.branches) + 1:
|
||
return "refs/heads/" + branch[len(svn_cfg.branches) + 1:]
|
||
return "refs/heads/" + branch
|
||
|
||
|
||
def map_author(svn_username: str, authors: Dict[str, str]) -> Tuple[str, str]:
|
||
"""Map SVN username to Git (name, email)."""
|
||
mapped = authors.get(svn_username)
|
||
if mapped:
|
||
# Format: "John Doe <john@example.com>"
|
||
if "<" in mapped and mapped.endswith(">"):
|
||
name = mapped[:mapped.index("<")].strip()
|
||
email = mapped[mapped.index("<") + 1:-1].strip()
|
||
return (name, email)
|
||
return (mapped, f"{svn_username}@local")
|
||
return (svn_username, f"{svn_username}@local")
|
||
|
||
|
||
# ─── Git tree building ────────────────────────────────────────
|
||
|
||
|
||
def _git(git_dir: str, *args: str, input: str = None, timeout: int = 120,
|
||
check: bool = True) -> str:
|
||
"""Run a git command in the bare canonical repo and return stdout."""
|
||
try:
|
||
r = subprocess.run(
|
||
["git", "--git-dir", git_dir, *args],
|
||
input=input,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=timeout,
|
||
)
|
||
except FileNotFoundError:
|
||
raise SyncError("git not found on PATH")
|
||
except subprocess.TimeoutExpired:
|
||
raise SyncError(f"git command timed out: git {' '.join(args)}")
|
||
|
||
if check and r.returncode != 0:
|
||
raise SyncError(
|
||
f"git command failed (exit {r.returncode}): "
|
||
f"git {' '.join(args)}\n{r.stderr.strip()}"
|
||
)
|
||
return r.stdout.strip() if r.returncode == 0 else ""
|
||
|
||
|
||
def _git_bytes(git_dir: str, *args: str, input: bytes = None,
|
||
timeout: int = 120) -> str:
|
||
"""Like _git but accepts bytes input (for binary file content).
|
||
|
||
``text=True`` is omitted so subprocess passes raw bytes on stdin.
|
||
"""
|
||
try:
|
||
r = subprocess.run(
|
||
["git", "--git-dir", git_dir, *args],
|
||
input=input,
|
||
capture_output=True,
|
||
timeout=timeout,
|
||
)
|
||
except FileNotFoundError:
|
||
raise SyncError("git not found on PATH")
|
||
except subprocess.TimeoutExpired:
|
||
raise SyncError(f"git command timed out: git {' '.join(args)}")
|
||
|
||
if r.returncode != 0:
|
||
raise SyncError(
|
||
f"git command failed (exit {r.returncode}): "
|
||
f"git {' '.join(args)}\n{r.stderr.strip()}"
|
||
)
|
||
return r.stdout.strip().decode("utf-8", errors="replace")
|
||
|
||
|
||
def _build_tree_from_changes(
|
||
git_dir: str,
|
||
parent_hash: Optional[str],
|
||
changes: List[Tuple],
|
||
svn_url: str,
|
||
revision: int,
|
||
auth_args: Optional[List[str]] = None,
|
||
) -> str:
|
||
"""Build a new Git tree by applying SVN changes to the parent tree.
|
||
|
||
Strategy:
|
||
1. Get flat file listing from parent tree via `git ls-tree -r`.
|
||
2. Apply add/modify/delete operations to the in-memory dict.
|
||
3. Write new tree via `git mktree`.
|
||
|
||
This avoids the complexity of nested TreeBuilders and handles
|
||
directory deletions simply (all files under the path are removed).
|
||
"""
|
||
# ── 1. Read parent tree into flat dict ─────────────────────
|
||
entries: Dict[str, Tuple[str, str]] = {} # path → (mode, oid)
|
||
if parent_hash:
|
||
out = _git(git_dir, "ls-tree", "-r", parent_hash, timeout=60)
|
||
for line in out.splitlines():
|
||
if not line:
|
||
continue
|
||
meta, path = line.split("\t", 1)
|
||
mode, obj_type, oid = meta.split(None, 2)
|
||
entries[path] = (mode, oid)
|
||
|
||
# ── 2. Apply changes ──────────────────────────────────────
|
||
for action, kind, rel_path, full_path, cf_path, cf_rev in changes:
|
||
if action in ("A", "M") and kind == "file":
|
||
content = svn_get_file(svn_url, full_path, revision, auth_args=auth_args)
|
||
blob_oid = _git_bytes(git_dir, "hash-object", "-w", "--stdin",
|
||
input=content, timeout=60)
|
||
entries[rel_path] = ("100644", blob_oid)
|
||
|
||
elif action == "D" and kind == "file":
|
||
entries.pop(rel_path, None)
|
||
|
||
elif action == "D" and kind == "dir":
|
||
prefix = rel_path.rstrip("/") + "/"
|
||
to_del = [k for k in entries if k == rel_path or k.startswith(prefix)]
|
||
for k in to_del:
|
||
del entries[k]
|
||
|
||
elif action == "R" and kind == "file":
|
||
entries.pop(rel_path, None)
|
||
content = svn_get_file(svn_url, full_path, revision, auth_args=auth_args)
|
||
blob_oid = _git_bytes(git_dir, "hash-object", "-w", "--stdin",
|
||
input=content, timeout=60)
|
||
entries[rel_path] = ("100644", blob_oid)
|
||
|
||
# ── 3. Write new tree via recursive mktree ─────────────────
|
||
return _tree_from_flat_entries(git_dir, entries)
|
||
|
||
|
||
def _tree_from_flat_entries(git_dir: str,
|
||
entries: Dict[str, Tuple[str, str]]) -> str:
|
||
"""Build a Git tree from a flat ``{path: (mode, oid)}`` dict.
|
||
|
||
``git mktree`` only accepts entries for a *single* level (no slashes
|
||
in paths), so we group entries by their first path component and
|
||
recurse into sub-trees.
|
||
"""
|
||
if not entries:
|
||
return _EMPTY_TREE
|
||
|
||
lines: List[str] = []
|
||
children: Dict[str, Dict[str, Tuple[str, str]]] = {}
|
||
|
||
for path, (mode, oid) in sorted(entries.items()):
|
||
idx = path.find("/")
|
||
if idx == -1:
|
||
lines.append(f"{mode} blob {oid}\t{path}")
|
||
else:
|
||
head, tail = path[:idx], path[idx + 1:]
|
||
children.setdefault(head, {})[tail] = (mode, oid)
|
||
|
||
for dir_name in sorted(children):
|
||
subtree = _tree_from_flat_entries(git_dir, children[dir_name])
|
||
if subtree != _EMPTY_TREE:
|
||
lines.append(f"040000 tree {subtree}\t{dir_name}")
|
||
|
||
if not lines:
|
||
return _EMPTY_TREE
|
||
|
||
input_str = "\n".join(lines)
|
||
return _git(git_dir, "mktree", input=input_str, timeout=60)
|
||
|
||
|
||
def _find_copy_source(cfg, db, changes) -> Optional[str]:
|
||
"""If *changes* includes an ``svn copy``, return the source Git commit hash.
|
||
|
||
When SVN creates a branch or tag via ``svn copy src dst`` the change
|
||
is recorded as ``(A, dir, …, copyfrom_path, copyfrom_rev)``. We
|
||
look up the source commit in the mapping DB so the new branch starts
|
||
with the correct tree.
|
||
|
||
Falls back to the latest mapping for the source branch when the exact
|
||
``(cf_rev, src_branch)`` pair is not found (common when the copyfrom
|
||
revision is a global revision that didn't touch the source branch).
|
||
"""
|
||
for c in changes:
|
||
if len(c) < 6:
|
||
continue
|
||
action, kind, _rel, _full, cf_path, cf_rev = c[:6]
|
||
if action == "A" and kind == "dir" and cf_path and cf_rev:
|
||
src_branch, _ = parse_svn_path(cf_path, cfg.svn)
|
||
if not src_branch:
|
||
continue
|
||
|
||
# 1. Try exact match first
|
||
src_hash = db.get_git_hash(cf_rev, src_branch)
|
||
if src_hash:
|
||
logger.debug(
|
||
"Copy source for %s@%d → %s (%s)",
|
||
cf_path, cf_rev, src_hash[:8], src_branch,
|
||
)
|
||
return src_hash
|
||
|
||
# 2. Fallback: latest mapping for the source branch <= cf_rev
|
||
row = db.conn.execute(
|
||
"SELECT git_commit_hash FROM svn_to_git "
|
||
"WHERE svn_branch=? AND svn_revision<=? "
|
||
"ORDER BY svn_revision DESC LIMIT 1",
|
||
(src_branch, cf_rev),
|
||
).fetchone()
|
||
if row:
|
||
logger.debug(
|
||
"Copy source for %s@%d → %s (%s, fallback)",
|
||
cf_path, cf_rev, row[0][:8], src_branch,
|
||
)
|
||
return row[0]
|
||
|
||
return None
|
||
|
||
|
||
# ─── Per-revision sync ────────────────────────────────────────
|
||
|
||
|
||
def sync_svn_revision(mirror: Mirror, revision: int,
|
||
auth_args: Optional[List[str]] = None) -> bool:
|
||
"""Translate a single SVN revision into Git commit(s).
|
||
|
||
Returns True if at least one Git commit was created.
|
||
"""
|
||
cfg = mirror.config
|
||
git_dir = str(mirror.canonical_dir)
|
||
|
||
# Fetch SVN metadata
|
||
log_entry = svn_get_log(cfg.svn.url, revision, auth_args=auth_args)
|
||
branch_changes = group_changes_by_branch(log_entry.paths, cfg.svn)
|
||
|
||
if not branch_changes:
|
||
logger.debug("Rev %d: no branch changes to sync", revision)
|
||
return False
|
||
|
||
any_commit = False
|
||
|
||
for branch, changes in branch_changes.items():
|
||
git_ref = branch_to_git_ref(branch, cfg.svn)
|
||
|
||
if mirror.db.mapping_exists(revision, branch):
|
||
logger.debug("Rev %d/%s already mapped – skipping", revision, branch)
|
||
continue
|
||
|
||
# Find parent commit for this branch
|
||
parent_hash_str = _git(git_dir, "rev-parse", "--verify", "--quiet",
|
||
git_ref, timeout=30, check=False)
|
||
parent_hash = parent_hash_str if parent_hash_str else None
|
||
|
||
# For a new branch/tag created via svn copy, use the source tree
|
||
if parent_hash is None:
|
||
parent_hash = _find_copy_source(cfg, mirror.db, changes)
|
||
|
||
# Build new tree
|
||
new_tree = _build_tree_from_changes(
|
||
git_dir, parent_hash, changes, cfg.svn.url, revision,
|
||
auth_args=auth_args,
|
||
)
|
||
|
||
# Map author
|
||
author_name, author_email = map_author(log_entry.author, cfg.authors)
|
||
author_line = f"{author_name} <{author_email}>"
|
||
|
||
# Create commit via commit-tree
|
||
c_args = ["commit-tree", new_tree]
|
||
if parent_hash:
|
||
c_args += ["-p", parent_hash]
|
||
|
||
commit_env = {
|
||
"GIT_AUTHOR_NAME": author_name,
|
||
"GIT_AUTHOR_EMAIL": author_email,
|
||
"GIT_COMMITTER_NAME": author_name,
|
||
"GIT_COMMITTER_EMAIL": author_email,
|
||
}
|
||
|
||
commit_hash = _git_env(git_dir, commit_env, *c_args,
|
||
input=log_entry.message, timeout=120)
|
||
|
||
# Update ref
|
||
_git(git_dir, "update-ref", git_ref, commit_hash, timeout=30)
|
||
|
||
# Record mapping
|
||
mirror.db.record_mapping(revision, commit_hash, branch, git_ref, "svn")
|
||
any_commit = True
|
||
|
||
logger.info(
|
||
"Synced SVN r%d → %s (%s / %s)",
|
||
revision, commit_hash[:8], branch, git_ref,
|
||
)
|
||
|
||
return any_commit
|
||
|
||
|
||
def _git_env(git_dir: str, env: Dict[str, str], *args: str,
|
||
input: str = None, timeout: int = 120) -> str:
|
||
"""Like _git but with additional environment variables (for commit authorship)."""
|
||
full_env = {**os.environ, "GIT_DIR": git_dir, **env}
|
||
try:
|
||
r = subprocess.run(
|
||
["git", *args],
|
||
input=input,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=timeout,
|
||
env=full_env,
|
||
)
|
||
except FileNotFoundError:
|
||
raise SyncError("git not found on PATH")
|
||
except subprocess.TimeoutExpired:
|
||
raise SyncError(f"git command timed out: git {' '.join(args)}")
|
||
|
||
if r.returncode != 0:
|
||
raise SyncError(
|
||
f"git command failed (exit {r.returncode}): "
|
||
f"git {' '.join(args)}\n{r.stderr.strip()}"
|
||
)
|
||
return r.stdout.strip()
|
||
|
||
|
||
# ─── Full sync cycle ──────────────────────────────────────────
|
||
|
||
|
||
def sync_all(mirror: Mirror) -> int:
|
||
"""Sync all pending SVN revisions for a single mirror (thread-safe).
|
||
|
||
Returns the number of revisions processed.
|
||
"""
|
||
with mirror.sync_lock():
|
||
return _sync_all_impl(mirror)
|
||
|
||
|
||
def _sync_all_impl(mirror: Mirror) -> int:
|
||
auth_args = mirror.config.svn.auth_args()
|
||
state_rev = mirror.db.get_state("last_svn_revision")
|
||
trunk_name = mirror.config.svn.trunk
|
||
last_rev = int(state_rev) if state_rev else (mirror.db.get_last_svn_revision(trunk_name) or 0)
|
||
latest_rev = svn_get_latest_revision(mirror.config.svn.url, auth_args=auth_args)
|
||
|
||
if latest_rev <= last_rev:
|
||
logger.debug("Mirror %s is up to date (r%d)", mirror.config.id, last_rev)
|
||
return 0
|
||
|
||
logger.info(
|
||
"Syncing %s SVN r%d → r%d",
|
||
mirror.config.id, last_rev + 1, latest_rev,
|
||
)
|
||
|
||
count = 0
|
||
for rev in range(last_rev + 1, latest_rev + 1):
|
||
try:
|
||
if sync_svn_revision(mirror, rev, auth_args=auth_args):
|
||
count += 1
|
||
except Exception:
|
||
logger.exception("Failed to sync SVN r%d for %s", rev, mirror.config.id)
|
||
raise
|
||
|
||
mirror.db.set_state("last_svn_revision", str(latest_rev))
|
||
logger.info("Synced %d revision(s) for %s", count, mirror.config.id)
|
||
return count
|
||
|
||
|
||
# ─── Continuous daemon loop ───────────────────────────────────
|
||
|
||
|
||
def run_daemon_loop(mirrors: Dict[str, Mirror]) -> None:
|
||
"""Run a continuous sync loop, checking each mirror on its own interval."""
|
||
last_sync: Dict[str, float] = {mid: 0.0 for mid in mirrors}
|
||
|
||
logger.info("Starting sync daemon with %d mirror(s)", len(mirrors))
|
||
try:
|
||
while True:
|
||
now = time.time()
|
||
for mid, m in mirrors.items():
|
||
interval = m.config.sync_interval
|
||
if now - last_sync[mid] < interval:
|
||
continue
|
||
|
||
if not m.base_dir.exists():
|
||
logger.debug("Mirror %s not yet created – skipping", mid)
|
||
continue
|
||
|
||
state = m._read_state()
|
||
if not state.get("initial_import_done"):
|
||
logger.debug("Mirror %s not yet imported – skipping", mid)
|
||
continue
|
||
|
||
logger.debug("Checking mirror %s…", mid)
|
||
try:
|
||
svn_count = sync_all(m)
|
||
if svn_count:
|
||
push_to_gitea(m)
|
||
git_count = sync_git_to_svn(m)
|
||
if svn_count or git_count:
|
||
last_sync[mid] = now
|
||
except (GiteaError) as ge:
|
||
logger.warning("Gitea push failed for %s: %s", mid, ge)
|
||
except Exception:
|
||
logger.exception("Sync cycle failed for %s", mid)
|
||
|
||
time.sleep(5)
|
||
except KeyboardInterrupt:
|
||
logger.info("Daemon stopped by user")
|