claude-werkzeuge/.vault-tools/auto-push/werkzeuge_spiegel.py

"""Spiegelt die Werkzeuge des Vaults in das öffentliche Repo claude-werkzeuge.

Ersteller: Pascal Michel
Erstellt:  2026-10-14

WAS GESPIEGELT WIRD
  Der committete Stand (HEAD) des Vaults unter .vault-tools/ und .claude/,
  ohne die Ordner in AUSSCHLUESSE, dazu werkzeuge-README.md als README.md.
  Ungesicherte Änderungen im Arbeitsbaum gehen nicht mit. Was die
  .gitignore des Vaults draußen hält (settings.local.json, __pycache__,
  lauf.log), ist deshalb auch hier draußen.

  AUSSCHLUESSE ist eine Negativliste: Ein neuer Werkzeugordner geht beim
  nächsten Lauf mit hinaus. So angeordnet (Pascal, 2026-10-14: künftig alle
  Werkzeuge außer denen des Trading-Projekts). Ein neues Werkzeug des
  Trading-Projekts gehört deshalb in AUSSCHLUESSE, bevor es committet wird.

WOHIN
  In einen Klon außerhalb des Vaults (KLON). Der Lauf holt den Remote-Stand,
  leert den Klon bis auf .git, packt den Stand neu aus und committet und
  pusht nur, wenn sich etwas geändert hat. Die Remote-Adresse entsteht aus
  der origin-Adresse des Vaults, letzter Pfadteil ersetzt; Benutzername und
  Host stehen so nirgends im Quelltext. Das Token kommt wie beim Vault-Push
  über askpass.sh aus GITEA_TOKEN.

SCHUTZ
  Geleert wird nur ein Klon, dessen origin genau auf die erwartete Adresse
  zeigt. Ein Klon im Vault, ein fremder Klon und ein nicht leerer Ordner
  ohne .git sind Befunde, nichts wird gelöscht.

BENANNTE LÜCKEN
  * Kein Geheimnis-Scan. Was im Vault unter .vault-tools/ oder .claude/
    committet ist, geht hinaus, wie beim Vault-Push.
  * Nur reguläre Dateien. Symlinks im Vault werden übersprungen.
  * Eine Änderung direkt im Repo claude-werkzeuge überschreibt der nächste
    Lauf.
  * Ein gescheiterter Push lässt den Commit im Klon stehen; der nächste Lauf
    setzt den Klon auf den Remote-Stand zurück und committet neu.
"""

from __future__ import annotations

import io
import os
import shutil
import stat
import subprocess
import tarfile
from pathlib import Path

KLON = Path(r"D:\ClaudeData\claude-werkzeuge")
REPO_NAME = "claude-werkzeuge.git"
README_QUELLE = ".vault-tools/auto-push/werkzeuge-README.md"
ASKPASS_REL = Path(".vault-tools") / "auto-push" / "askpass.sh"

QUELLEN = (".vault-tools/", ".claude/")
# Verankert ab der Vault-Wurzel, nie als bloße Ordnernamen.
AUSSCHLUESSE = (
    ".vault-tools/trading-plattform-pruefung/",
    ".vault-tools/watermarks-remover/",
)

TIMEOUT_GIT_S = 30
TIMEOUT_NETZ_S = 90


class SpiegelFehler(Exception):
    """Ein Schritt ist gescheitert; die Meldung ist der Befund."""


