claude-regelwerk/.vault-tools/plan-kopie/welle_vergleich.py

"""Gegenpruefung nach einer Executer-Welle: Menge der geaenderten Dateien und Byte-Vergleich gegen die Kopie.

Koordinator-Werkzeug im Scratchpad. Fragen und Herkunft der Antwort:
  1. Welche Dateien hat die Welle geaendert?      Quelle: `git status --porcelain -z` im Projekt.
  2. Welche durfte sie aendern?                    Quelle: Abschnitt "## Zieldateien" der uebergebenen Taskdateien.
  3. Ist jede Zieldatei byte-gleich mit dem Plan?  Quelle: die Kopie `mat`, auf der plan_materialisiere.py
                                                   alle Codebloecke des Plans angewendet hat.
Was es NICHT sieht (Fangnetz, kein Nachweis):
  - Dateien, die .gitignore ausblendet (data/, artifacts/, node_modules/): Ein Executer, der dorthin
    schreibt, bleibt unsichtbar.
  - Ob die Kopie selbst sauber ist. Eine verschmutzte Kopie erzeugt Fehlalarm, kein falsches Gruen.
  - Prosa-Dateien (--prosa): nur Mengenpruefung, kein Byte-Vergleich; die liest der Koordinator selbst.
  - Bereits committete Aenderungen: Verglichen wird der Arbeitsbaum gegen HEAD.
Eine fehlende Voraussetzung (Kopie fehlt, Taskdatei ohne Zieldateien, git laeuft nicht) ist ein Befund.
Aufruf: python welle_vergleich.py --projekt <pfad> --mat <pfad> [--prosa <rel> ...] <taskdatei> [...]
Exit 0 ohne Befund, 2 mit Befund.
"""

import argparse
import re
import subprocess
import sys
import traceback
from pathlib import Path, PureWindowsPath

ZIEL = re.compile(r"^- `(.+?)`\s*\((.*)\)\s*$")


def zieldateien(taskdatei: Path, projekt: Path, befunde: list[str]) -> dict[str, bool]:
    """Relativer Pfad -> True, wenn die Datei geloescht werden soll."""
    try:
        zeilen = taskdatei.read_text(encoding="utf-8").replace("\r\n", "\n").split("\n")
    except OSError as fehler:
        befunde.append(f"{taskdatei.name}: nicht lesbar ({fehler})")
        return {}
    drin = False
    ziele: dict[str, bool] = {}
    for zeile in zeilen:
        if zeile.startswith("## "):
            drin = zeile.startswith("## Zieldateien")
            continue
        if not drin:
            continue
        treffer = ZIEL.match(zeile)
        if not treffer:
            continue
        absolut = PureWindowsPath(treffer.group(1))
        try:
            rel = absolut.relative_to(PureWindowsPath(str(projekt)))
        except ValueError:
            befunde.append(f"{taskdatei.name}: Zieldatei {absolut} liegt nicht unter dem Projekt {projekt}")
            continue
        ziele[rel.as_posix()] = "gelöscht" in treffer.group(2) or "geloescht" in treffer.group(2)
    if not ziele:
        befunde.append(f"{taskdatei.name}: keine Zieldateien gefunden -- ohne sie misst der Lauf nichts")
    return ziele


def geaendert(projekt: Path, befunde: list[str]) -> dict[str, str]:
    """Relativer Pfad -> zweistelliger Status aus git."""
    try:
        ergebnis = subprocess.run(
            ["git", "-C", str(projekt), "status", "--porcelain", "-z", "--untracked-files=all"],
            capture_output=True,
            stdin=subprocess.DEVNULL,
        )
    except OSError as fehler:
        befunde.append(f"git laesst sich nicht starten ({fehler})")
        return {}
    if ergebnis.returncode != 0:
        befunde.append(f"git status endet mit {ergebnis.returncode}: {ergebnis.stderr.decode('utf-8', 'replace').strip()}")
        return {}
    teile = ergebnis.stdout.decode("utf-8", "replace").split("\0")
    aus: dict[str, str] = {}
    i = 0
    while i < len(teile):
        eintrag = teile[i]
        i += 1
        if len(eintrag) < 4:
            continue
        status, pfad = eintrag[:2], eintrag[3:]
        if status[0] in "RC":
            alt = teile[i] if i < len(teile) else "?"
            i += 1
            befunde.append(f"{pfad}: git meldet Umbenennung oder Kopie aus {alt} -- im Plan nicht vorgesehen")
        aus[pfad] = status
    return aus


