603 lines
23 KiB
Python
603 lines
23 KiB
Python
"""Prüft die Karten eines Projekts gegen den Stil-Katalog.
|
|
|
|
Ein Lauf je Projekt, Zustand nur im Speicher — er überlebt keinen Serverneustart.
|
|
Eine Anfrage je Prüfstelle: einmal die Kartenebene (titel, badge, ziel, regeln) und
|
|
dazu je Funktion einmal (titel, ziel, regeln, testen). Jede Anfrage bekommt nur den
|
|
Katalogteil, der für sie gilt, und nur ihren Ausschnitt der Karte — das hält den
|
|
Prompt kurz und die Zuordnung eindeutig.
|
|
Höchstens PRUEFUNG_PARALLEL gleichzeitig; Probleme erscheinen, sobald ihre Stelle
|
|
geprüft ist.
|
|
"""
|
|
|
|
import asyncio
|
|
import difflib
|
|
import json
|
|
import re
|
|
from collections import Counter
|
|
|
|
import agents
|
|
import config
|
|
import features
|
|
import inhalt as inhaltspruefung
|
|
import quellcode
|
|
import struktur
|
|
|
|
# projektname -> {"laeuft": bool, "fertig": int, "gesamt": int, "fehler": int,
|
|
# "probleme": [...], "meldung": str}
|
|
_laeufe = {}
|
|
|
|
# create_task gibt die Aufgabe nur schwach an die Schleife weiter — ohne eigene
|
|
# Referenz kann der Lauf mitten drin weggesammelt werden.
|
|
_aufgaben = set()
|
|
|
|
# Das Modell zeigt nicht mehr auf einzelne Zeilen, sondern schreibt ganze Felder neu.
|
|
# Adressierung war die Fehlerquelle: verschmolzene Zeilen, erfundene Zeilen, Zeilen aus
|
|
# dem Kontextblock. Welche Zeile sich geändert hat, rechnet der Code aus (_stellen).
|
|
SYSTEM = """Du prüfst {gegenstand} gegen einen Stil-Katalog.
|
|
|
|
{stil}
|
|
|
|
Du bekommst die Felder dieser Stelle. Schreibe jedes Feld neu, das dem Katalog nicht
|
|
genügt. Felder, die schon passen, lässt du weg.
|
|
|
|
Antworte ausschließlich mit JSON:
|
|
{{"felder": {{"<feld>": {{"regel": "...", "neu": "...", "grund": "..."}}}}}}
|
|
|
|
- <feld>: {felder}
|
|
- regel: die verletzte Katalogzeile, wörtlich abgeschrieben
|
|
- neu: der vollständige Inhalt des Feldes, so wie er dastehen soll
|
|
- grund: was am alten Inhalt falsch war, höchstens 8 Wörter
|
|
|
|
regel muss Zeichen für Zeichen im Katalog oben stehen. Findest du dort keine Zeile,
|
|
die der Inhalt verletzt, ändere nichts. Dein Geschmack ist keine Regel.
|
|
|
|
Bei regeln und testen ist neu eine Liste: eine Zeile je Eintrag, jede mit "- " davor.
|
|
Behalte dort jede Zeile, die für sich in Ordnung ist, wortgleich bei. Fehlt ein
|
|
Eintrag, ergänze ihn an der richtigen Stelle. Steht einer zu Unrecht da, lass ihn weg.
|
|
Ein Feld darf leer werden, wenn der Katalog das erlaubt.
|
|
|
|
Ändere so wenig wie möglich. Ein leeres oder fehlendes Feld ist kein Verstoß, solange
|
|
keine Katalogzeile es verlangt. Denk dir keine Inhalte aus, die es nicht gibt.
|
|
|
|
Was du schreibst, muss den Katalog erfüllen — es wird gemessen und sonst abgelehnt.
|
|
Steht unten „Gemessen", sind das Verstöße, die feststehen. Behebe sie; dort brauchst
|
|
du keine regel zu nennen.
|
|
{hinweis}
|
|
Nichts zu ändern: {{"felder": {{}}}}
|
|
Erfinde nichts und ändere nur, was der Katalog verlangt.
|
|
Der Katalog ist abschließend — leite keine weiteren Vorgaben daraus ab."""
|
|
|
|
NACHBESSERN = """Dein Vorschlag verstößt weiter gegen messbare Vorgaben:
|
|
{maengel}
|
|
|
|
Schreibe die betroffenen Felder erneut. Gleiches Format wie zuvor."""
|
|
|
|
# Eine Quelle für „was ist eine Liste" — sonst zerlegt der eine Teil, was der
|
|
# andere als Block behandelt, und aus einer Ergänzung wird ein Komplettersatz.
|
|
ZEILENFELDER = struktur.ZEILENFELDER
|
|
# Felder, die leer bleiben dürfen; ein leerer Vorschlag streicht die Stelle
|
|
LEERBAR = ZEILENFELDER + ("ziel",)
|
|
# Alles, was es überhaupt gibt; je Bereich gilt eine engere Auswahl
|
|
ALLE_FELDER = ("titel", "badge", "ziel", "regeln", "testen")
|
|
# Anfragen je Stimme, bis der Vorschlag die messbaren Vorgaben erfüllt
|
|
VERSUCHE = 2
|
|
# Unabhängige Stimmen je Prüfstelle. Verschiedene Temperaturen, damit sie sich
|
|
# unterscheiden — bei identischen Anfragen wäre die Mehrheit wertlos.
|
|
TEMPERATUREN = (0.0, 0.3, 0.6, 0.9)
|
|
# Einstimmig. Ein Verstoß, den nicht jede Stimme sieht, ist ein Grenzfall — und
|
|
# Grenzfälle sind genau das, was den Lauf sonst nie zur Ruhe kommen lässt.
|
|
MEHRHEIT = len(TEMPERATUREN)
|
|
|
|
# Was der jeweilige Prüfer sieht und melden darf. Der Bereichsname ist zugleich das
|
|
# Dateipräfix im Stil-Ordner — „feature" liest allgemein-* und feature-*.
|
|
BEREICHE = {
|
|
"feature": {
|
|
"gegenstand": "die Kartenebene einer Feature-Karte",
|
|
"felder": ("titel", "badge", "ziel", "regeln"),
|
|
# Ohne die Funktionstitel kann niemand beurteilen, ob die Karte kern ist
|
|
"hinweis": (
|
|
"Der Block funktionen ist nur Kontext für die Badge-Einstufung."
|
|
" Melde nichts daraus."
|
|
),
|
|
},
|
|
"funktion": {
|
|
"gegenstand": "eine einzelne Funktion einer Feature-Karte",
|
|
"felder": ("titel", "ziel", "regeln", "testen"),
|
|
"hinweis": "",
|
|
},
|
|
}
|
|
|
|
|
|
def _felder_text(felder: tuple) -> str:
|
|
return f"{', '.join(felder[:-1])} oder {felder[-1]}"
|
|
|
|
|
|
def _stil_lesen(bereich: str) -> str:
|
|
"""Katalog für einen Bereich: die allgemein-Dateien plus die des Bereichs.
|
|
Leerer Ordner heißt Fehlbedienung — dann lieber abbrechen als blind prüfen."""
|
|
|
|
def gehoert(datei, gruppe):
|
|
return datei.stem == gruppe or datei.stem.startswith(f"{gruppe}-")
|
|
|
|
dateien = sorted(
|
|
d for d in config.STIL_DIR.glob("*.md")
|
|
if gehoert(d, "allgemein") or gehoert(d, bereich)
|
|
)
|
|
if not dateien:
|
|
raise ValueError(f"Kein Stil-Katalog für „{bereich}“ in {config.STIL_DIR}")
|
|
return "\n\n".join(d.read_text().strip() for d in dateien)
|
|
|
|
|
|
def _block(marker: str, text: str) -> str:
|
|
"""Leere Felder gar nicht erst mitschicken — sonst meldet der Prüfer sie als fehlend."""
|
|
return f"<!-- {marker} -->\n{text}\n" if text.strip() else ""
|
|
|
|
|
|
def aufgaben(feature: dict) -> list[dict]:
|
|
"""Eine Prüfstelle für die Kartenebene, dazu je Funktion eine.
|
|
`felder` ist das, was geprüft wird; `kontext` geht nur zur Einordnung mit."""
|
|
s = feature["sektionen"]
|
|
liste = [{
|
|
"feature": feature["id"],
|
|
"funktion": "",
|
|
"bereich": "feature",
|
|
"felder": {
|
|
"titel": s["titel"],
|
|
"badge": s["art"],
|
|
"ziel": s["ziel"],
|
|
"regeln": s["regeln"],
|
|
},
|
|
"kontext": {"funktionen": "\n".join(f"- {f['text']}" for f in s["funktionen"])},
|
|
}]
|
|
for f in s["funktionen"]:
|
|
liste.append({
|
|
"feature": feature["id"],
|
|
"funktion": f["text"],
|
|
"bereich": "funktion",
|
|
"felder": {
|
|
"titel": f["text"],
|
|
"ziel": f["ziel"],
|
|
"regeln": f["regeln"],
|
|
"testen": f["testen"],
|
|
},
|
|
"kontext": {},
|
|
})
|
|
return liste
|
|
|
|
|
|
def inhalt(aufgabe: dict, offen: list[dict] = ()) -> str:
|
|
"""Der Ausschnitt für das Modell. `offen` sind gemessene Verstöße, für die der
|
|
Code keinen Ersatz kennt — sie hängen als Auftrag hinten dran."""
|
|
text = "".join(
|
|
_block(k, v) for k, v in {**aufgabe["felder"], **aufgabe["kontext"]}.items()
|
|
)
|
|
if offen:
|
|
text += "\nGemessen, bitte neu schreiben:\n"
|
|
text += "\n".join(struktur.auftrag(b) for b in offen) + "\n"
|
|
return text
|
|
|
|
|
|
def status(projekt: str) -> dict:
|
|
lauf = _laeufe.get(projekt)
|
|
if not lauf:
|
|
return {
|
|
"laeuft": False, "art": "", "fertig": 0, "gesamt": 0, "fehler": 0,
|
|
"probleme": [], "meldung": "",
|
|
}
|
|
return lauf
|
|
|
|
|
|
def _ohne_strich(text: str) -> str:
|
|
"""Listenzeilen kommen mal mit, mal ohne „- " zurück — je Zeile vereinheitlicht,
|
|
damit die Anzeige die Stelle wiederfindet."""
|
|
zeilen = (text or "").strip().splitlines()
|
|
return "\n".join(re.sub(r"^[-*]\s*", "", z.strip()) for z in zeilen)
|
|
|
|
|
|
_STRUKTUR = set(',:}]')
|
|
|
|
|
|
def _quotes_maskieren(roh: str) -> str:
|
|
"""Karteninhalt enthält Anführungszeichen (`Klick auf "Feature entfernen"`).
|
|
Das Modell gibt ihn wortwörtlich zurück und maskiert sie oft nicht — dann ist das
|
|
JSON kaputt. Ein Zeichen schließt den String nur, wenn danach JSON-Struktur folgt."""
|
|
aus = []
|
|
im_string = False
|
|
i = 0
|
|
while i < len(roh):
|
|
z = roh[i]
|
|
if im_string and z == "\\" and i + 1 < len(roh):
|
|
aus.append(roh[i:i + 2]) # bestehende Escape-Sequenz unverändert
|
|
i += 2
|
|
continue
|
|
if z == '"':
|
|
if not im_string:
|
|
im_string = True
|
|
elif roh[i + 1:].lstrip()[:1] in _STRUKTUR or not roh[i + 1:].strip():
|
|
im_string = False
|
|
else:
|
|
aus.append("\\")
|
|
aus.append(z)
|
|
i += 1
|
|
return "".join(aus)
|
|
|
|
|
|
def _json_lesen(text: str) -> dict:
|
|
"""Holt das erste vollständige JSON-Objekt aus der Antwort.
|
|
|
|
raw_decode statt einer Regex: Antworten enden manchmal mit einem Nachsatz oder
|
|
einem zweiten Objekt — beides lief vorher in „Extra data".
|
|
"""
|
|
start = text.find("{")
|
|
if start < 0:
|
|
raise ValueError("Antwort enthält kein JSON")
|
|
roh = text[start:]
|
|
# strict=False lässt Zeilenumbrüche in Strings durch
|
|
leser = json.JSONDecoder(strict=False)
|
|
try:
|
|
return leser.raw_decode(roh)[0]
|
|
except json.JSONDecodeError:
|
|
return leser.raw_decode(_quotes_maskieren(roh))[0]
|
|
|
|
|
|
def _text(wert) -> str:
|
|
"""Listenfelder kommen mal als ein String mit Umbrüchen, mal als JSON-Liste."""
|
|
if isinstance(wert, list):
|
|
return "\n".join(str(z) for z in wert)
|
|
return str(wert) if wert is not None else ""
|
|
|
|
|
|
def _normal(zeile: str) -> str:
|
|
return re.sub(r"\s+", " ", zeile.strip().strip("-*·` ")).lower()
|
|
|
|
|
|
GEMESSEN = "## Vom System gemessen"
|
|
|
|
|
|
def katalogzeilen(stil: str) -> set:
|
|
"""Die zitierbaren Regeln: alle Aufzählungszeilen außer denen, die der Code
|
|
selbst misst. Sonst beruft sich das Modell auf die Füllwortliste und wendet sie
|
|
auf Wörter an, die gar nicht darin stehen."""
|
|
raus = set()
|
|
messbar = False
|
|
for zeile in stil.splitlines():
|
|
if zeile.startswith("## "):
|
|
messbar = zeile.strip() == GEMESSEN
|
|
elif zeile.strip().startswith("- ") and not messbar:
|
|
raus.add(_normal(zeile))
|
|
return raus
|
|
|
|
|
|
def _belegt(regel: str, zeilen: set) -> bool:
|
|
"""Steht die zitierte Regel wirklich im Katalog?
|
|
|
|
Ohne diesen Nachweis findet ein Modell immer irgendetwas zu verbessern und der
|
|
Lauf konvergiert nie. Kleine Zitierfehler lassen wir durch, freie Erfindungen nicht.
|
|
"""
|
|
regel = _normal(regel)
|
|
if not regel:
|
|
return False
|
|
if regel in zeilen:
|
|
return True
|
|
return any(difflib.SequenceMatcher(None, regel, z).ratio() >= 0.9 for z in zeilen)
|
|
|
|
|
|
def _felder_lesen(text: str, erlaubt: tuple, zeilen: set = None,
|
|
gemessen: tuple = ()) -> dict:
|
|
"""feld -> {"neu": str, "grund": str}. Unbekannte Felder fallen weg.
|
|
Ohne belegte Katalogzeile fällt die Änderung weg — außer der Code hat den
|
|
Verstoß selbst gemessen, dann steht er ohnehin fest."""
|
|
daten = _json_lesen(text).get("felder") or {}
|
|
raus = {}
|
|
for feld, wert in daten.items():
|
|
if feld not in erlaubt:
|
|
continue
|
|
if not isinstance(wert, dict): # manche Modelle kürzen auf den reinen Wert ab
|
|
wert = {"neu": wert, "grund": ""}
|
|
neu = _ohne_strich(_text(wert.get("neu")))
|
|
if "\n" in neu and feld not in ZEILENFELDER:
|
|
continue # nur Listen haben Zeilen
|
|
if not neu and feld not in LEERBAR:
|
|
continue # titel und badge dürfen nicht verschwinden
|
|
if zeilen is not None and feld not in gemessen:
|
|
if not _belegt(_text(wert.get("regel")), zeilen):
|
|
continue
|
|
raus[feld] = {"neu": neu, "grund": _text(wert.get("grund")).strip(),
|
|
"regel": _text(wert.get("regel")).strip()}
|
|
return raus
|
|
|
|
|
|
def _maengel(bereich: str, vorschlaege: dict) -> dict:
|
|
"""Was am Vorschlag selbst noch messbar falsch ist. Leer = brauchbar."""
|
|
raus = {}
|
|
for feld, wert in vorschlaege.items():
|
|
befunde = struktur.befunde(bereich, {feld: wert["neu"]})
|
|
if befunde:
|
|
raus[feld] = befunde
|
|
return raus
|
|
|
|
|
|
def _stellen(feld: str, alt: str, neu: str, grund: str, regel: str = "") -> list[dict]:
|
|
"""Aus alt und neu die einzelnen Änderungen ableiten.
|
|
|
|
`original` ist damit immer eine echte Zeile der Karte — das Modell kann sie
|
|
nicht mehr danebenlegen. Jede Stelle ist für sich anwendbar: Ersetzungen und
|
|
Streichungen greifen über den Zeilentext, Einfügungen hängen sich an die
|
|
Zeile davor und schreiben sie mit.
|
|
"""
|
|
if feld not in ZEILENFELDER:
|
|
return [] if alt.strip() == neu.strip() else [_stelle(feld, alt, neu, grund, regel)]
|
|
|
|
a = struktur._zeilen(feld, alt)
|
|
b = struktur._zeilen(feld, neu)
|
|
if a == b:
|
|
return []
|
|
if not a: # leeres Listenfeld — nichts, woran sich etwas festmachen ließe
|
|
return [_stelle(feld, "", "\n".join(b), grund, regel, ganz=True)]
|
|
|
|
raus = []
|
|
for tag, i1, i2, j1, j2 in difflib.SequenceMatcher(None, a, b).get_opcodes():
|
|
if tag == "equal":
|
|
continue
|
|
if tag == "insert":
|
|
# An die Zeile davor hängen; am Anfang an die dahinter
|
|
if i1 > 0:
|
|
anker, block = a[i1 - 1], [a[i1 - 1]] + b[j1:j2]
|
|
else:
|
|
anker, block = a[0], b[j1:j2] + [a[0]]
|
|
raus.append(_stelle(feld, anker, "\n".join(block), grund, regel))
|
|
continue
|
|
# replace und delete: die erste alte Zeile trägt den Ersatz, der Rest fällt weg
|
|
raus.append(_stelle(feld, a[i1], "\n".join(b[j1:j2]), grund, regel))
|
|
for zeile in a[i1 + 1:i2]:
|
|
raus.append(_stelle(feld, zeile, "", grund, regel))
|
|
|
|
# Mehrere Änderungen an einer Liste sind ein Umbau, kein Strauß von Einzelfixes.
|
|
# Einzeln angenommen ergäben sie Dubletten oder Lücken — also nur zusammen.
|
|
if len(raus) > 1:
|
|
return [_stelle(feld, "\n".join(a), "\n".join(b), grund, regel, ganz=True)]
|
|
return raus
|
|
|
|
|
|
def _stelle(feld: str, original: str, vorschlag: str, grund: str, regel: str = "",
|
|
ganz: bool = False) -> dict:
|
|
return {"feld": feld, "original": original, "vorschlag": vorschlag,
|
|
"ganzes_feld": ganz, "regel": regel, # belegte Katalogzeile, nachvollziehbar
|
|
"problem": grund or "Entspricht nicht dem Katalog"}
|
|
|
|
|
|
async def _stimme(system: str, aufgabe: dict, offen: list, erlaubt: tuple,
|
|
zeilen: set, temperatur: float) -> dict:
|
|
"""Eine Meinung, schon gegen die messbaren Vorgaben gefiltert. Was der Prüfer
|
|
selbst falsch schreibt, bekommt einen Versuch zur Nachbesserung."""
|
|
gemessen = tuple({b["feld"] for b in offen})
|
|
nachricht = inhalt(aufgabe, offen)
|
|
vorschlaege = {}
|
|
for versuch in range(VERSUCHE):
|
|
antwort = await agents.frage(system, nachricht, temperatur)
|
|
vorschlaege = _felder_lesen(antwort, erlaubt, zeilen, gemessen)
|
|
maengel = _maengel(aufgabe["bereich"], vorschlaege)
|
|
if not maengel:
|
|
return vorschlaege
|
|
if versuch == VERSUCHE - 1:
|
|
return {f: v for f, v in vorschlaege.items() if f not in maengel}
|
|
nachricht = inhalt(aufgabe, offen) + "\n" + NACHBESSERN.format(
|
|
maengel="\n".join(
|
|
f'- {feld} „{b["original"]}“: {b["problem"]}'
|
|
for feld, befunde in maengel.items() for b in befunde
|
|
)
|
|
)
|
|
return vorschlaege
|
|
|
|
|
|
def _konsens(fassungen: list[dict]) -> dict:
|
|
"""Die Fassung, die den anderen am ähnlichsten ist. Ein Ausreißer unter drei
|
|
zieht das Ergebnis so nicht mit."""
|
|
if len(fassungen) == 1:
|
|
return fassungen[0]
|
|
|
|
def naehe(a):
|
|
return sum(difflib.SequenceMatcher(None, a["neu"], b["neu"]).ratio()
|
|
for b in fassungen)
|
|
|
|
return max(fassungen, key=naehe)
|
|
|
|
|
|
def _mehrheit(proben: list[dict], gemessen: tuple = ()) -> dict:
|
|
"""Nur was alle Stimmen unabhängig bemängeln, wird ein Vorschlag.
|
|
|
|
Bei Grenzfällen urteilt ein Modell mal so, mal so — einzelne Stimmen führten
|
|
dazu, dass jede Runde andere Felder aufschlugen und der Lauf nie zur Ruhe kam.
|
|
|
|
Gemessene Verstöße sind davon ausgenommen. Sie stehen fest, bevor gefragt wird;
|
|
zur Abstimmung steht nur noch der Wortlaut. Vorher verschwand ein gezählter
|
|
Verstoß, sobald eine einzige Stimme das Feld nicht anfasste.
|
|
"""
|
|
zaehler = Counter(feld for p in proben for feld in p)
|
|
return {
|
|
feld: _konsens([p[feld] for p in proben if feld in p])
|
|
for feld, treffer in zaehler.items()
|
|
if treffer >= (1 if feld in gemessen else MEHRHEIT)
|
|
}
|
|
|
|
|
|
async def _neue_felder(system: str, aufgabe: dict, offen: list, erlaubt: tuple,
|
|
zeilen: set = None) -> dict:
|
|
"""Mehrere unabhängige Stimmen, dann die Mehrheit."""
|
|
proben = await asyncio.gather(*(
|
|
_stimme(system, aufgabe, offen, erlaubt, zeilen, t) for t in TEMPERATUREN
|
|
), return_exceptions=True)
|
|
# Eine kaputte Antwort kostet ihre Stimme, nicht die ganze Prüfstelle
|
|
gute = [p for p in proben if isinstance(p, dict)]
|
|
if not gute:
|
|
raise next(p for p in proben if isinstance(p, BaseException))
|
|
gemessen = tuple({b["feld"] for b in offen})
|
|
vorschlaege = _mehrheit(gute, gemessen)
|
|
|
|
# Bleibt ein gezählter Verstoß ohne Vorschlag, greift der eigene Ersatz des
|
|
# Messers. Sonst bliebe ein Fehler stehen, den der Code sicher kennt.
|
|
for b in offen:
|
|
if b["feld"] not in vorschlaege and b["vorschlag"]:
|
|
alt = struktur._zeilen(b["feld"], aufgabe["felder"].get(b["feld"], ""))
|
|
neu = [b["vorschlag"] if z == b["original"] else z for z in alt]
|
|
vorschlaege[b["feld"]] = {"neu": "\n".join(neu), "grund": b["problem"],
|
|
"regel": "Vom System gemessen"}
|
|
return vorschlaege
|
|
|
|
|
|
async def _pruefe(stil: dict, aufgabe: dict) -> list[dict]:
|
|
"""Erst messen, dann umschreiben lassen, dann den Unterschied ausrechnen."""
|
|
bereich = BEREICHE[aufgabe["bereich"]]
|
|
system = SYSTEM.format(
|
|
stil=stil[aufgabe["bereich"]],
|
|
gegenstand=bereich["gegenstand"],
|
|
felder=_felder_text(bereich["felder"]),
|
|
hinweis=bereich["hinweis"],
|
|
)
|
|
offen = struktur.befunde(aufgabe["bereich"], aufgabe["felder"])
|
|
vorschlaege = await _neue_felder(system, aufgabe, offen, bereich["felder"],
|
|
katalogzeilen(stil[aufgabe["bereich"]]))
|
|
|
|
probleme = []
|
|
for feld, wert in vorschlaege.items():
|
|
probleme += _stellen(feld, aufgabe["felder"].get(feld, ""), wert["neu"],
|
|
wert["grund"], wert.get("regel", ""))
|
|
# Herkunft stempelt der Code, nicht das Modell — es kennt nur seinen Ausschnitt
|
|
for p in probleme:
|
|
p["feature"] = aufgabe["feature"]
|
|
p["funktion"] = aufgabe["funktion"]
|
|
return probleme
|
|
|
|
|
|
async def _stil_aufgaben(projekt: str):
|
|
"""Prüfstellen der Stilprüfung samt der Kataloge, die sie brauchen."""
|
|
stil = {b: _stil_lesen(b) for b in BEREICHE}
|
|
liste = [a for f in features.pruefbar(projekt) for a in aufgaben(f)]
|
|
return liste, lambda a, _zaehl: _pruefe(stil, a)
|
|
|
|
|
|
def _quelle(projekt: str):
|
|
alle = features.pruefbar(projekt)
|
|
quelle = quellcode.dateien()
|
|
if not quelle:
|
|
raise ValueError("Kein Quelltext gefunden")
|
|
return alle, quelle
|
|
|
|
|
|
async def _feature_aufgaben(projekt: str):
|
|
"""Ein Lauf über alle Funktionen: steht hinter jeder auch Code? Und beschreibt
|
|
keine zwei Funktionen dieselbe Sache?
|
|
|
|
Ein Aufruf statt vieler, weil das Ergebnis je Karte gebündelt wird — zwei
|
|
Vorschläge für dieselbe Funktionsliste wären nicht unabhängig anwendbar.
|
|
Der Zähler läuft trotzdem je Funktion mit.
|
|
"""
|
|
alle, quelle = _quelle(projekt)
|
|
text = quellcode.als_text(quelle)
|
|
anzahl = sum(len(f["sektionen"]["funktionen"]) for f in alle)
|
|
|
|
async def abgleich(_, zaehl):
|
|
return await inhaltspruefung.pruefe_feature_im_code(
|
|
projekt, alle, quelle, text, zaehl, _melder(projekt))
|
|
|
|
return _platzhalter(anzahl), abgleich
|
|
|
|
|
|
async def _code_aufgaben(projekt: str):
|
|
"""Ein Lauf über alle Bedienstellen — Vollständigkeit ist eine Eigenschaft der
|
|
Menge, nicht der einzelnen Karte."""
|
|
alle, quelle = _quelle(projekt)
|
|
punkte = quellcode.interaktionen(quelle)
|
|
|
|
async def abgleich(_, zaehl):
|
|
return await inhaltspruefung.pruefe_code_im_feature(
|
|
projekt, alle, quelle, zaehl, _melder(projekt))
|
|
|
|
return _platzhalter(len(punkte)), abgleich
|
|
|
|
|
|
def _melder(projekt: str):
|
|
"""Sagt dem Lauf, was schiefging, ohne ihn abzubrechen.
|
|
|
|
Ein Sammellauf ist eine einzige Aufgabe. Flöge er wegen einer Stelle, wären
|
|
alle anderen Befunde weg — und der Balken stünde trotzdem auf 100 %. So bleibt
|
|
das Ergebnis erhalten und der Fehlstand steht daneben.
|
|
"""
|
|
def melde(text: str, fehler: int = 0):
|
|
if lauf := _laeufe.get(projekt):
|
|
lauf["meldung"] = text
|
|
lauf["fehler"] += fehler
|
|
return melde
|
|
|
|
|
|
def _platzhalter(anzahl: int) -> list:
|
|
"""Eine echte Aufgabe, der Rest nur für den Fortschrittsbalken."""
|
|
return [True] + [None] * max(0, anzahl - 1)
|
|
|
|
|
|
ARTEN = {
|
|
"stil": _stil_aufgaben,
|
|
"feature": _feature_aufgaben, # Feature im Code: was steht ohne Code da?
|
|
"code": _code_aufgaben, # Code im Feature: was fehlt auf den Karten?
|
|
}
|
|
|
|
|
|
async def _lauf(projekt: str, liste: list[dict], arbeite):
|
|
lauf = _laeufe[projekt]
|
|
schranke = asyncio.Semaphore(config.PRUEFUNG_PARALLEL)
|
|
|
|
def zaehl():
|
|
lauf["fertig"] += 1
|
|
|
|
async def eine(aufgabe):
|
|
if aufgabe is None:
|
|
return # Platzhalter: zählt der Sammellauf selbst hoch
|
|
async with schranke:
|
|
if _laeufe.get(projekt) is not lauf:
|
|
return # verworfen (Panel geschlossen) — keine weitere Anfrage
|
|
try:
|
|
lauf["probleme"].extend(await arbeite(aufgabe, zaehl))
|
|
except Exception as e:
|
|
# Zählt trotzdem als erledigt, sonst hängt der Fortschritt.
|
|
# Die Meldung nennt die Zahl, damit „100 %" nicht vollständig vortäuscht.
|
|
lauf["fehler"] += 1
|
|
lauf["meldung"] = (
|
|
f"{lauf['fehler']} von {lauf['gesamt']} Prüfungen fehlgeschlagen: {e}"
|
|
)
|
|
if aufgabe is not True:
|
|
lauf["fertig"] += 1
|
|
|
|
try:
|
|
await asyncio.gather(*(eine(a) for a in liste))
|
|
finally:
|
|
lauf["laeuft"] = False
|
|
|
|
|
|
async def starte(projekt: str, art: str = "stil") -> dict:
|
|
"""Startet einen Lauf im Hintergrund; ein laufender wird nicht verdoppelt."""
|
|
if art not in ARTEN:
|
|
raise ValueError(f"Unbekannte Prüfart „{art}“")
|
|
if status(projekt)["laeuft"]:
|
|
raise ValueError("Prüfung läuft bereits")
|
|
agents.pruefe_konfiguration() # früher Stopp statt halbem Lauf
|
|
# Vorbereiten vor dem Task: ein Fehler darin wäre sonst unsichtbar
|
|
liste, arbeite = await ARTEN[art](projekt)
|
|
_laeufe[projekt] = {
|
|
"laeuft": True,
|
|
"art": art,
|
|
"fertig": 0,
|
|
"gesamt": len(liste),
|
|
"fehler": 0,
|
|
"probleme": [],
|
|
"meldung": "",
|
|
}
|
|
aufgabe = asyncio.create_task(_lauf(projekt, liste, arbeite))
|
|
_aufgaben.add(aufgabe)
|
|
aufgabe.add_done_callback(_aufgaben.discard)
|
|
return _laeufe[projekt]
|
|
|
|
|
|
def beende(projekt: str):
|
|
"""Ergebnis verwerfen (Panel schließen)."""
|
|
_laeufe.pop(projekt, None)
|