350 lines
15 KiB
Python
350 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""Testfaelle fuer Normalisierung und Dublettenerkennung.
|
||
|
||
python3 -m unittest test_dubletten -v
|
||
|
||
Ohne Datenbank. Was die Datenbank angeht, prueft dubletten.py selbst beim
|
||
Lauf: die Zahlen im Bericht sind die Kontrolle.
|
||
"""
|
||
|
||
import datetime as dt
|
||
import unittest
|
||
|
||
import dubletten as d
|
||
import normalisierung as n
|
||
|
||
|
||
class Normalisierung(unittest.TestCase):
|
||
def test_entities_und_tags(self):
|
||
self.assertEqual(n.normalisiere("<b>Habeck</b> & Co."), "habeck co")
|
||
|
||
def test_anfuehrungszeichen_vereinheitlicht(self):
|
||
self.assertEqual(n.normalisiere("„Wende“ und »Wende«"),
|
||
n.normalisiere('"Wende" und "Wende"'))
|
||
|
||
def test_gedankenstriche_vereinheitlicht(self):
|
||
self.assertEqual(n.normalisiere("Bund – Land"), n.normalisiere("Bund - Land"))
|
||
|
||
def test_nfkc(self):
|
||
# dieselbe Zeichenfolge einmal zusammengesetzt, einmal vorkomponiert
|
||
self.assertEqual(n.normalisiere("Grün"), n.normalisiere("Grün"))
|
||
|
||
def test_pluskasten_faellt_weg(self):
|
||
self.assertEqual(n.normalisiere("+++ Newsblog +++ Kanzler tritt zurueck"),
|
||
n.normalisiere("Kanzler tritt zurueck"))
|
||
|
||
def test_trigramme(self):
|
||
self.assertEqual(n.trigramme("a b c d"), {("a", "b", "c"), ("b", "c", "d")})
|
||
|
||
|
||
class SimHash(unittest.TestCase):
|
||
def test_gleicher_text_gleicher_hash(self):
|
||
t = n.trigramme(n.normalisiere("Der Bundestag hat das Gesetz beschlossen"))
|
||
self.assertEqual(d.simhash(t), d.simhash(t))
|
||
|
||
def test_reproduzierbar_ueber_prozesse(self):
|
||
# Der Wert ist festgeschrieben. Aendert sich die Hashfunktion oder die
|
||
# Normalisierung, faellt dieser Test - und dann muss die
|
||
# coder_version steigen, sonst stimmen alte Codings nicht mehr.
|
||
t = n.trigramme(n.normalisiere("Der Bundestag hat das Gesetz beschlossen"))
|
||
self.assertEqual(f"{d.simhash(t):016x}", ERWARTETER_SIMHASH)
|
||
|
||
def test_kleine_aenderung_naeher_als_fremder_text(self):
|
||
"""SimHash ordnet richtig - aber nicht so scharf, wie man hofft.
|
||
|
||
Ein einziges geaendertes Wort verruecken drei Trigramme, und bei
|
||
einer Meldung aus wenigen Dutzend Woertern sind das schnell zehn Bit
|
||
Abstand. Die Grenze von 3 aus der Spezifikation verlangt darum
|
||
praktisch identischen Text; sie ist eine Praezisions-, keine
|
||
Aehnlichkeitsschwelle. Der Test haelt nur die Ordnung fest.
|
||
"""
|
||
basis = "Der Bundestag hat das Gesetz am Freitag beschlossen und damit den Weg frei gemacht"
|
||
nah = basis.replace("frei gemacht", "freigemacht")
|
||
fern = "Ein Sturmtief zieht am Wochenende ueber Norddeutschland hinweg und bringt Regen"
|
||
h = lambda x, y: d.hamming(d.simhash(n.trigramme(n.normalisiere(x))),
|
||
d.simhash(n.trigramme(n.normalisiere(y))))
|
||
self.assertEqual(h(basis, basis), 0)
|
||
self.assertLess(h(basis, nah), h(basis, fern))
|
||
|
||
def test_verschiedene_texte_grosse_distanz(self):
|
||
a = d.simhash(n.trigramme(n.normalisiere(
|
||
"Der Bundestag hat das Gesetz am Freitag beschlossen")))
|
||
b = d.simhash(n.trigramme(n.normalisiere(
|
||
"Ein Sturmtief zieht am Wochenende ueber Norddeutschland hinweg")))
|
||
self.assertGreater(d.hamming(a, b), 3)
|
||
|
||
def test_baender_schubfachschluss(self):
|
||
# Zwei Werte mit genau 3 abweichenden Bits muessen sich mindestens
|
||
# ein Band teilen - darauf beruht die Kandidatensuche.
|
||
a = 0
|
||
b = 0b111
|
||
gemeinsam = set(d.baender(a)) & set(d.baender(b))
|
||
self.assertTrue(gemeinsam)
|
||
|
||
|
||
class Jaccard(unittest.TestCase):
|
||
def test_identisch(self):
|
||
s = {("a", "b", "c")}
|
||
self.assertEqual(d.jaccard(s, s), 1.0)
|
||
|
||
def test_disjunkt(self):
|
||
self.assertEqual(d.jaccard({("a",)}, {("b",)}), 0.0)
|
||
|
||
def test_leer(self):
|
||
self.assertEqual(d.jaccard(set(), {("a",)}), 0.0)
|
||
|
||
|
||
class UnionFind(unittest.TestCase):
|
||
def test_transitiv(self):
|
||
uf = d.UnionFind()
|
||
uf.vereinige(3, 1)
|
||
uf.vereinige(2, 3)
|
||
self.assertEqual(uf.finde(1), uf.finde(2))
|
||
|
||
def test_reihenfolge_egal(self):
|
||
a, b = d.UnionFind(), d.UnionFind()
|
||
for x, y in [(1, 2), (2, 3), (4, 5)]:
|
||
a.vereinige(x, y)
|
||
for x, y in [(4, 5), (3, 2), (2, 1)]:
|
||
b.vereinige(x, y)
|
||
self.assertEqual([a.finde(i) for i in range(1, 6)],
|
||
[b.finde(i) for i in range(1, 6)])
|
||
|
||
|
||
class Ressortkuerzel(unittest.TestCase):
|
||
"""Das Kuerzel, das Haeuser der Agenturzeile voranstellen.
|
||
|
||
Es abzutrennen ist der Unterschied zwischen "findet die Uebernahme" und
|
||
"findet sie nicht" - siehe NOTES.md.
|
||
"""
|
||
|
||
def test_kuerzel_faellt_weg(self):
|
||
self.assertEqual(
|
||
n.normalisiere("Raumfahrt: Deutsche Rakete erfolgreich ins All gestartet",
|
||
ressort_abtrennen=True),
|
||
n.normalisiere("Deutsche Rakete erfolgreich ins All gestartet",
|
||
ressort_abtrennen=True))
|
||
|
||
def test_verschiedene_kuerzel_gleicher_kern(self):
|
||
a = n.normalisiere("Notfaelle: Fuenf Tote bei Unfall mit Frachtflugzeug in Miami",
|
||
ressort_abtrennen=True)
|
||
b = n.normalisiere("Flugverkehr: Fuenf Tote bei Unfall mit Frachtflugzeug in Miami",
|
||
ressort_abtrennen=True)
|
||
self.assertEqual(a, b)
|
||
|
||
def test_kurzer_rest_bleibt_unangetastet(self):
|
||
# "Habeck: Wir haben uns geirrt" - der Kopf ist die Aussage, nicht
|
||
# ein Ressort. Vier Woerter Rest sind zu wenig, um ihn wegzuwerfen.
|
||
self.assertIn("habeck", n.normalisiere("Habeck: Wir haben uns geirrt",
|
||
ressort_abtrennen=True))
|
||
|
||
def test_langer_kopf_bleibt_unangetastet(self):
|
||
titel = "Die lange Vorrede mit sehr vielen Woertern: der eigentliche Kern der Meldung"
|
||
self.assertIn("vorrede", n.normalisiere(titel, ressort_abtrennen=True))
|
||
|
||
def test_ohne_schalter_bleibt_alles_stehen(self):
|
||
self.assertIn("raumfahrt", n.normalisiere("Raumfahrt: Deutsche Rakete gestartet ins All"))
|
||
|
||
def test_teaser_behaelt_seine_doppelpunkte(self):
|
||
# Im Teaser steht der Doppelpunkt mitten im Satz; die Basis
|
||
# titel+teaser trennt darum nichts ab.
|
||
i = _item(1, "Ein Titel ueber irgendetwas Wichtiges",
|
||
"Der Minister sagte: das sei nicht hinnehmbar und muesse aufhoeren",
|
||
basis="titel+teaser")
|
||
self.assertIn("minister", i.text)
|
||
|
||
|
||
class TitelBasis(unittest.TestCase):
|
||
def test_uebernahme_mit_verschiedenen_kuerzeln_findet_zusammen(self):
|
||
items = [_item(1, "Raumfahrt: Deutsche Rakete erfolgreich ins All gestartet",
|
||
"Ein Teaser des einen Hauses ueber den Start.",
|
||
quelle="heise", basis="titel"),
|
||
_item(2, "Deutsche Rakete erfolgreich ins All gestartet",
|
||
"Ein voellig anders formulierter Teaser des anderen Hauses.",
|
||
quelle="tagesschau", basis="titel")]
|
||
uf, _ = d.gruppiere(items)
|
||
self.assertEqual(uf.finde(1), uf.finde(2))
|
||
|
||
def test_dieselben_items_ueber_titel_und_teaser_finden_nicht_zusammen(self):
|
||
# Der Gegenbeweis zur Messung: mit Teaser im Sack faellt das Paar
|
||
# unter die Schwelle.
|
||
items = [_item(1, "Raumfahrt: Deutsche Rakete erfolgreich ins All gestartet",
|
||
"Ein Teaser des einen Hauses ueber den Start.",
|
||
quelle="heise", basis="titel+teaser"),
|
||
_item(2, "Deutsche Rakete erfolgreich ins All gestartet",
|
||
"Ein voellig anders formulierter Teaser des anderen Hauses.",
|
||
quelle="tagesschau", basis="titel+teaser")]
|
||
uf, _ = d.gruppiere(items)
|
||
self.assertNotEqual(uf.finde(1), uf.finde(2))
|
||
|
||
def test_verschiedene_meldungen_bleiben_getrennt(self):
|
||
items = [_item(1, "Bundestag beschliesst Reform der Netzentgelte am Freitag",
|
||
quelle="faz", basis="titel"),
|
||
_item(2, "Sturmtief Xaver erreicht am Abend die Nordseekueste",
|
||
quelle="ntv", basis="titel")]
|
||
uf, _ = d.gruppiere(items)
|
||
self.assertNotEqual(uf.finde(1), uf.finde(2))
|
||
|
||
def test_kurzer_titel_wird_nicht_verglichen(self):
|
||
i = _item(1, "Kanzler tritt zurueck", basis="titel")
|
||
self.assertFalse(i.vergleichbar)
|
||
|
||
|
||
class Agentur(unittest.TestCase):
|
||
def test_klammer(self):
|
||
self.assertEqual(d.agentur("Berlin (dpa) - etwas geschah"), "dpa")
|
||
|
||
def test_reuters(self):
|
||
self.assertEqual(d.agentur("Meldung", "Wie Reuters berichtet"), "reuters")
|
||
|
||
def test_ohne_beleg_keine_vermutung(self):
|
||
self.assertIsNone(d.agentur("Der Bundestag hat entschieden"))
|
||
|
||
|
||
def _item(id, titel, teaser=None, quelle="x", tag=1, gesehen_h=0, basis="titel+teaser"):
|
||
return d.Item(id, quelle, titel, teaser,
|
||
dt.datetime(2026, 9, tag, gesehen_h, tzinfo=dt.timezone.utc),
|
||
dt.date(2026, 9, tag), f"hash{id}", basis=basis)
|
||
|
||
|
||
AGENTURMELDUNG = ("Bundestag beschliesst Reform der Netzentgelte",
|
||
"Der Bundestag hat am Freitag die Reform der Netzentgelte beschlossen. "
|
||
"Die Kosten sollen kuenftig gleichmaessiger verteilt werden.")
|
||
|
||
|
||
class Gruppierung(unittest.TestCase):
|
||
def test_gleiche_meldung_zwei_haeuser(self):
|
||
items = [_item(1, *AGENTURMELDUNG, quelle="faz"),
|
||
_item(2, *AGENTURMELDUNG, quelle="welt")]
|
||
uf, erg = d.gruppiere(items)
|
||
self.assertEqual(uf.finde(1), uf.finde(2))
|
||
|
||
def test_verschiedene_meldungen_bleiben_getrennt(self):
|
||
items = [
|
||
_item(1, *AGENTURMELDUNG),
|
||
_item(2, "Sturmtief Xaver erreicht die Nordseekueste",
|
||
"Der Deutsche Wetterdienst warnt vor Orkanboeen an der Kueste "
|
||
"und rechnet mit Sturmfluten am Abend.", quelle="ntv"),
|
||
]
|
||
uf, erg = d.gruppiere(items)
|
||
self.assertNotEqual(uf.finde(1), uf.finde(2))
|
||
|
||
def test_datumsfenster_trennt(self):
|
||
items = [_item(1, *AGENTURMELDUNG, tag=1), _item(2, *AGENTURMELDUNG, tag=5)]
|
||
uf, erg = d.gruppiere(items)
|
||
self.assertNotEqual(uf.finde(1), uf.finde(2))
|
||
|
||
def test_zu_kurze_items_werden_nicht_gepaart(self):
|
||
items = [_item(1, "Kanzler tritt zurueck"), _item(2, "Kanzler tritt zurueck")]
|
||
self.assertFalse(items[0].vergleichbar)
|
||
uf, erg = d.gruppiere(items)
|
||
self.assertNotEqual(uf.finde(1), uf.finde(2))
|
||
|
||
def test_leitartikel_ist_das_aelteste(self):
|
||
items = [_item(9, *AGENTURMELDUNG, gesehen_h=8),
|
||
_item(4, *AGENTURMELDUNG, gesehen_h=3, quelle="b"),
|
||
_item(7, *AGENTURMELDUNG, gesehen_h=3, quelle="c")]
|
||
uf, erg = d.gruppiere(items)
|
||
gruppe = erg[uf.finde(9)]
|
||
self.assertEqual(gruppe["leitartikel"].id, 4) # frueheste Zeit, dann kleinste id
|
||
|
||
def test_gruppen_id_haengt_nur_am_leitartikel(self):
|
||
# Das ist die Zusage aus gruppen_id(): ein spaeter hinzukommendes
|
||
# Mitglied darf die Kennung der Gruppe nicht veraendern.
|
||
zwei = [_item(1, *AGENTURMELDUNG, gesehen_h=1),
|
||
_item(2, *AGENTURMELDUNG, gesehen_h=2, quelle="b")]
|
||
drei = zwei + [_item(3, *AGENTURMELDUNG, gesehen_h=9, quelle="c")]
|
||
uf2, e2 = d.gruppiere(zwei)
|
||
uf3, e3 = d.gruppiere(drei)
|
||
self.assertEqual(d.nutzlast(zwei[0], e2[uf2.finde(1)])["gruppe"],
|
||
d.nutzlast(drei[0], e3[uf3.finde(1)])["gruppe"])
|
||
|
||
def test_kette_ueber_tage_wird_getrennt(self):
|
||
"""Taegliche Wiederholung darf nicht zu einer Gruppe verschmelzen.
|
||
|
||
Drei gleichlautende Sendungen an drei aufeinanderfolgenden Tagen
|
||
haengen paarweise im Fenster zusammen. Transitiv waeren sie eine
|
||
Gruppe ueber drei Tage - das ist keine Uebernahme.
|
||
"""
|
||
items = [_item(1, *AGENTURMELDUNG, tag=1),
|
||
_item(2, *AGENTURMELDUNG, tag=2),
|
||
_item(3, *AGENTURMELDUNG, tag=3),
|
||
_item(4, *AGENTURMELDUNG, tag=4)]
|
||
uf, erg = d.gruppiere(items)
|
||
gruppen = {uf.finde(i.id) for i in items}
|
||
self.assertGreater(len(gruppen), 1)
|
||
for g in erg.values():
|
||
tage = [i.datum for i in g["mitglieder"]]
|
||
self.assertLessEqual((max(tage) - min(tage)).days, 1)
|
||
|
||
def test_gleiches_haus_an_anderen_tagen_nicht_gepaart(self):
|
||
"""Wiederkehrendes Format ist keine Uebernahme.
|
||
|
||
Diese Regel sitzt auf Paarebene, damit sie fensterunabhaengig ist -
|
||
sie ist der Grund, dass ein Slot-Lauf dieselben Gruppen liefert wie
|
||
ein Backfill.
|
||
"""
|
||
selbe = [_item(1, *AGENTURMELDUNG, quelle="tagesschau", tag=1),
|
||
_item(2, *AGENTURMELDUNG, quelle="tagesschau", tag=2)]
|
||
uf, _ = d.gruppiere(selbe)
|
||
self.assertNotEqual(uf.finde(1), uf.finde(2))
|
||
|
||
# am selben Tag dagegen schon
|
||
selber_tag = [_item(1, *AGENTURMELDUNG, quelle="tagesschau", tag=1),
|
||
_item(2, *AGENTURMELDUNG, quelle="tagesschau", tag=1)]
|
||
uf, _ = d.gruppiere(selber_tag)
|
||
self.assertEqual(uf.finde(1), uf.finde(2))
|
||
|
||
# verschiedene Haeuser an benachbarten Tagen bleiben eine Uebernahme
|
||
haeuser = [_item(1, *AGENTURMELDUNG, quelle="faz", tag=1),
|
||
_item(2, *AGENTURMELDUNG, quelle="welt", tag=2)]
|
||
uf, _ = d.gruppiere(haeuser)
|
||
self.assertEqual(uf.finde(1), uf.finde(2))
|
||
|
||
def test_haeuserzahl_in_der_nutzlast(self):
|
||
# Eine Gruppe aus einem Haus ist Wiederholung, keine Uebernahme.
|
||
items = [_item(1, *AGENTURMELDUNG, quelle="faz"),
|
||
_item(2, *AGENTURMELDUNG, quelle="welt")]
|
||
uf, erg = d.gruppiere(items)
|
||
self.assertEqual(d.nutzlast(items[0], erg[uf.finde(1)])["haeuser"], 2)
|
||
eigen = [_item(1, *AGENTURMELDUNG, quelle="faz"),
|
||
_item(2, *AGENTURMELDUNG, quelle="faz")]
|
||
uf, erg = d.gruppiere(eigen)
|
||
self.assertEqual(d.nutzlast(eigen[0], erg[uf.finde(1)])["haeuser"], 1)
|
||
|
||
def test_einzelstueck_bekommt_eigene_gruppe(self):
|
||
items = [_item(1, *AGENTURMELDUNG)]
|
||
uf, erg = d.gruppiere(items)
|
||
last = d.nutzlast(items[0], erg[uf.finde(1)])
|
||
self.assertEqual(last["gruppengroesse"], 1)
|
||
self.assertEqual(last["leitartikel"], 1)
|
||
self.assertIsNone(last["jaccard_min"])
|
||
|
||
def test_nutzlast_haelt_den_textstand_fest(self):
|
||
items = [_item(1, *AGENTURMELDUNG)]
|
||
uf, erg = d.gruppiere(items)
|
||
last = d.nutzlast(items[0], erg[uf.finde(1)])
|
||
self.assertEqual(last["basis_hash"], "hash1")
|
||
self.assertEqual(last["basis"], "titel+teaser")
|
||
# und bei der anderen Basis steht auch die andere drin
|
||
nur_titel = [_item(1, *AGENTURMELDUNG, basis="titel")]
|
||
uf2, erg2 = d.gruppiere(nur_titel)
|
||
self.assertEqual(d.nutzlast(nur_titel[0], erg2[uf2.finde(1)])["basis"], "titel")
|
||
|
||
def test_nutzlast_ohne_langen_text(self):
|
||
# Abnahmekriterium: kein Text > 100 Zeichen in codings.nutzlast.
|
||
# Die Datenbank erzwingt es; hier faellt es frueher auf.
|
||
items = [_item(1, *AGENTURMELDUNG)]
|
||
uf, erg = d.gruppiere(items)
|
||
for wert in d.nutzlast(items[0], erg[uf.finde(1)]).values():
|
||
if isinstance(wert, str):
|
||
self.assertLessEqual(len(wert), 100)
|
||
|
||
|
||
# Festwert. Faellt dieser Test, hat sich das Verfahren geaendert und die
|
||
# coder_version muss steigen.
|
||
ERWARTETER_SIMHASH = "91c3912900958489"
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|