251 lines
12 KiB
Python
251 lines
12 KiB
Python
"""Dedup-Kanäle (Akronym!, Subset-Falle, verwandt-Gedächtnis) + Dopplungs-Detektor."""
|
||
from backend import db, dedup, redundanz, textkit
|
||
from .conftest import topic_anlegen
|
||
|
||
|
||
def _atom(topic, titel, definition, soll_id=None, quelle=1, start=0, ende=10):
|
||
aid = db.insert("atome", topic=topic, titel=titel, definition=definition,
|
||
status="aktiv", soll_id=soll_id)
|
||
db.insert("anker", atom_id=aid, quelle_id=quelle, start=start, ende=ende,
|
||
zitat="z")
|
||
return aid
|
||
|
||
|
||
def test_auto_merge_anker_overlap():
|
||
topic = topic_anlegen()
|
||
a = _atom(topic, "A", "Definition eins", quelle=1, start=0, ende=100)
|
||
b = _atom(topic, "A2", "Definition eins etwas länger", quelle=1, start=20,
|
||
ende=110)
|
||
dedup._auto_merge_anker(topic)
|
||
aktive = db.query("SELECT id FROM atome WHERE topic=? AND status='aktiv'", topic)
|
||
assert len(aktive) == 1
|
||
assert db.one("SELECT merged_into FROM atome WHERE status='verworfen'")[
|
||
"merged_into"] in (a, b)
|
||
|
||
|
||
def test_akronym_kanal_liefert_kandidat():
|
||
topic = topic_anlegen()
|
||
a = _atom(topic, "DNS", "Das System löst Namen zu Adressen auf.",
|
||
quelle=1, start=0, ende=10)
|
||
b = _atom(topic, "Domain Name System", "Es übersetzt Domainnamen in Adressen.",
|
||
quelle=2, start=0, ende=10)
|
||
paare = dedup._kandidaten(topic)
|
||
assert any({p[0]["id"], p[1]["id"]} == {a, b} and p[2] == "akronym"
|
||
for p in paare)
|
||
|
||
|
||
def test_titel_fuzzy_subset_falle():
|
||
topic = topic_anlegen()
|
||
_atom(topic, "Zuhören", "Aufmerksames Aufnehmen von Gesagtem im Gespräch.",
|
||
quelle=1, start=0, ende=10)
|
||
_atom(topic, "Aktives Zuhören", "Technik der Gesprächsführung mit Rückfragen.",
|
||
quelle=2, start=0, ende=10)
|
||
paare = dedup._kandidaten(topic)
|
||
# Ein-Wort-Titel „Zuhören" darf NICHT über die Fuzzy-100-Subset-Falle kommen
|
||
assert not any(p[2] == "titel_fuzzy" for p in paare)
|
||
|
||
|
||
def test_verwandt_kante_verhindert_neupruefung():
|
||
topic = topic_anlegen()
|
||
a = _atom(topic, "Begriff Alpha", "Gemeinsame Wörter in dieser Definition hier.",
|
||
quelle=1, start=0, ende=10)
|
||
b = _atom(topic, "Begriff Alpha", "Gemeinsame Wörter in dieser Definition dort.",
|
||
quelle=2, start=50, ende=60)
|
||
assert dedup._kandidaten(topic)
|
||
db.insert("kanten", topic=topic, von_atom=min(a, b), zu_atom=max(a, b),
|
||
art="verwandt")
|
||
assert dedup._kandidaten(topic) == [] # nie zweimal Tokens
|
||
|
||
|
||
def test_negations_guard_blockiert():
|
||
topic = topic_anlegen()
|
||
_atom(topic, "Determinismus", "Der Automat ist deterministisch aufgebaut.",
|
||
quelle=1, start=0, ende=10)
|
||
_atom(topic, "Determinismus ", "Der Automat ist nicht deterministisch aufgebaut.",
|
||
quelle=2, start=0, ende=10)
|
||
assert dedup._kandidaten(topic) == []
|
||
|
||
|
||
async def test_keiner_atome_nicht_erneut_gefragt():
|
||
"""Endlosschleifen-Guard: geprüfte Atome ohne Soll erzeugen keine neuen
|
||
Zuordnungs-Tasks mehr (Echtlauf 2: dedup↔zuordnung drehte sich im Kreis)."""
|
||
from backend import inventar
|
||
topic = topic_anlegen()
|
||
run_id = db.insert("runs", topic=topic, status="running")
|
||
db.insert("soll", topic=topic, punkt="Punkt", status="bestaetigt")
|
||
aid = _atom(topic, "Fremdes Atom", "Definition ohne Bezug zum Punkt hier.")
|
||
task = {"run_id": run_id, "topic": topic, "runde": 1, "item": "z0",
|
||
"knoten": "soll_zuordnung",
|
||
"payload": db.j({"atom_ids": [aid]})}
|
||
# Fake-Zuordnung antwortet KEINER (Titel unbekannt) → Atom bleibt ohne Soll
|
||
erg = await inventar.soll_zuordnung(task)
|
||
assert erg.daten["zugeordnet"] == 0
|
||
atom = db.one("SELECT soll_id, soll_geprueft FROM atome WHERE id=?", aid)
|
||
assert atom["soll_id"] is None and atom["soll_geprueft"] == 1
|
||
# Dedup erzeugt für dieses Atom KEINE neue Zuordnungs-Aufgabe mehr
|
||
erg2 = await dedup.dedup({"run_id": run_id, "topic": topic, "runde": 1,
|
||
"knoten": "dedup", "item": "r1:dedup",
|
||
"payload": "{}"})
|
||
assert erg2.neue_tasks == []
|
||
|
||
|
||
def test_zahlen_namen_kanal():
|
||
"""Dieselbe Studie in Paraphrase (Audit: a20≈a78 Biwer 2023)."""
|
||
topic = topic_anlegen()
|
||
sid1 = db.insert("soll", topic=topic, punkt="P1", status="bestaetigt")
|
||
sid2 = db.insert("soll", topic=topic, punkt="P2", status="bestaetigt")
|
||
a = _atom(topic, "Studie zu Pausen",
|
||
"Biwer et al. 2023 verglichen 25 mit 35 Studierenden bei Pausen.",
|
||
soll_id=sid1, quelle=1, start=0, ende=10)
|
||
b = _atom(topic, "Selbstgesteuerte Pausen",
|
||
"In der Untersuchung von Biwer aus 2023 zeigten 35 Personen mehr "
|
||
"Ermüdung.", soll_id=sid2, quelle=2, start=0, ende=10)
|
||
paare = dedup._kandidaten(topic)
|
||
assert any({p[0]["id"], p[1]["id"]} == {a, b} and p[2] == "zahlen_namen"
|
||
for p in paare)
|
||
|
||
|
||
def test_gleicher_soll_ohne_jaccard_huerde():
|
||
topic = topic_anlegen()
|
||
sid = db.insert("soll", topic=topic, punkt="P", status="bestaetigt")
|
||
a = _atom(topic, "Ziel je Runde",
|
||
"Vor jeder Runde einen konkreten Zielsatz festhalten.",
|
||
soll_id=sid, quelle=1, start=0, ende=10)
|
||
b = _atom(topic, "Erfolgskriterium",
|
||
"Was hinterher anders sein soll, wird vorab notiert.",
|
||
soll_id=sid, quelle=2, start=0, ende=10)
|
||
paare = dedup._kandidaten(topic)
|
||
assert any({p[0]["id"], p[1]["id"]} == {a, b} and p[2] == "gleicher_soll"
|
||
for p in paare)
|
||
|
||
|
||
async def test_enthalten_merged_in_den_umfassenderen():
|
||
topic = topic_anlegen()
|
||
run_id = db.insert("runs", topic=topic, status="running")
|
||
sid = db.insert("soll", topic=topic, punkt="P", status="bestaetigt")
|
||
kurz = _atom(topic, "Basisregeln", "Kurze Definition.",
|
||
soll_id=sid, quelle=1, start=0, ende=10)
|
||
lang = _atom(topic, "Basisregeln und Sonderfaelle",
|
||
"Deutlich längere Definition mit allen Sonderfällen und mehr "
|
||
"Kontext für Lernende.", soll_id=sid, quelle=2, start=0, ende=10)
|
||
await dedup.dedup({"run_id": run_id, "topic": topic, "runde": 1,
|
||
"knoten": "dedup", "item": "r1:dedup", "payload": "{}"})
|
||
gewinner = db.one("SELECT id FROM atome WHERE topic=? AND status='aktiv'",
|
||
topic)
|
||
assert gewinner["id"] == lang # längere Definition gewinnt
|
||
assert db.one("SELECT merged_into FROM atome WHERE id=?", kurz)[
|
||
"merged_into"] == lang
|
||
|
||
|
||
def test_redundanz_shingle_kanal():
|
||
"""Identische 5-Wort-Folge in sonst verschiedenen Sätzen (Audit:
|
||
„Ich melde mich in 20 Minuten" 3×, von Jaccard übersehen)."""
|
||
topic = topic_anlegen()
|
||
_section(topic, 0, "Bei Störungen hilft der Satz Ich melde mich in zwanzig "
|
||
"Minuten als freundliche Antwort im Büro.")
|
||
_section(topic, 1, "Kollegen akzeptieren meist ein kurzes Ich melde mich "
|
||
"in zwanzig Minuten und arbeiten dann weiter an ihren "
|
||
"eigenen Aufgaben im Projekt.")
|
||
paare = redundanz.kandidaten_paare(topic)
|
||
assert len(paare) == 1
|
||
|
||
|
||
def _section(topic, kapitel_ord, text, ziel_suffix=""):
|
||
kid = db.insert("kapitel", topic=topic, titel=f"K{kapitel_ord}", ord=kapitel_ord)
|
||
zid = db.insert("lernziele", topic=topic, text="z" + ziel_suffix)
|
||
bid = db.insert("bausteine", topic=topic, ziel_id=zid, titel=f"B{kapitel_ord}",
|
||
kapitel_id=kid, ord=0)
|
||
db.insert("sections", baustein_id=bid, text=text)
|
||
return bid
|
||
|
||
|
||
def test_redundanz_detektor_findet_dopplung():
|
||
topic = topic_anlegen()
|
||
satz = "Ein Protokoll regelt den Austausch von Nachrichten zwischen Partnern."
|
||
_section(topic, 0, f"Einleitung hier. {satz}")
|
||
spaeter = _section(topic, 1, f"Anderes Kapitel beginnt. {satz}")
|
||
paare = redundanz.kandidaten_paare(topic)
|
||
assert len(paare) == 1
|
||
assert max(paare[0]["a"], paare[0]["b"], key=lambda x: x["pos"])["bid"] == spaeter
|
||
|
||
|
||
def test_redundanz_gleiche_section_kein_paar():
|
||
topic = topic_anlegen()
|
||
satz = "Ein Protokoll regelt den Austausch von Nachrichten zwischen Partnern."
|
||
_section(topic, 0, f"{satz} Noch etwas Text dazwischen. {satz}")
|
||
assert redundanz.kandidaten_paare(topic) == []
|
||
|
||
|
||
def test_redundanz_verschiedene_saetze_kein_paar():
|
||
topic = topic_anlegen()
|
||
_section(topic, 0, "Ein Protokoll regelt den Nachrichtenaustausch der Partner.")
|
||
_section(topic, 1, "Kapitel zwei behandelt ein völlig anderes Feld der Praxis.")
|
||
assert redundanz.kandidaten_paare(topic) == []
|
||
|
||
|
||
async def test_fehlendes_urteil_nachcall_dann_befund(monkeypatch):
|
||
"""Fehlt ein Paar-Urteil, folgt EIN Nach-Call; bleibt es aus → sichtbarer
|
||
parse-Befund und KEINE verwandt-Kante (Paar bleibt künftig prüfbar)."""
|
||
from backend import llm as llm_mod
|
||
topic = topic_anlegen()
|
||
run_id = db.insert("runs", topic=topic, status="running")
|
||
a = _atom(topic, "DNS", "Das System löst Namen zu Adressen auf.",
|
||
quelle=1, start=0, ende=10)
|
||
b = _atom(topic, "Domain Name System", "Es übersetzt Domainnamen.",
|
||
quelle=2, start=0, ende=10)
|
||
calls = {"n": 0}
|
||
|
||
async def ohne_urteil(**kw):
|
||
calls["n"] += 1
|
||
return "===URTEIL===\nPAAR: 99\nURTEIL: verschieden\nGRUND: x"
|
||
|
||
monkeypatch.setattr(llm_mod, "call", ohne_urteil)
|
||
await dedup.dedup({"run_id": run_id, "topic": topic, "runde": 1,
|
||
"knoten": "dedup", "item": "r1:dedup", "payload": "{}"})
|
||
assert calls["n"] == 4 # 2 Panel-Stimmen + 2 Nach-Call-Stimmen
|
||
assert db.one("SELECT id FROM befunde WHERE run_id=? AND art='parse'",
|
||
run_id)
|
||
assert db.query("SELECT id FROM kanten WHERE topic=?", topic) == []
|
||
aktiv = db.query("SELECT id FROM atome WHERE topic=? AND status='aktiv'",
|
||
topic)
|
||
assert {r["id"] for r in aktiv} == {a, b} # nichts still verschmolzen
|
||
|
||
|
||
def test_fakten_kanal_findet_studien_paraphrase():
|
||
"""Audit: „Ariga & Lleras 2011" stand 3× im Guide, Wort-Jaccard nur 0.21 —
|
||
geteilte Zahlen+Namen machen das Paar jetzt zum Kandidaten."""
|
||
topic = topic_anlegen()
|
||
_section(topic, 0, "Die Untersuchung von Ariga und Lleras aus dem Jahr "
|
||
"2011 zeigte über lange Zeit stabile Aufmerksamkeit "
|
||
"bei den Probanden im Labor.")
|
||
_section(topic, 1, "Kurze bewusste Unterbrechungen halfen laut Ariga "
|
||
"sowie Lleras 2011 den Teilnehmern messbar besser "
|
||
"durch anstrengende Arbeitsphasen zu kommen.",
|
||
ziel_suffix="2")
|
||
paare = redundanz.kandidaten_paare(topic)
|
||
assert len(paare) == 1
|
||
|
||
|
||
async def test_wortgleiche_paare_ohne_freispruch(monkeypatch):
|
||
"""Fast wörtliche Paare (gemeinsame 5-Wort-Folge) darf das Panel nicht
|
||
als „didaktisch gewollt" freisprechen — immer Befund."""
|
||
from backend import llm as llm_mod
|
||
topic = topic_anlegen()
|
||
run_id = db.insert("runs", topic=topic, status="running")
|
||
satz = ("Die Pausen sollten proportional zur Arbeitszeit verlängert "
|
||
"werden, sagt die Studie aus Maastricht deutlich.")
|
||
_section(topic, 0, satz)
|
||
_section(topic, 1, satz + " Dazu kommt hier noch ein Zusatz.",
|
||
ziel_suffix="2")
|
||
|
||
async def gewollt(**kw):
|
||
return "===URTEIL===\nPAAR: 1\nURTEIL: didaktisch_gewollt"
|
||
|
||
monkeypatch.setattr(llm_mod, "call", gewollt)
|
||
erg = await redundanz.pruefen({"run_id": run_id, "topic": topic,
|
||
"knoten": "redundanz", "runde": 1,
|
||
"item": "r1:redundanz", "payload": "{}"})
|
||
assert erg.daten["befunde"] == 1
|
||
assert db.one("SELECT id FROM befunde WHERE run_id=? AND art='redundanz'",
|
||
run_id)
|