km/image/km/gitrepo.py

"""Die Ablage als Git-Checkout: klonen, ziehen, alles Geaenderte einliefern.

Weg 1 aus spine/core#8: das Volume traegt ein Checkout, die Wahrheit liegt im
Remote, der Dienst committet und pusht selbst. Muster wie
info/ai/wissens_mcp/repo.py -- bei abgelehntem Push einmal rebasen und
wiederholen. Ein gescheiterter Push wird nicht verschluckt, sondern als
Zustand gemeldet (`status()`), damit niemand glaubt, die Wahrheit laege im
Remote, wenn sie nur lokal liegt.

Zugangsdaten ueber eine ~/.netrc (0600), die der Start aus der Umgebung
schreibt -- nie in der URL, nie auf der Kommandozeile.
"""

from __future__ import annotations

import re
import subprocess
import threading
import time
from pathlib import Path
from urllib.parse import urlparse

from config import Config

_REJECTED = re.compile(r"\b(rejected|non-fast-forward|fetch first|behind)\b", re.IGNORECASE)


class Ablage:
    """Ein Git-Checkout, den km fuehrt: die Ablage selbst oder eine Quelle
    (z. B. die info-Doku), die unter <ns>/quellen/<name> eingehaengt ist und
    ihr eigenes Remote behaelt -- Aenderungen aus der Oberflaeche gehen dort
    zurueck, nicht in eine Kopie."""

    def __init__(self, cfg: Config, path: Path | None = None, remote: str | None = None,
                 branch: str | None = None, name: str = "ablage") -> None:
        self.cfg = cfg
        self.name = name
        self.path = path or cfg.ablage
        self.remote = cfg.git_remote if path is None else remote
        self.branch = branch or cfg.git_branch
        self.lock = threading.Lock()
        self.last_push_error: str | None = None
        self.last_push: float | None = None

    # ------------------------------------------------------------ Hilfen
    def _git(self, *args: str, check: bool = True) -> subprocess.CompletedProcess:
        return subprocess.run(["git", *args], cwd=self.path, check=check,
                              capture_output=True, text=True)

    def write_netrc(self) -> None:
        """Zugang fuer jeden Remote-Host (Ablage und Quellen) in ~/.netrc, 0600."""
        if not (self.cfg.git_user and self.cfg.git_password):
            return
        hosts = {urlparse(u).hostname for u in [self.cfg.git_remote, *[src[1] for src in self.cfg.sources]] if u}
        netrc = Path.home() / ".netrc"
        existing = netrc.read_text() if netrc.exists() else ""
        for host in sorted(h for h in hosts if h):
            if f"machine {host} " not in existing:
                existing += f"machine {host} login {self.cfg.git_user} password {self.cfg.git_password}\n"
        netrc.write_text(existing)
        netrc.chmod(0o600)

    # ------------------------------------------------------------ Lebenszyklus
    def ensure(self) -> None:
        """Checkout vorhanden machen: klonen, sonst init -- nie stillschweigend leer."""
        a = self.path
        if not (a / ".git").exists():
            # Kein clone: das Verzeichnis existiert oft schon, weil SilverBullet
            # dasselbe Volume mountet und dort seine Dateien anlegt, bevor km
            # dran ist. Also in place initialisieren und den Remote-Stand holen.
            a.mkdir(parents=True, exist_ok=True)
            self._git("init", "-q", "-b", self.branch)
            if self.remote:
                self._git("remote", "add", "origin", self.remote)
                r = self._git("fetch", "-q", "origin", check=False)
                if r.returncode != 0:
                    raise SystemExit(f"km: fetch von {self.remote} fehlgeschlagen: {r.stderr.strip()[:300]}")
                has_branch = self._git("rev-parse", "--verify", "-q", f"origin/{self.branch}", check=False).returncode == 0
                if has_branch:
                    r = self._git("checkout", "-q", "-B", self.branch, f"origin/{self.branch}", check=False)
                    if r.returncode != 0:
                        raise SystemExit(f"km: checkout kollidiert mit vorhandenen Dateien: {r.stderr.strip()[:300]}")
                # sonst: leeres Remote, der erste Push legt den Branch an
            self._git("config", "user.name", self.cfg.git_name)
            self._git("config", "user.email", self.cfg.git_email)
        if self.name != "ablage":
            return
        # Was die Oberflaeche fuer sich selbst ablegt, gehoert nicht in die
        # Wahrheit; die Quellen haben ihr eigenes Remote und werden hier ignoriert.
        gi = a / ".gitignore"
        existing = gi.read_text() if gi.exists() else ""
        # */doku/ sind die Quellen-Checkouts -- deren Wahrheit liegt im
        # jeweiligen Remote (z. B. ahmann/info), nicht ein zweites Mal hier.
        # _plug/ bleibt drin: die Plug-Dateien (TreeView) gehoeren zur Ablage,
        # sonst fehlt nach einem frischen Checkout die Seitenleiste.
        for line in (".silverbullet*", "*/doku/"):
            if line not in existing.splitlines():
                existing += line + "\n"
        gi.write_text(existing)

    def pull(self) -> None:
        if not self.remote:
            return
        r = self._git("pull", "--rebase", "--autostash", "origin", self.branch, check=False)
        if r.returncode != 0 and "couldn't find remote ref" not in r.stderr:
            print(f"km: pull {self.name}: {r.stderr.strip()[:300]}", flush=True)

    def commit_all(self, message: str) -> bool:
        """Alles Geaenderte einliefern (auch Aenderungen aus SilverBullet). True, wenn ein Commit entstand."""
        self._git("add", "-A")
        if self._git("diff", "--cached", "--quiet", check=False).returncode == 0:
            return False
        self._git("-c", f"user.name={self.cfg.git_name}", "-c", f"user.email={self.cfg.git_email}",
                  "commit", "-q", "-m", message)
        return True

    def push(self) -> None:
        if not self.remote:
            return
        for _ in range(3):
            r = self._git("push", "-u", "origin", self.branch, check=False)
            if r.returncode == 0:
                self.last_push_error = None
                self.last_push = time.time()
                return
            if not _REJECTED.search(r.stderr):
                self.last_push_error = r.stderr.strip()[:300]
                print(f"km: push {self.name} fehlgeschlagen: {self.last_push_error}", flush=True)
                return
            self._git("pull", "--rebase", "origin", self.branch, check=False)
        self.last_push_error = "push nach 3 Versuchen abgelehnt"
        print(f"km: {self.last_push_error}", flush=True)

    def head(self) -> str | None:
        r = self._git("rev-parse", "--short", "HEAD", check=False)
        return r.stdout.strip() or None

    def status(self) -> dict:
        return {"head": self.head(), "remote": bool(self.remote),
                "letzter_push": self.last_push, "push_fehler": self.last_push_error}


def sources_of(cfg: Config) -> list[Ablage]:
    """Die Quellen als Checkouts unter <ns>/doku/<name> in der Ablage: die
    Oberflaeche sieht sie, km zieht sie je Takt und liefert Aenderungen aus
    der Oberflaeche in ihr eigenes Remote ein."""
    return [Ablage(cfg, cfg.ablage / cfg.namespace / "doku" / name, url, branch, name=name)
            for name, url, branch in cfg.sources]