311 lines
15 KiB
Python
311 lines
15 KiB
Python
"""Mechanische Gates (tokenfrei): prüfen LLM-Scan-Ausgaben gegen den echten Code.
|
|
|
|
Port von phase0/pruefe.py als Bibliothek. Phase-0-validiert: fängt erfundene Anker,
|
|
umgeschriebene Zitate (auch Minimal-Umformungen) und Format-Drift; Escaping-Rauschen
|
|
der Modelle wird vor dem Vergleich normalisiert.
|
|
"""
|
|
import difflib
|
|
import re
|
|
from pathlib import Path
|
|
|
|
KANTEN_TYPEN = {"ruft-auf", "nutzt", "wird-genutzt-von", "löst-aus", "wird-ausgelöst-von"}
|
|
|
|
SCHLUESSEL = ("beschreibung", "input", "output") # Pflichtfelder je Einheit
|
|
|
|
|
|
def entschaerfe(s: str) -> str:
|
|
"""Escaping-Rauschen normalisieren: \\" → \" und \\\\ → \\ (Erkenntnis 11)."""
|
|
return s.replace('\\"', '"').replace("\\\\", "\\")
|
|
|
|
|
|
def _beleg_text(rest: str) -> str:
|
|
"""Beleg-Inhalt aus dem Zeilenrest: äußere Anführungszeichen abstreifen, Rest wörtlich.
|
|
Kein Quote-Delimiter-Parsing — Belege enthalten selbst Anführungszeichen/Regex
|
|
(Abnahme-Befund 17), die Zeile gehört ab `beleg:` komplett dem Zitat."""
|
|
rest = rest.strip()
|
|
if len(rest) >= 2 and rest.startswith('"') and rest.endswith('"'):
|
|
rest = rest[1:-1]
|
|
elif rest.startswith('"'):
|
|
rest = rest[1:]
|
|
return entschaerfe(rest)
|
|
|
|
|
|
def parse_einheiten(text: str) -> tuple[list[dict], list[str], list[str]]:
|
|
"""Einheiten-Markdown → (einheiten, zusatz_anker, struktur_fehler).
|
|
Einheit: {id, zeile, felder: {beschreibung/input/output}, belege: [(nr, zitat)],
|
|
kanten: [(nr, typ, ziel)], anker: [ids]}. Streng schreiben, tolerant parsen."""
|
|
einheiten: list[dict] = []
|
|
zusatz_anker: list[str] = []
|
|
fehler: list[str] = []
|
|
aktuelle = None
|
|
modus = None
|
|
for nr, zeile in enumerate(text.splitlines(), 1):
|
|
# Einheiten-Kopf: `## datei::symbol` oder Datei-Einheit `## pfad.py` (Skripte
|
|
# ohne def/class — Befund 18)
|
|
if zeile.startswith("## ") and ("::" in zeile or re.fullmatch(r"## \S+\.\w+", zeile.strip())):
|
|
aktuelle = {"id": zeile[3:].strip(), "zeile": nr, "felder": {}, "belege": [],
|
|
"kanten": [], "anker": []}
|
|
einheiten.append(aktuelle)
|
|
modus = None
|
|
elif zeile.startswith(("anker:", "ergaenze-anker:")):
|
|
aid = zeile.split(":", 1)[1].strip().split(" zu ")[0].strip()
|
|
zusatz_anker.append(aid)
|
|
if aktuelle is not None and zeile.startswith("anker:"):
|
|
aktuelle["anker"].append(aid)
|
|
elif aktuelle is None:
|
|
continue
|
|
elif m := re.match(r"(beschreibung|input|output):(.*)$", zeile):
|
|
aktuelle["felder"][m.group(1)] = m.group(2).strip()
|
|
modus = m.group(1)
|
|
elif zeile.strip() in ("entscheidungen:", "fakten:"):
|
|
modus = "punkte"
|
|
elif zeile.strip() == "kanten:":
|
|
modus = "kanten"
|
|
elif m := re.match(r"\s+beleg:\s*(.+)$", zeile):
|
|
aktuelle["belege"].append((nr, _beleg_text(m.group(1))))
|
|
elif modus == "punkte" and (m := re.search(r"\bbeleg:\s*(.+)$", zeile)):
|
|
aktuelle["belege"].append((nr, _beleg_text(m.group(1)))) # tolerant: beleg inline
|
|
elif modus == "kanten" and (m := re.match(r"- ([\wäöü-]+):\s*(\S+)\s*$", zeile)):
|
|
aktuelle["kanten"].append((nr, m.group(1), m.group(2)))
|
|
elif modus == "kanten" and zeile.startswith("- "):
|
|
fehler.append(f"Z{nr}: Kante unlesbar (Zusatztext?): {zeile.strip()!r}")
|
|
return einheiten, zusatz_anker, fehler
|
|
|
|
|
|
def _symbol_in_datei(symbol: str, quelle: str, top_level_nur: bool) -> bool:
|
|
if "." in symbol: # qualifizierte Methode `Klasse.methode` (Abnahme-Befund)
|
|
klasse, _, methode = symbol.partition(".")
|
|
return (_symbol_in_datei(klasse, quelle, top_level_nur)
|
|
and re.search(rf"def\s+{re.escape(methode)}\b", quelle) is not None)
|
|
if top_level_nur:
|
|
muster = rf"^(?:async\s+)?(?:def|class)\s+{re.escape(symbol)}\b|^{re.escape(symbol)}\s*[:=]"
|
|
else: # Zusatz-Anker dürfen verschachtelt sein (Innenfunktionen, Erkenntnis M3-Lauf)
|
|
muster = rf"(?:def|class)\s+{re.escape(symbol)}\b|^{re.escape(symbol)}\s*[:=]"
|
|
return re.search(muster, quelle, re.M) is not None
|
|
|
|
|
|
def pruefe_fragment(text: str, wurzel: Path, chunk_datei: str,
|
|
quelle_cache: dict | None = None) -> tuple[list[str], list[str], list[dict]]:
|
|
"""Gates für EIN Scan-Fragment: Struktur, Anker-Existenz, Zitat-Treue, Pflichtfelder,
|
|
Kanten-Grammatik. → (fehler, warnungen, einheiten). Kanten-ZIELE in fremden Dateien
|
|
werden nur auf Datei-Existenz geprüft (das Symbol prüft der Scan der Zieldatei)."""
|
|
cache = quelle_cache if quelle_cache is not None else {}
|
|
|
|
def quelle(datei: str) -> str | None:
|
|
if datei not in cache:
|
|
p = wurzel / datei
|
|
cache[datei] = p.read_text(encoding="utf-8", errors="replace") if p.is_file() else None
|
|
return cache[datei]
|
|
|
|
einheiten, zusatz_anker, fehler = parse_einheiten(text)
|
|
warnungen: list[str] = []
|
|
if not einheiten:
|
|
fehler.append("keine Einheit geparst (Format-Drift oder leere Antwort)")
|
|
|
|
def pruefe_id(eid: str, kontext: str, top_level: bool):
|
|
datei, _, symbol = eid.partition("::")
|
|
q = quelle(datei)
|
|
if q is None:
|
|
fehler.append(f"{kontext}: Datei fehlt: {datei}")
|
|
elif symbol and not _symbol_in_datei(symbol, q, top_level):
|
|
fehler.append(f"{kontext}: Symbol '{symbol}' nicht in {datei}")
|
|
|
|
for e in einheiten:
|
|
kontext = e["id"]
|
|
if not e["id"].startswith(chunk_datei + "::") and e["id"] != chunk_datei:
|
|
fehler.append(f"{kontext}: Einheit gehört nicht zur gescannten Datei {chunk_datei}")
|
|
continue
|
|
pruefe_id(e["id"], f"Z{e['zeile']} einheit", top_level=True)
|
|
for feld in SCHLUESSEL:
|
|
if not e["felder"].get(feld):
|
|
(fehler if feld == "beschreibung" else warnungen).append(f"{kontext}: {feld}: fehlt")
|
|
q = quelle(chunk_datei) or ""
|
|
for nr, beleg in e["belege"]:
|
|
if beleg not in q:
|
|
fehler.append(f"{kontext} Z{nr}: Zitat nicht wörtlich in {chunk_datei}: \"{beleg[:50]}…\"")
|
|
if not e["belege"]:
|
|
warnungen.append(f"{kontext}: keine Belege")
|
|
for aid in e["anker"]:
|
|
pruefe_id(aid, f"{kontext} anker", top_level=False)
|
|
for nr, typ, ziel in e["kanten"]:
|
|
if typ not in KANTEN_TYPEN:
|
|
fehler.append(f"{kontext} Z{nr}: unbekannter Kanten-Typ '{typ}'")
|
|
ziel_datei = ziel.partition("::")[0]
|
|
if ziel_datei == chunk_datei:
|
|
pruefe_id(ziel, f"{kontext} Z{nr} kante", top_level=False)
|
|
elif quelle(ziel_datei) is None:
|
|
fehler.append(f"{kontext} Z{nr} kante: Datei fehlt: {ziel_datei}")
|
|
for aid in zusatz_anker:
|
|
if not aid.startswith(chunk_datei + "::"):
|
|
fehler.append(f"ergaenze-anker außerhalb der gescannten Datei: {aid}")
|
|
return fehler, warnungen, einheiten
|
|
|
|
|
|
def fehlende_symbole(chunk_symbole: list[str], einheiten: list[dict],
|
|
zusatz_anker: list[str]) -> list[str]:
|
|
"""Abdeckung eines Chunks: welche seiner Top-Level-Symbole sind keiner Einheit
|
|
zugeordnet (weder als ID noch als Anker)?"""
|
|
zugeordnet = {e["id"].partition("::")[2] for e in einheiten}
|
|
zugeordnet |= {a.partition("::")[2] for a in zusatz_anker}
|
|
zugeordnet |= {a.partition("::")[2] for e in einheiten for a in e["anker"]}
|
|
return [s for s in chunk_symbole if s not in zugeordnet]
|
|
|
|
|
|
def repariere_kanten(fragment: str, wurzel: Path, chunk_datei: str,
|
|
quelle_cache: dict | None = None) -> tuple[str, list[str]]:
|
|
"""Mechanische Reparatur VOR den Gates: Kanten mit ungültigem Ziel (erfundene Datei,
|
|
fehlendes Symbol, fehlender datei::-Prefix) werden GESTRICHEN statt das ganze
|
|
Fragment scheitern zu lassen — Kanten sind Zusatz-Metadaten, keine Belege.
|
|
→ (bereinigtes_fragment, gestrichene_zeilen)."""
|
|
cache = quelle_cache if quelle_cache is not None else {}
|
|
|
|
def quelle(datei: str) -> str | None:
|
|
if datei not in cache:
|
|
p = wurzel / datei
|
|
cache[datei] = p.read_text(encoding="utf-8", errors="replace") if p.is_file() else None
|
|
return cache[datei]
|
|
|
|
def ziel_ok(ziel: str) -> bool:
|
|
if "::" not in ziel:
|
|
return False
|
|
datei, _, symbol = ziel.partition("::")
|
|
q = quelle(datei)
|
|
if q is None:
|
|
return False
|
|
return datei != chunk_datei or _symbol_in_datei(symbol, q, top_level_nur=False)
|
|
|
|
gestrichen: list[str] = []
|
|
zeilen_neu: list[str] = []
|
|
for zeile in fragment.splitlines():
|
|
m = re.match(r"- ([\wäöü-]+):\s*(\S+)\s*$", zeile)
|
|
if m and m.group(1) in KANTEN_TYPEN and not ziel_ok(m.group(2)):
|
|
gestrichen.append(zeile.strip())
|
|
continue
|
|
# Freitext hinter einem Kanten-Typ ("- löst-aus: SystemExit bei …") → streichen
|
|
m_frei = re.match(r"- ([\wäöü-]+):\s+\S+\s+\S", zeile)
|
|
if m_frei and m_frei.group(1) in KANTEN_TYPEN:
|
|
gestrichen.append(zeile.strip())
|
|
continue
|
|
if zeile.startswith(("anker:", "ergaenze-anker:")): # kaputte Anker-Zeilen ebenso
|
|
aid = zeile.split(":", 1)[1].strip().split(" zu ")[0].strip()
|
|
datei, _, symbol = aid.partition("::")
|
|
q = quelle(datei) if datei else None
|
|
if datei != chunk_datei or not symbol or q is None \
|
|
or not _symbol_in_datei(symbol, q, top_level_nur=False):
|
|
gestrichen.append(zeile.strip())
|
|
continue
|
|
zeilen_neu.append(zeile)
|
|
return "\n".join(zeilen_neu), gestrichen
|
|
|
|
|
|
_BELEG_RE = re.compile(r"(\bbeleg:\s*)(.+)$")
|
|
|
|
|
|
def repariere_belege(fragment: str, wurzel: Path, chunk_datei: str,
|
|
quelle_cache: dict | None = None) -> tuple[str, list[str]]:
|
|
"""Mechanische Zitat-Reparatur: ein Beleg, der nicht wörtlich in der Quelle steht,
|
|
wird per Fuzzy-Match auf die ähnlichste Quellzeile GESCHNAPPT und durch deren
|
|
wörtlichen Text ersetzt (Abnahme-Befund: Modelle schreiben z. B. `:=` zu `=` um —
|
|
ein Zeichen daneben). Unter der Ähnlichkeitsschwelle bleibt der Beleg unverändert
|
|
und fällt im Gate durch → Befund-Feedback. → (fragment, korrigierte_belege)."""
|
|
cache = quelle_cache if quelle_cache is not None else {}
|
|
if chunk_datei not in cache:
|
|
p = wurzel / chunk_datei
|
|
cache[chunk_datei] = p.read_text(encoding="utf-8", errors="replace") if p.is_file() else None
|
|
quelle = cache[chunk_datei]
|
|
if quelle is None:
|
|
return fragment, []
|
|
zeilen = [z.strip() for z in quelle.splitlines() if z.strip()]
|
|
korrigiert: list[str] = []
|
|
|
|
def snap(m: re.Match) -> str:
|
|
zitat = _beleg_text(m.group(2))
|
|
if not zitat or zitat in quelle:
|
|
return m.group(0)
|
|
# Fenster-Matching: Abdeckung des Zitats durch gemeinsame Blöcke mit einer
|
|
# Quellzeile — funktioniert auch für kurze Zitate aus langen Zeilen.
|
|
beste, beste_deckung = None, 0.0
|
|
for z in zeilen:
|
|
sm = difflib.SequenceMatcher(None, zitat.lower(), z.lower(), autojunk=False)
|
|
if sm.real_quick_ratio() * max(len(zitat), len(z)) < len(zitat) * 0.8:
|
|
continue
|
|
bloecke = [b for b in sm.get_matching_blocks() if b.size > 0]
|
|
deckung = sum(b.size for b in bloecke) / len(zitat)
|
|
if deckung > beste_deckung:
|
|
beste_deckung = deckung
|
|
start = min(b.b for b in bloecke)
|
|
ende = max(b.b + b.size for b in bloecke)
|
|
beste = z[start:ende] # wörtliches Quell-Fenster
|
|
if beste is None or beste_deckung < 0.8 or beste not in quelle:
|
|
return m.group(0)
|
|
korrigiert.append(f"{zitat!r} → {beste[:60]!r}")
|
|
return m.group(1) + '"' + beste[:78] + '"'
|
|
|
|
neu = "\n".join(_BELEG_RE.sub(snap, zeile) for zeile in fragment.splitlines())
|
|
return neu, korrigiert
|
|
|
|
|
|
def repariere_referenzen(sichten_text: str, bekannte_ids: set[str]) -> tuple[str, list[str]]:
|
|
"""Mechanische Reparatur einer Sicht: unbekannte Einheiten-Referenzen werden aus
|
|
`einheiten:`-/`tragende-einheiten:`-Zeilen und Flow-Schritten GESTRICHEN; verliert
|
|
ein Teilfeature/Schritt alle Referenzen, fällt die Zeile bzw. der `###`-Block weg.
|
|
(Abnahme-Befund: Sichten referenzieren Einheiten toter Chunks — per Feedback nicht
|
|
heilbar, weil die IDs schlicht nicht existieren.) → (text, gestrichene_refs)."""
|
|
gestrichen: list[str] = []
|
|
# Symbol → IDs: falscher Datei-Prefix bei eindeutigem Symbol wird KORRIGIERT statt
|
|
# gestrichen (Befund 19: repair.py-Einheit als board_inventory.py referenziert)
|
|
je_symbol: dict[str, list[str]] = {}
|
|
for eid in bekannte_ids:
|
|
je_symbol.setdefault(eid.partition("::")[2], []).append(eid)
|
|
|
|
def filtere(refs: str) -> list[str]:
|
|
gueltig = []
|
|
for ref in refs.split(","):
|
|
ref = ref.strip()
|
|
if not ref:
|
|
continue
|
|
if ref in bekannte_ids:
|
|
gueltig.append(ref)
|
|
continue
|
|
kandidaten = je_symbol.get(ref.partition("::")[2], [])
|
|
if len(kandidaten) == 1:
|
|
gueltig.append(kandidaten[0])
|
|
else:
|
|
gestrichen.append(ref)
|
|
return gueltig
|
|
|
|
zeilen_neu: list[str] = []
|
|
for zeile in sichten_text.splitlines():
|
|
if m := re.match(r"^(einheiten|tragende-einheiten):\s*(.+)$", zeile):
|
|
gueltig = filtere(m.group(2))
|
|
if gueltig:
|
|
zeilen_neu.append(f"{m.group(1)}: {', '.join(gueltig)}")
|
|
elif m.group(1) == "einheiten" and zeilen_neu and zeilen_neu[-1].startswith("### "):
|
|
zeilen_neu.pop() # Teilfeature ohne einzige gültige Referenz fällt weg
|
|
else:
|
|
zeilen_neu.append(f"{m.group(1)}:")
|
|
elif m := re.match(r"^(\d+\.\s.+\s—\s)(.+)$", zeile):
|
|
gueltig = filtere(m.group(2))
|
|
if gueltig:
|
|
zeilen_neu.append(m.group(1) + ", ".join(gueltig))
|
|
# Schritt ohne gültige Referenz fällt weg
|
|
else:
|
|
zeilen_neu.append(zeile)
|
|
return "\n".join(zeilen_neu), gestrichen
|
|
|
|
|
|
def pruefe_referenzen(sichten_text: str, bekannte_ids: set[str]) -> list[str]:
|
|
"""Referenz-Integrität einer Sicht: `einheiten:`-/`tragende-einheiten:`-Zeilen und
|
|
Flow-Schritte (`1. … — id, id`) dürfen nur bekannte Einheiten-IDs nennen."""
|
|
fehler = []
|
|
referenz_zeilen = [m.group(1) for m in
|
|
re.finditer(r"^(?:einheiten|tragende-einheiten):\s*(.+)$", sichten_text, re.M)]
|
|
referenz_zeilen += [m.group(1) for m in
|
|
re.finditer(r"^\d+\.\s.+\s—\s(.+)$", sichten_text, re.M)]
|
|
for zeile in referenz_zeilen:
|
|
for ref in zeile.split(","):
|
|
ref = ref.strip()
|
|
if ref and ref not in bekannte_ids:
|
|
fehler.append(f"unbekannte Einheiten-Referenz: {ref}")
|
|
return fehler
|