187 lines
6.0 KiB
Python
187 lines
6.0 KiB
Python
"""SVN repository remote interface (via svn CLI)."""
|
||
|
||
import os
|
||
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,
|
||
auth_args: Optional[list] = None) -> subprocess.CompletedProcess:
|
||
if auth_args:
|
||
args = auth_args + args
|
||
args = ["--non-interactive"] + args
|
||
env = {**os.environ, "LC_ALL": "C.UTF-8"}
|
||
kwargs = dict(
|
||
capture_output=True,
|
||
timeout=timeout,
|
||
env=env,
|
||
)
|
||
if input is not None:
|
||
kwargs["input"] = input
|
||
else:
|
||
kwargs["stdin"] = subprocess.DEVNULL
|
||
try:
|
||
r = subprocess.run(["svn"] + args, **kwargs)
|
||
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, auth_args: Optional[list] = None) -> int:
|
||
"""Return the HEAD revision number of an SVN repository."""
|
||
r = _run_svn(["info", "--xml", url], timeout=60, auth_args=auth_args)
|
||
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, auth_args: Optional[list] = None) -> SVNLogEntry:
|
||
"""Fetch log entry with changed paths for a single SVN revision."""
|
||
r = _run_svn(
|
||
["log", "-r", str(revision), "--verbose", "--xml", url],
|
||
timeout=120, auth_args=auth_args,
|
||
)
|
||
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,
|
||
auth_args: Optional[list] = None) -> 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, auth_args=auth_args,
|
||
)
|
||
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,
|
||
auth_args: Optional[list] = None) -> 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,
|
||
auth_args=auth_args)
|
||
return r.stdout
|