svn-git-server/svn_mirror/svn.py

"""SVN repository remote interface (via svn CLI)."""

import subprocess
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
from typing import List, Optional


class SVNError(Exception):
    pass


@dataclass
class SVNPathChange:
    action: str  # 'A', 'M', 'D', 'R'
    kind: str  # 'file', 'dir'
    path: str  # absolute SVN path, e.g. /trunk/src/main.c
    copyfrom_path: Optional[str] = None
    copyfrom_rev: Optional[int] = None


@dataclass
class SVNLogEntry:
    revision: int
    author: str
    date: str
    message: str
    paths: List[SVNPathChange] = field(default_factory=list)


def _run_svn(args: list, input: bytes = None, timeout: int = 120) -> subprocess.CompletedProcess:
    try:
        r = subprocess.run(
            ["svn"] + args,
            input=input,
            capture_output=True,
            timeout=timeout,
        )
    except FileNotFoundError:
        raise SVNError("svn CLI not found – is subversion installed?")
    except subprocess.TimeoutExpired:
        raise SVNError(f"svn command timed out: svn {' '.join(args)}")

    if r.returncode != 0:
        msg = (r.stderr or r.stdout or b"").decode("utf-8", errors="replace").strip()
        raise SVNError(f"svn failed (exit {r.returncode}): {msg}")

    return r


def get_latest_revision(url: str) -> int:
    """Return the HEAD revision number of an SVN repository."""
    r = _run_svn(["info", "--xml", url], timeout=60)
    root = ET.fromstring(r.stdout)
    entry = root.find(".//entry")
    if entry is None:
        raise SVNError("Could not find <entry> in svn info output")
    rev = entry.get("revision")
    if rev is None:
        raise SVNError("No revision attribute in svn info entry")
    return int(rev)


def get_log(url: str, revision: int) -> SVNLogEntry:
    """Fetch log entry with changed paths for a single SVN revision."""
    r = _run_svn(
        ["log", "-r", str(revision), "--verbose", "--xml", url],
        timeout=120,
    )
    root = ET.fromstring(r.stdout)
    logentry = root.find("logentry")

    if logentry is None:
        raise SVNError(f"Revision {revision} not found")

    author_el = logentry.find("author")
    date_el = logentry.find("date")
    msg_el = logentry.find("msg")

    author = author_el.text.strip() if author_el is not None and author_el.text else "unknown"
    date = date_el.text.strip() if date_el is not None and date_el.text else ""
    msg = msg_el.text.strip() if msg_el is not None and msg_el.text else ""

    paths = []
    paths_el = logentry.find("paths")
    if paths_el is not None:
        for path_el in paths_el.findall("path"):
            action = path_el.get("action", "M")
            kind = path_el.get("kind", "file")
            text = (path_el.text or "").strip()
            cf_path = path_el.get("copyfrom-path")
            cf_rev_str = path_el.get("copyfrom-rev")

            paths.append(SVNPathChange(
                action=action,
                kind=kind,
                path=text,
                copyfrom_path=cf_path,
                copyfrom_rev=int(cf_rev_str) if cf_rev_str else None,
            ))

    return SVNLogEntry(
        revision=revision,
        author=author,
        date=date,
        message=msg,
        paths=paths,
    )


def get_log_range(url: str, start_rev: int, end_rev: int) -> List[SVNLogEntry]:
    """Fetch log entries for a range of SVN revisions in one call.

    Returns entries from *start_rev* to *end_rev* (inclusive), oldest first.
    """
    r = _run_svn(
        ["log", "-r", f"{start_rev}:{end_rev}", "--verbose", "--xml", url],
        timeout=300,
    )
    root = ET.fromstring(r.stdout)
    entries: List[SVNLogEntry] = []
    # svn log returns newest first; reverse to get oldest first
    for logentry in reversed(root.findall("logentry")):
        rev_str = logentry.get("revision")
        if rev_str is None:
            continue
        revision = int(rev_str)

        author_el = logentry.find("author")
        date_el = logentry.find("date")
        msg_el = logentry.find("msg")

        author = author_el.text.strip() if author_el is not None and author_el.text else "unknown"
        date = date_el.text.strip() if date_el is not None and date_el.text else ""
        msg = msg_el.text.strip() if msg_el is not None and msg_el.text else ""

        paths = []
        paths_el = logentry.find("paths")
        if paths_el is not None:
            for path_el in paths_el.findall("path"):
                action = path_el.get("action", "M")
                kind = path_el.get("kind", "file")
                text = (path_el.text or "").strip()
                cf_path = path_el.get("copyfrom-path")
                cf_rev_str = path_el.get("copyfrom-rev")

                paths.append(SVNPathChange(
                    action=action,
                    kind=kind,
                    path=text,
                    copyfrom_path=cf_path,
                    copyfrom_rev=int(cf_rev_str) if cf_rev_str else None,
                ))

        entries.append(SVNLogEntry(
            revision=revision,
            author=author,
            date=date,
            message=msg,
            paths=paths,
        ))

    return entries


def get_file(url: str, path: str, revision: int) -> bytes:
    """Return the full content of a file at a given SVN revision.

    `path` is the absolute SVN path, e.g. /trunk/src/main.c.
    """
    full_url = url.rstrip("/") + path
    r = _run_svn(["cat", "-r", str(revision), full_url], timeout=60)
    return r.stdout