def erste_abweichung(echt: bytes, kopie: bytes) -> str:
    if echt.replace(b"\r\n", b"\n") == kopie.replace(b"\r\n", b"\n"):
        return "nur die Zeilenenden weichen ab (CRLF gegen LF)"
    a, b = echt.split(b"\n"), kopie.split(b"\n")
    for nummer, (links, rechts) in enumerate(zip(a, b), start=1):
        if links != rechts:
            return f"Zeile {nummer}: Baum {links[:160]!r} | Plan {rechts[:160]!r}"
    return f"Laenge: Baum {len(a)} Zeilen, Plan {len(b)} Zeilen; die gemeinsamen Zeilen sind gleich"


def pruefe(args: argparse.Namespace, befunde: list[str]) -> None:
    projekt, mat = Path(args.projekt), Path(args.mat)
    if not projekt.is_dir():
        befunde.append(f"Projektordner {projekt} fehlt")
        return
    if not mat.is_dir():
        befunde.append(f"Kopie {mat} fehlt -- ohne sie gibt es keinen Byte-Vergleich")
        return
    prosa = {p.replace("\\", "/") for p in args.prosa}
    ziele: dict[str, bool] = {}
    for name in args.taskdateien:
        ziele.update(zieldateien(Path(name), projekt, befunde))
    status = geaendert(projekt, befunde)

    for pfad in sorted(set(status) - set(ziele)):
        befunde.append(f"{pfad}: geaendert ({status[pfad]!r}), liegt aber ausserhalb der Zieldateien dieser Welle")
    for pfad in sorted(prosa - set(ziele)):
        befunde.append(f"{pfad}: als Prosa uebergeben, steht aber in keiner Taskdatei als Zieldatei")

    gleich = 0
    for pfad, loeschen in sorted(ziele.items()):
        echt, kopie = projekt / pfad, mat / pfad
        if loeschen:
            if echt.exists():
                befunde.append(f"{pfad}: soll geloescht sein, steht aber noch im Baum")
            if kopie.exists():
                befunde.append(f"{pfad}: soll geloescht sein, steht aber noch in der Kopie")
            if not echt.exists() and "D" not in status.get(pfad, ""):
                befunde.append(f"{pfad}: fehlt im Baum, git meldet aber keine Loeschung ({status.get(pfad)!r})")
            continue
        if pfad not in status:
            befunde.append(f"{pfad}: Zieldatei ist unveraendert gegenueber HEAD")
            continue
        if not echt.is_file():
            befunde.append(f"{pfad}: Zieldatei fehlt im Baum")
            continue
        if pfad in prosa:
            continue
        if not kopie.is_file():
            befunde.append(f"{pfad}: fehlt in der Kopie -- kein Byte-Vergleich moeglich")
            continue
        a, b = echt.read_bytes(), kopie.read_bytes()
        if a == b:
            gleich += 1
        else:
            befunde.append(f"{pfad}: weicht vom Plan ab -- {erste_abweichung(a, b)}")
    print(f"{len(ziele)} Zieldateien, {len(status)} geaenderte Dateien laut git, {gleich} byte-gleich mit der Kopie, {len(prosa & set(ziele))} nur als Menge geprueft.")


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--projekt", required=True)
    parser.add_argument("--mat", required=True)
    parser.add_argument("--prosa", nargs="*", default=[])
    parser.add_argument("taskdateien", nargs="+")
    args = parser.parse_args()
    befunde: list[str] = []
    try:
        pruefe(args, befunde)
    except Exception:  # Ein Absturz ist selbst ein Befund; die gesammelten Befunde gehen trotzdem raus.
        befunde.append("unerwarteter Abbruch: " + traceback.format_exc().strip().splitlines()[-1])
    for befund in befunde:
        print("BEFUND " + befund)
    print(f"{len(befunde)} Befund(e).")
    return 2 if befunde else 0


if __name__ == "__main__":
    sys.exit(main())