def git(wo: Path, wurzel: Path, *argumente: str, timeout: int = TIMEOUT_GIT_S) -> bytes:
    umgebung = dict(
        os.environ,
        GIT_TERMINAL_PROMPT="0",
        GIT_ASKPASS=str(wurzel / ASKPASS_REL),
    )
    try:
        lauf = subprocess.run(
            ["git", "-C", str(wo), *argumente],
            env=umgebung,
            stdin=subprocess.DEVNULL,
            capture_output=True,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired as abgelaufen:
        raise SpiegelFehler(f"git {argumente[0]} nach {timeout} s abgebrochen") from abgelaufen
    except OSError as fehler:
        raise SpiegelFehler(f"git nicht startbar: {type(fehler).__name__}: {fehler}") from fehler
    if lauf.returncode != 0:
        text = (lauf.stdout + b"\n" + lauf.stderr).decode("utf-8", errors="replace")
        raise SpiegelFehler(f"git {argumente[0]}: Exit {lauf.returncode}: {' '.join(text.split())[:300]}")
    return lauf.stdout


def gehoert_dazu(pfad: str) -> bool:
    return pfad.startswith(QUELLEN) and not pfad.startswith(AUSSCHLUESSE)


def ziel_adresse(wurzel: Path) -> str:
    adresse = git(wurzel, wurzel, "remote", "get-url", "origin").decode("utf-8").strip()
    if "/" not in adresse:
        raise SpiegelFehler(f"origin des Vaults ohne Pfad: {adresse}")
    return adresse.rsplit("/", 1)[0] + "/" + REPO_NAME


def raeume(pfad: Path) -> None:
    """Löscht Datei oder Ordner, auch schreibgeschützte Dateien."""

    def erzwinge(funktion, ziel, _fehler):
        os.chmod(ziel, stat.S_IWRITE)
        funktion(ziel)

    if pfad.is_dir() and not pfad.is_symlink():
        shutil.rmtree(pfad, onexc=erzwinge)
    else:
        pfad.unlink()


def bereite_klon(wurzel: Path, klon: Path, adresse: str) -> None:
    """Klont beim ersten Lauf, setzt danach auf den Remote-Stand zurück."""
    basis = wurzel.resolve()
    ziel = klon.resolve()
    if ziel == basis or basis in ziel.parents:
        raise SpiegelFehler(f"Klon {klon} liegt im Vault; abgebrochen, nichts geleert")
    if not (klon / ".git").is_dir():
        if klon.exists() and any(klon.iterdir()):
            raise SpiegelFehler(f"{klon} ist kein Git-Klon und nicht leer; abgebrochen, nichts geleert")
        klon.parent.mkdir(parents=True, exist_ok=True)
        git(klon.parent, wurzel, "clone", "-q", "-c", "core.autocrlf=false", adresse, str(klon),
            timeout=TIMEOUT_NETZ_S)
        return
    vorhanden = git(klon, wurzel, "remote", "get-url", "origin").decode("utf-8").strip()
    if vorhanden != adresse:
        raise SpiegelFehler(
            f"origin des Klons zeigt auf {vorhanden}, erwartet {adresse}; abgebrochen, nichts geleert"
        )
    git(klon, wurzel, "fetch", "-q", "origin", timeout=TIMEOUT_NETZ_S)
    git(klon, wurzel, "reset", "-q", "--hard", "origin/main")


def spiegele(wurzel: Path, klon: Path = KLON) -> tuple[str, str]:
    """Gibt ("gespiegelt" oder "unveraendert", Text) zurück. Wirft SpiegelFehler."""
    adresse = ziel_adresse(wurzel)
    stand = git(wurzel, wurzel, "rev-parse", "--short", "HEAD").decode("utf-8").strip()
    # Ganzer HEAD statt Pfadangaben: git archive bricht mit "pathspec did not match"
    # ab, sobald eine Quelle leer ist. Gefiltert wird danach mit gehoert_dazu.
    archiv = git(wurzel, wurzel, "archive", "--format=tar", "HEAD")

    with tarfile.open(fileobj=io.BytesIO(archiv)) as tar:
        mitglieder = [m for m in tar.getmembers() if m.isfile() and gehoert_dazu(m.name)]
        if README_QUELLE not in {m.name for m in mitglieder}:
            raise SpiegelFehler(f"README-Quelle {README_QUELLE} fehlt im Vault-Stand {stand}")

        bereite_klon(wurzel, klon, adresse)
        for eintrag in klon.iterdir():
            if eintrag.name != ".git":
                raeume(eintrag)
        for mitglied in mitglieder:
            tar.extract(mitglied, klon, filter="data")
        readme = tar.extractfile(README_QUELLE)
        (klon / "README.md").write_bytes(readme.read())

    git(klon, wurzel, "add", "-A")
    roh = git(klon, wurzel, "diff", "--cached", "--name-only", "-z").decode("utf-8")
    geaendert = [z for z in roh.split("\0") if z]
    if not geaendert:
        return "unveraendert", f"Vault-Stand {stand}"
    git(klon, wurzel, "commit", "-q", "-m",
        f"Abgleich mit dem Vault-Stand {stand}\n\n{len(geaendert)} Datei(en) geändert.\n"
        "Erzeugt von .vault-tools/auto-push/werkzeuge_spiegel.py.")
    git(klon, wurzel, "push", "-q", "origin", "HEAD:main", timeout=TIMEOUT_NETZ_S)
    return "gespiegelt", f"{len(geaendert)} Datei(en), Vault-Stand {stand}"