diff --git a/.gitignore b/.gitignore index de734a4..6e34306 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ .venv/ __pycache__/ .env +stichprobe-*.txt diff --git a/NOTES.md b/NOTES.md new file mode 100644 index 0000000..6f2a76e --- /dev/null +++ b/NOTES.md @@ -0,0 +1,156 @@ +# Befunde aus dem Dublettenmodul + +Gemessen am Bestand vom 4.–7.9.2026: 3407 Items, 12 Quellen, 106 Slots. + +## Das Verfahren der Spezifikation findet keine Übernahmen + +67 Gruppen, 147 Mitglieder, **null davon über Häuser hinweg**. Alle gefundenen +Gruppen sind Wiederholungen innerhalb eines Feeds. + +Das ist kein Schwellenproblem, das sich durch Nachjustieren lösen ließe. Ein +vollständiger Vergleich aller 599.004 hausübergreifenden Paare des stärksten +Tages (6.9., 1162 vergleichbare Items) ergibt: + +| Jaccard über Titel+Teaser | Paare über Häuser | +|---|---| +| ≥ 0.85 (Schwelle der Spec) | **0** | +| ≥ 0.7 | 1 | +| ≥ 0.5 | 3 | +| ≥ 0.3 | 8 | + +Der höchste erreichte Wert im gesamten Tag liegt bei 0.65. Die Schwelle wird +nie berührt. + +Die Hamming-Grenze ist dabei unschuldig: von 224 Paaren mit Jaccard ≥ 0.85 +verwirft sie 6. Der Vorfilter über SimHash-Bänder kostet also fast nichts — +er ist gut, er hat nur nichts zu tun. + +## Warum: der Titel wird übernommen, der Teaser nicht + +Die Übernahmen sind da, sie sehen nur anders aus als angenommen. Beispiele +vom 6.9., alle mit **zeichengleichem Titel** nach Normalisierung: + +``` +zeit / handelsblatt Wahl in Sachsen-Anhalt: Klüssendorf: Frage nach SPD-Vorsitz … +zeit / handelsblatt Ukraine-Krieg: Selenskyj mit wenig Hoffnung auf baldigen Frieden +heise / tagesschau Deutsche Isar-Aerospace-Rakete erfolgreich ins All gestartet +handelsblatt / tagesschau +++ Iran-Krieg +++: Tote und Verletzte bei neuen Angriffen +``` + +Die Häuser übernehmen die Agenturüberschrift wörtlich und schreiben den +Teaser selbst. Werden beide in einen Sack geworfen, verdünnt der Teaser genau +das Signal, das trägt: dieselben Paare landen bei Jaccard 0.29 bis 0.38. + +Gemessen, wenn **nur der Titel** verglichen wird (6.9., 1144 Items mit +mindestens fünf Titelwörtern): + +| Kriterium | Paare | davon über Häuser | +|---|---|---| +| Titel zeichengleich | 61 | **22** | +| Titel-Jaccard ≥ 0.75 | 16 | **9** | + +31 hausübergreifende Übernahmen an einem Tag statt null. Die Differenz +zwischen „zeichengleich" und „≥ 0.75" ist fast immer ein vorangestelltes +Ressortkürzel des Hauses: `Raumfahrt: Deutsche Isar-Aerospace-Rakete …` gegen +`Deutsche Isar-Aerospace-Rakete …`, oder `Notfälle:` gegen `Flugverkehr:` vor +derselben dpa-Zeile. + +## Vorschlag + +Vergleichsbasis auf den **Titel** umstellen, den Teaser nur noch als +Bestätigung heranziehen, wo er auf beiden Seiten vorliegt. Zusätzlich das +führende Ressortpräfix (`Wort:` am Anfang) in der Normalisierung abtrennen — +danach fällt ein Teil der 0.75er-Fälle in die Zeichengleichheit. + +Das ist eine Änderung an Abschnitt 3 der Spezifikation, nicht an ihrer Absicht: +„near-exact, kein thematisches Clustering" bleibt. Die Abnahmekriterien wären +neu zu setzen, weil sich mit der Basis auch die Grundgesamtheit ändert. + +## Umgesetzt, mit diesem Ergebnis + +Der Vorschlag ist gebaut und ist jetzt die Vorgabe (`--basis titel`, dazu das +abgetrennte Ressortkürzel). Auf demselben Bestand: + +| | Titel + Teaser | Titel | +|---|---|---| +| Gruppen gesamt | 67 | 135 | +| Mitglieder | 147 | 290 | +| **Gruppen über mehrere Häuser** | **0** | **63** | +| Items in solchen Gruppen | 0 | 143 | + +Gruppengrößen: 240 Paare, 33 Dreier, 12 Vierer, 5 Fünfer. Die größte ist +dieselbe Zeile über den Start der Isar-Aerospace-Rakete bei handelsblatt, +heise, tagesschau und zeit. + +Auf dem gemessenen Tag (6.9.) erreicht das Verfahren 42 von 45 +hausübergreifenden Paaren mit Jaccard ≥ 0.70. Die Schwellen kosten dabei +nichts: an der Jaccard-Schwelle gehen 0 Paare verloren, an der +Hamming-Grenze 0. Die drei fehlenden scheitern am SimHash-Bandvorfilter — +bei sechs bis acht Trigrammen je Titel ist SimHash ein grobes Werkzeug. Ein +Trigramm-Index als zweiter Kandidatenweg würde sie einfangen. + +Die Schwelle selbst ist unkritisch: zwischen Jaccard 0.70 und 0.85 liegt auf +diesem Bestand kein einziges Paar. Nach dem Abtrennen des Ressortkürzels sind +die Übernahmen entweder zeichengleich oder deutlich verschieden. + +## Nebenbefund: Volltext hilft hier nicht + +Auf den ersten Blick läge nahe, den Volltext zu nehmen, wo er da ist. Genau +das verbietet sich: die Abdeckung ist ungleich (faz 311/489, heise 174/196, +aber derstandard, welt, dlf, taz, sz jeweils 0), und ein Paar aus einem Haus +mit und einem ohne Volltext ist der häufigste Fall überhaupt. Eine je Item +verschiedene Vergleichsbasis macht genau die Paare unauffindbar, um die es +geht. + +## Nebenbefund: `agentur_vermutet` greift kaum + +Zwei Belege im gesamten Bestand, beide `reuters`. Die Agenturkennung steht in +RSS-Titeln und -Teasern praktisch nie; sie taucht erst im Volltext oder in der +Bildunterschrift auf. Das Feld bleibt, liefert aber auf RSS-Basis fast immer +`null`. + +## Was beim Aufräumen weggefallen ist + +Nach den Experimenten einmal gemessen, welche Regel auf den 3407 Items +überhaupt jemals greift: + +| Regel | Treffer | +|---|---| +| Ressortkürzel abtrennen | 1870 | +| Anführungszeichen/Striche vereinheitlichen | 654 | +| `+++ … +++` entfernen | 8 | +| NFKC | 4 | +| `Eilmeldung:` entfernen | **0** | +| Ortsmarke `Berlin (dpa) -` entfernen | **0** | +| HTML-Tags entfernen | **0** (auch im Teaser) | +| Entities auflösen im Titel | **0** | + +Entfernt wurden `_EILMELDUNG` und `_ORTSMARKE`. Beide stehen so in Abschnitt 3 +der Spezifikation, greifen aber auf RSS-Titeln nie — die Ortsmarke steht im +Volltext einer Agenturmeldung, nicht in der Überschrift. Kommt Volltext als +Quelle hinzu, gehören sie zurück. + +Das Entfernen von HTML-Tags bleibt trotz null Treffern: Phase 1 zieht die +`links` aus dem Teaser-HTML, HTML ist im Feed also grundsätzlich zu erwarten. +Der doppelte Entity-Durchlauf (vor *und* nach dem Tag-Entfernen) ist auf einen +reduziert. + +Ebenfalls raus: der ungenutzte `anzahl`-Parameter von `baender()`, zwei +ungenutzte Argumente von `stichprobe()`, der `version`-Parameter von +`eintragen()`, ein `__main__`-Block in `version.py`, der eine Version ohne +Moduleinstellungen ausgab und damit nie die echte. + +Zusammengelegt: `lade()` hatte am Ende selbst gruppiert, und `main()` rechnete +dieselbe Gruppierung anschließend ein zweites Mal. Beides steckt jetzt in +`lade_und_gruppiere()` — beim Slot-Lauf gehört es ohnehin zusammen, weil sich +erst an den Gruppen zeigt, wieviel geladen werden muss. + +Zwei Kommentarblöcke zur Vergleichsbasis widersprachen sich nach der +Umstellung; der ältere behauptete noch „verglichen wird immer Titel + Teaser". + +Die Schwellen selbst sind alle belegt. Von 778 Kandidatenpaaren verwirft das +Datumsfenster 223, die Regel „gleiches Haus, anderer Tag" 48 und die +Hamming-Grenze 330; 177 werden bestätigt. Nur die Jaccard-Bestätigung verwirft +nichts mehr — sie bleibt als Präzisionszusage, weil SimHash kollidieren kann. +Der Kettenschnitt trifft genau eine von 135 Komponenten und bleibt aus +demselben Grund. diff --git a/README.md b/README.md index edce8c3..daabfaa 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,10 @@ Spezifikation: [`phase_02_spec.md`](phase_02_spec.md). | `migrations/` | Schemaänderungen, aufsteigend anzuwenden | | `db.py` | Auflösung von `DATABASE_URL`, Verbindung | | `check_db.py` | Rauchtest: Anmeldung, Bestand, fehlende Voraussetzungen | +| `normalisierung.py` | Textnormalisierung und Trigramme, gemeinsam für alle Module | +| `version.py` | `coder_version` aus den Komponenten | +| `dubletten.py` | Modul 1: exakte Übernahmen | +| `test_dubletten.py` | Testfälle dazu, ohne Datenbank | ## Einrichtung @@ -43,24 +47,46 @@ psql "$DATABASE_URL" -c 'select name, angewandt_am from schema_migrationen order Die Zählung trägt das Präfix `p2-`, weil die Ingest-Migrationen eine eigene haben und beide auf dieselbe Datenbank laufen. -**Migrationen brauchen einen anderen Zugang als der Betrieb.** Die Tabellen -gehören `postgres`, nicht `wurzelwerk`; `alter table` scheitert sonst mit -`must be owner of table codings`. Im Dev-Container: +Zwei Migrationen brauchen einen Superuser, weil sie Rechte vergeben, die sie +selbst noch nicht haben: `p2-002` (`create extension vector` — pgvector ist +nicht als `trusted` gekennzeichnet) und `p2-003` (Besitzerwechsel). Im +Dev-Container über den lokalen Socket: ```sh -podman exec -i postgres psql -U postgres -d wurzelwerk -v ON_ERROR_STOP=1 -f - < migrations/p2-001-codings-mehrschichtig.sql +podman exec -i postgres psql -U postgres -d wurzelwerk -v ON_ERROR_STOP=1 -f - < migrations/p2-003-eigentuemer.sql ``` -Dass die Anwendungsrolle ihr eigenes Schema nicht besitzt, ist eine offene -Frage — entweder bleibt es so und Migrationen laufen als Administrator, oder -die Tabellen wechseln per `alter table … owner to wurzelwerk` den Besitzer. +Alle übrigen laufen über `DATABASE_URL`. Seit `p2-003` gehören die Tabellen +der Rolle `wurzelwerk` statt `postgres` — vorher scheiterte jedes `alter table` +mit `must be owner of table codings`, und die Datenbank war damit der einzige +Ausreißer auf der Instanz: `gitea` und `hedgedoc` besitzen ihre Tabellen +vollständig. Die Erweiterungen selbst (`pg_trgm`, `vector`) bleiben beim +Superuser, wo sie hingehören. ## Voraussetzungen an die Datenbank -`pgvector` fehlt im Image `postgres:latest`: `select * from pg_available_extensions -where name = 'vector'` bleibt leer, `create extension vector` scheitert. Ohne -sie kein `item_embeddings`. Das ist ein Image-Tausch auf `pgvector/pgvector:pg18`, -keine Migration. Die vier übrigen Module laufen ohne. +Das Embedding-Modul braucht `pgvector`. Im offiziellen Image `postgres:latest` +ist die Erweiterung nicht enthalten; der Postgres-Container läuft deshalb auf +`docker.io/pgvector/pgvector:pg18-trixie` — derselbe Build wie das offizielle +Image (`18.6-1.pgdg13+2`, Debian 13, glibc 2.41), nur mit pgvector zusätzlich. + +Der Tag ist mit Bedacht gewählt. Auf derselben Instanz liegen weitere Dienste; +ein Wechsel der Hauptversion oder der Distribution wäre ein Ausfall, ein +Wechsel der glibc ein Collation-Problem in den Textindizes aller Datenbanken +(`datlocprovider = 'c'`). `pg18-trixie` schließt beides aus und zieht nur +Minor-Updates nach. Die versionierten Tags (`0.8.4-pg18-trixie`) taugen dafür +nicht: sie hängen der Postgres-Minor-Version hinterher und hätten 18.6 auf +18.4 zurückgedreht. + +Das Anlegen der Erweiterung in `wurzelwerk` ist davon getrennt und geschieht +in `p2-002`, nicht beim Containerstart. + +Ein Vektor belegt 4100 Byte. Bei den derzeit rund 1400 Items pro Tag sind das +etwa 2 GB im Jahr, mit HNSW-Index eher das Doppelte — tragbar. Die in der +Spezifikation angenommenen 200–400 Items je Slot wären rund 38.000 am Tag und +damit etwa 55 GB im Jahr, **je `coder_version`**. Ein Modellwechsel verdoppelt +den Bedarf, bis der alte Bestand gelöscht ist. Sollte es eng werden, halbiert +`halfvec(1024)` den Platz. ## Grenzen des Datensatzes @@ -70,3 +96,82 @@ heise 174/196, aber derstandard, welt, dlf, taz, sz jeweils 0). Das ist keine Lücke, die sich schließen lässt, sondern eine Eigenschaft der Quellen. Module müssen auf Titel und Teaser allein tragfähig bleiben und dürfen die Vergleichsbasis nicht davon abhängig machen, ob ein Item zufällig Volltext hat. + +## Modul 1: Dubletten + +```sh +.venv/bin/python dubletten.py --backfill # Gesamtbestand +.venv/bin/python dubletten.py --slot 2026-09-07T08:00 # ein Poll-Slot +.venv/bin/python dubletten.py --backfill --trocken --stichprobe 200 +.venv/bin/python dubletten.py --backfill --basis titel+teaser # die andere Basis +.venv/bin/python -m unittest test_dubletten +``` + +Normalisierung, SimHash über Wort-Trigramme, Kandidaten über gemeinsame +SimHash-Bänder, Bestätigung über Jaccard, Union-Find. Jedes verarbeitete Item +bekommt eine Zeile, auch ein Einzelstück — sonst ließe sich „keine Dublette" +nicht von „nicht verarbeitet" unterscheiden. Unveränderte Zeilen werden nicht +angefasst, `kodiert_am` bleibt also stehen, wenn sich nichts geändert hat. + +Gesamtbestand (3407 Items) 0,2 s, ein Slot 0,4 s. Das Zeitbudget von 3 Minuten +je Slot ist für dieses Modul kein Thema. + +**Ergebnis auf dem Bestand vom 4.–7.9.2026:** 135 Gruppen mit 290 Mitgliedern, +davon 63 Gruppen mit 143 Items über mehrere Häuser hinweg — das ist der +eigentliche Zweck. Größte Gruppe: fünf Häuser mit derselben Zeile über den +Start der Isar-Aerospace-Rakete. + +### Vier Abweichungen von der Spezifikation + +**Vergleichsbasis ist der Titel, nicht Titel + Teaser.** Gemessen wurde beides. +Über Titel + Teaser findet das Verfahren auf diesem Bestand *keine einzige* +hausübergreifende Übernahme; über den Titel allein 63 Gruppen. Grund: die +Häuser übernehmen die Agenturüberschrift wörtlich und schreiben den Teaser +selbst, der Teaser verdünnt also genau das Signal, das trägt. Die Zahlen +stehen in `NOTES.md`. `--basis titel+teaser` schaltet auf die alte Basis +zurück; sie ergibt eine eigene `coder_version` und überschreibt nichts. + +**Das Ressortkürzel vor dem Titel wird abgetrennt.** `Raumfahrt: Deutsche +Rakete …` und `Deutsche Rakete …` sind dieselbe Meldung, `Notfälle:` und +`Flugverkehr:` stehen vor derselben dpa-Zeile. Die Regel ist eng gefasst: +höchstens vier Wörter vor dem Doppelpunkt, mindestens fünf danach — sonst +verlöre `Habeck: Wir haben uns geirrt` seine Aussage. + +**Die Gruppen-UUID kommt vom Leitartikel, nicht aus der Mitgliedermenge.** +Sonst änderte ein später eintreffendes Mitglied die Kennung einer Gruppe, die +Phase 3 bereits kodiert hat. Der Leitartikel ist das älteste Mitglied; ein +neues Mitglied ist zwangsläufig jünger und verschiebt das Minimum nicht. + +**Dasselbe Haus an verschiedenen Tagen wird nicht gepaart.** Wiederkehrende +Formate — „tagesschau in 100 Sekunden", „Wetter", „Die Nachrichten" — sind +zeichengleich, aber jeweils ein eigener Vorgang. Ohne diese Regel wuchsen sie +transitiv zu einer Gruppe aus 19 Sendungen über drei Tage zusammen. + +### Warum ein Slot-Lauf seinen Rand wachsen lässt + +Die Items eines Slots brauchen ihre Partner im Datumsfenster, diese Partner +ihrerseits ihr eigenes Fenster. Ein fester Rand genügt trotzdem nicht: +Übernahmen bilden Ketten — dieselbe Schlagzeile erscheint am Montag bei einem, +am Dienstag beim zweiten, am Mittwoch beim dritten Haus. Wird eine solche +Kette am geladenen Rand abgeschnitten, fällt die Gruppe anders aus als im +Backfill. + +Der Lauf lädt deshalb, gruppiert, und lädt erneut, solange eine Gruppe bis an +den Rand reicht. Geschrieben wird nur der innere Bereich, in dem jedes Item +seine vollständige Nachbarschaft gesehen hat. + +Nachgewiesen: nach einem Backfill ändern beliebig viele Slot-Läufe nichts mehr +(`neu 0, geändert 0`), und ein anschließender Backfill ebenso wenig. Live-Lauf +und Neuaufbau liefern dasselbe Ergebnis — Invariante 2 und 4. + +### Was bleibt + +Auf dem gemessenen Tag findet das Verfahren 42 von 45 hausübergreifenden +Paaren. Die drei fehlenden gehen am SimHash-Bandvorfilter verloren, nicht an +den Schwellen — bei sechs bis acht Trigrammen je Titel ist SimHash grob. Ein +Trigramm-Index als zweiter Kandidatenweg würde sie einfangen; das ist nicht +gebaut, weil die Spezifikation Präzision über Recall stellt. + +Die Stichprobe (`--stichprobe 200`) schreibt die Paare zum Prüfen von Hand +heraus, die zweifelhaftesten zuerst. Sie enthält Titel im Klartext und ist +deshalb nicht eingecheckt. diff --git a/check_db.py b/check_db.py index ec6c7f4..18b1433 100755 --- a/check_db.py +++ b/check_db.py @@ -22,9 +22,18 @@ def main(): erw = [r[0] for r in conn.execute("select extname from pg_extension order by 1")] print(f" Extensions: {', '.join(erw)}") - for fehlend in ("vector",): - if fehlend not in erw: - print(f" fehlt noch fuer Phase 2: {fehlend}") + + # Zwischen "im Image nicht vorhanden" und "nur noch nicht angelegt" + # unterscheiden - das erste ist ein Image-Tausch, das zweite eine + # Migration. + if "vector" not in erw: + verfuegbar = conn.execute( + "select default_version from pg_available_extensions where name = 'vector'" + ).fetchone() + if verfuegbar: + print(f" vector {verfuegbar[0]} liegt bereit, ist aber nicht angelegt") + else: + print(" vector fehlt im Image - Phase 2 kann keine Embeddings ablegen") print("\nTabellen") for name, in conn.execute( diff --git a/db.py b/db.py index 8d67510..39540c8 100644 --- a/db.py +++ b/db.py @@ -11,12 +11,12 @@ import pathlib _ENV = pathlib.Path(__file__).with_name(".env") -def _aus_env_datei(pfad=_ENV): +def _aus_env_datei(): """Minimaler .env-Leser. Kein Shell-Parsing, keine Ersetzungen.""" werte = {} - if not pfad.exists(): + if not _ENV.exists(): return werte - for zeile in pfad.read_text(encoding="utf-8").splitlines(): + for zeile in _ENV.read_text(encoding="utf-8").splitlines(): zeile = zeile.strip() if not zeile or zeile.startswith("#") or "=" not in zeile: continue diff --git a/dubletten.py b/dubletten.py new file mode 100755 index 0000000..53e3e63 --- /dev/null +++ b/dubletten.py @@ -0,0 +1,659 @@ +#!/usr/bin/env python3 +"""Phase 2, Modul 1: exakte Uebernahmen (ebene = 'dublette'). + +Agenturmeldungen erscheinen bei mehreren Haeusern nahezu gleichlautend. +Dieses Modul fasst sie zusammen, damit Phase 3 nur den Leitartikel kodiert. + +Kein thematisches Clustering. Nur near-exact: erkannt wird, dass zwei Haeuser +denselben Text ausgespielt haben, nicht dass sie ueber dasselbe schreiben. + + python3 dubletten.py --backfill # Gesamtbestand + python3 dubletten.py --slot 2026-09-07T08:00 + python3 dubletten.py --backfill --trocken --stichprobe 200 + +Verfahren nach Abschnitt 3 der Spezifikation: Normalisierung, SimHash ueber +Wort-Trigramme, Kandidaten ueber Hamming-Distanz, Bestaetigung ueber Jaccard, +Union-Find. + +Wo das Verfahren von der Spezifikation abweicht, steht die Begruendung an +Ort und Stelle: BASIS_VORGABE, gruppen_id(), die Paarregel in paare() und +_zerlege_ketten(). README.md fasst sie zusammen, NOTES.md haelt die Messungen +fest, auf denen sie beruhen. +""" + +import argparse +import datetime as dt +import hashlib +import itertools +import re +import sys +import time +import uuid + +from psycopg.types.json import Jsonb + +import normalisierung +import version as versionsmodul +from db import verbindung + +EBENE = "dublette" + +# -------------------------------------------------------------------------- +# Vergleichsbasis: "titel" oder "titel+teaser". +# +# Die Spezifikation sagt "Titel + Teaser, oder Volltext falls vorhanden". +# Beides taugt hier nicht. +# +# Volltext nicht, weil ihn nur ein Teil der Quellen liefert - Ingest liest +# RSS, und viele Haeuser spielen dort kein content:encoded aus (Stand +# 7.9.2026: faz 311/489, heise 174/196, aber derstandard, welt, dlf, taz, sz +# jeweils 0). Eine je Item verschiedene Basis macht ausgerechnet den +# haeufigsten Fall unauffindbar: dieselbe Meldung bei einem Haus mit und bei +# einem ohne Volltext. +# +# Titel + Teaser nicht, weil die Haeuser die Agenturueberschrift woertlich +# uebernehmen, den Teaser aber selbst schreiben. Gemessen wurde beides: ueber +# Titel + Teaser findet das Verfahren auf diesem Bestand keine einzige +# hausuebergreifende Uebernahme, ueber den Titel allein 63 Gruppen. Der +# Teaser verduennt genau das Signal, das traegt. Zahlen in NOTES.md. +# -------------------------------------------------------------------------- +BASIS_VORGABE = "titel" + +# Mindestwortzahl je Basis. Wenige Trigramme erreichen leicht einen hohen +# Jaccard-Wert, ohne dass die Meldungen etwas miteinander zu tun haetten - +# lieber eine Dublette verpassen als zwei Vorgaenge verschmelzen +# (Abschnitt 3, "Erwartung"). Titel sind kuerzer, die Schwelle entsprechend. +MINDESTWOERTER = {"titel": 5, "titel+teaser": 8} + +# Reissleine fuer die Randerweiterung beim Slot-Lauf. +RANDERWEITERUNGEN = 6 + +EINSTELLUNGEN = { + "basis": BASIS_VORGABE, + "simhash_bits": 64, + "trigramm": "wort-3", + "hamming_max": 3, + "jaccard_min": 0.85, + "fenster_tage": 1, + "mindestwoerter": MINDESTWOERTER[BASIS_VORGABE], + "ressort_abgetrennt": True, +} + +_AGENTUREN = [ + ("dpa", re.compile(r"\((?:dpa|dpa-AFX)[^)]{0,20}\)|\bdpa\b")), + ("afp", re.compile(r"\(AFP[^)]{0,20}\)|\bAFP\b")), + ("reuters", re.compile(r"\bReuters\b")), + ("ap", re.compile(r"\(AP\)|\bAssociated Press\b")), + ("epd", re.compile(r"\bepd\b")), + ("kna", re.compile(r"\bKNA\b")), + ("sid", re.compile(r"\bSID\b")), + ("dts", re.compile(r"\bdts Nachrichtenagentur\b")), +] + +# Fester Namensraum, damit Gruppen-UUIDs ueber Laeufe und Maschinen hinweg +# gleich bleiben. Willkuerlich gewaehlt, aber unveraenderlich. +NAMENSRAUM = uuid.UUID("6f9b4d2e-1c53-4a7f-9e18-2b0d7c5a3e41") + + +# -------------------------------------------------------------------------- +# SimHash +# -------------------------------------------------------------------------- +def _merkmalshash(trigramm): + """64 Bit je Trigramm. + + blake2b und nicht hash(): Pythons hash() fuer Zeichenketten ist je + Prozess zufaellig gesalzen. Damit waere nichts reproduzierbar. + """ + roh = "\x1f".join(trigramm).encode("utf-8") + return int.from_bytes(hashlib.blake2b(roh, digest_size=8).digest(), "big") + + +def simhash(trigramme): + if not trigramme: + return 0 + gewichte = [0] * 64 + for tri in trigramme: + h = _merkmalshash(tri) + for bit in range(64): + gewichte[bit] += 1 if (h >> bit) & 1 else -1 + wert = 0 + for bit in range(64): + # Gleichstand faellt auf 0. Willkuerlich, aber deterministisch. + if gewichte[bit] > 0: + wert |= 1 << bit + return wert + + +def hamming(a, b): + return (a ^ b).bit_count() + + +BAENDER = 4 + + +def baender(wert): + """SimHash in vier Stuecke zu 16 Bit zerlegen. + + Schubfachschluss: unterscheiden sich zwei Werte in hoechstens 3 Bits, so + ist bei vier Baendern mindestens eines gleich. Nur Paare, die sich ein + Band teilen, muessen ueberhaupt verglichen werden - das ersetzt den + Vergleich jeder Meldung mit jeder. Die Zahl der Baender haengt an + hamming_max und darf nicht ohne dieses geaendert werden. + """ + breite = 64 // BAENDER + maske = (1 << breite) - 1 + return tuple((i, (wert >> (i * breite)) & maske) for i in range(BAENDER)) + + +def jaccard(a, b): + if not a or not b: + return 0.0 + schnitt = len(a & b) + return schnitt / (len(a) + len(b) - schnitt) + + +class UnionFind: + def __init__(self): + self.eltern = {} + + def finde(self, x): + self.eltern.setdefault(x, x) + while self.eltern[x] != x: + self.eltern[x] = self.eltern[self.eltern[x]] + x = self.eltern[x] + return x + + def vereinige(self, a, b): + wa, wb = self.finde(a), self.finde(b) + if wa != wb: + # Kleinere Wurzel gewinnt: macht das Ergebnis unabhaengig von der + # Reihenfolge, in der die Paare hereinkommen. + hoch, tief = max(wa, wb), min(wa, wb) + self.eltern[hoch] = tief + + +def agentur(*texte): + """Agentur nur mit Textbeleg. Keine Vermutung ins Blaue.""" + text = " ".join(t for t in texte if t) + for name, muster in _AGENTUREN: + if muster.search(text): + return name + return None + + +def gruppen_id(leitartikel_id): + """UUID aus dem Leitartikel, nicht aus der Mitgliedermenge. + + Die Spezifikation leitet sie aus der sortierten Menge der raw_item_id ab. + Das haelt nicht: erscheint dieselbe Meldung drei Stunden spaeter bei einem + weiteren Haus, aendert sich die Menge und damit die UUID - eine Gruppe, + die Phase 3 bereits kodiert hat, hiesse ploetzlich anders. + + Der Leitartikel ist das aeltestgesehene Mitglied. Kommt ein Mitglied + hinzu, ist es zwangslaeufig juenger und aendert das Minimum nicht. Die + Kennung bleibt damit stabil, solange die Gruppe waechst. Sie wechselt + nur, wenn zwei bestehende Gruppen verschmelzen - dann ist es richtig, + dass sich etwas aendert. + """ + return str(uuid.uuid5(NAMENSRAUM, f"dublette:{leitartikel_id}")) + + +# -------------------------------------------------------------------------- +# Laden +# -------------------------------------------------------------------------- +class Item: + __slots__ = ("id", "quelle", "titel", "teaser", "gesehen", "datum", + "inhalt_hash", "basis", "text", "trigramme", "simhash", "woerter") + + def __init__(self, id, quelle, titel, teaser, gesehen, datum, inhalt_hash, + basis=None): + self.id = id + self.quelle = quelle + self.titel = titel + self.teaser = teaser + self.gesehen = gesehen + self.datum = datum + self.inhalt_hash = inhalt_hash + self.basis = basis or BASIS_VORGABE + if self.basis == "titel": + # Ressortkuerzel abtrennen: dieselbe Agenturzeile bekommt bei + # jedem Haus ein anderes vorangestellt, oder gar keins. + self.text = normalisierung.normalisiere(titel, ressort_abtrennen=True) + else: + self.text = normalisierung.normalisiere(titel, teaser) + self.woerter = len(normalisierung.woerter(self.text)) + self.trigramme = normalisierung.trigramme(self.text) + self.simhash = simhash(self.trigramme) + + @property + def vergleichbar(self): + return self.woerter >= MINDESTWOERTER[self.basis] + + +_SPALTEN = """id, quelle, titel, teaser, abgerufen_am, + coalesce(pubdate, abgerufen_am)::date as datum, inhalt_hash""" + + +def lade_bereich(conn, basis, von=None, bis=None): + """Items eines Datumsbereichs, ohne Grenzen der Gesamtbestand. + + Das Datum kommt aus pubdate mit Rueckfall auf abgerufen_am: Phase 1 + vermerkt ausdruecklich, dass pubdate fehlen oder luegen kann. + """ + if von is None: + zeilen = conn.execute(f"select {_SPALTEN} from raw_items order by id") + else: + zeilen = conn.execute( + f"""select {_SPALTEN} from raw_items + where coalesce(pubdate, abgerufen_am)::date between %s and %s + order by id""", (von, bis)) + return [Item(*z, basis=basis) for z in zeilen] + + +def _beruehrt_rand(ergebnis, lade_von, lade_bis): + """Reicht eine Gruppe bis an den Rand des geladenen Bereichs? + + Dann kann jenseits davon noch ein Mitglied liegen, und die Gruppe ist + unvollstaendig gerechnet. + """ + for gruppe in ergebnis.values(): + if len(gruppe["mitglieder"]) < 2: + continue + for m in gruppe["mitglieder"]: + if m.datum <= lade_von or m.datum >= lade_bis: + return True + return False + + +def lade_und_gruppiere(conn, slot=None, fenster_tage=None, basis=None, **grenzen): + """Liefert (items, uf, ergebnis, schreibbereich). + + Laden und Gruppieren haengen beim Slot-Lauf zusammen und sind darum eine + Einheit: wieviel geladen werden muss, zeigt sich erst an den Gruppen. + + Ohne slot der Gesamtbestand, dann ist alles beschreibbar. + + Mit slot wird ein Rand mitgeladen und nur der innere Bereich beschrieben - + dort, wo jedes Item seine vollstaendige Nachbarschaft gesehen hat. Eine + unvollstaendig gerechnete Zeile darf eine vollstaendige nicht + ueberschreiben. + + Ein fester Rand genuegt dabei nicht. Uebernahmen bilden Ketten: dieselbe + Schlagzeile erscheint am Montag bei einem, am Dienstag beim zweiten, am + Mittwoch beim dritten Haus. Wo eine Kette am geladenen Rand abgeschnitten + wird, faellt die Gruppe anders aus als im Backfill - und genau das darf + nicht sein (Invariante 2 und 4). Also wird geladen, gruppiert, und wenn + eine Gruppe bis an den Rand reicht, weiter geladen. In der Praxis laeuft + das ein- bis zweimal. + """ + fenster_tage = EINSTELLUNGEN["fenster_tage"] if fenster_tage is None else fenster_tage + + if slot is None: + items = lade_bereich(conn, basis) + uf, ergebnis = gruppiere(items, fenster_tage=fenster_tage, **grenzen) + return items, uf, ergebnis, None + + von, bis = conn.execute( + """select min(coalesce(pubdate, abgerufen_am)::date), + max(coalesce(pubdate, abgerufen_am)::date) + from raw_items where slot = %s""", (slot,)).fetchone() + if von is None: + return [], None, None, None + + fenster = dt.timedelta(days=fenster_tage) + schreibbereich = (von - fenster, bis + fenster) + + rand = 2 * fenster_tage + for versuch in range(RANDERWEITERUNGEN): + lade_von = von - dt.timedelta(days=rand) + lade_bis = bis + dt.timedelta(days=rand) + items = lade_bereich(conn, basis, lade_von, lade_bis) + uf, ergebnis = gruppiere(items, fenster_tage=fenster_tage, **grenzen) + if not _beruehrt_rand(ergebnis, lade_von, lade_bis): + return items, uf, ergebnis, schreibbereich + rand += fenster_tage + + # Reissleine: lieber den ganzen Bestand rechnen als etwas Falsches + # schreiben. Bisher nie erreicht. + items = lade_bereich(conn, basis) + uf, ergebnis = gruppiere(items, fenster_tage=fenster_tage, **grenzen) + return items, uf, ergebnis, schreibbereich + + +# -------------------------------------------------------------------------- +# Gruppieren +# -------------------------------------------------------------------------- +def paare(items, fenster_tage=EINSTELLUNGEN["fenster_tage"], + hamming_max=EINSTELLUNGEN["hamming_max"], + jaccard_min=EINSTELLUNGEN["jaccard_min"]): + """Bestaetigte Paare finden. + + Erst Kandidaten ueber gemeinsame SimHash-Baender, dann die drei Pruefungen + aus der Spezifikation: Datumsnaehe, Hamming-Distanz, Jaccard-Wert. + """ + nach_band = {} + brauchbar = [i for i in items if i.vergleichbar] + for item in brauchbar: + for band in baender(item.simhash): + nach_band.setdefault(band, []).append(item) + + gesehen = set() + fenster = dt.timedelta(days=fenster_tage) + for eimer in nach_band.values(): + if len(eimer) < 2: + continue + for a, b in itertools.combinations(eimer, 2): + schluessel = (a.id, b.id) if a.id < b.id else (b.id, a.id) + if schluessel in gesehen: + continue + gesehen.add(schluessel) + if abs(a.datum - b.datum) > fenster: + continue + # Dasselbe Haus an verschiedenen Tagen ist keine Uebernahme, + # sondern ein wiederkehrendes Format: "tagesschau in 100 + # Sekunden", "Wetter", "Die Nachrichten". Gleicher Text, anderer + # Vorgang - kein Textverfahren kann die auseinanderhalten, also + # muss die Regel es tun. + # + # Die Regel sitzt bewusst hier auf Paarebene und nicht auf der + # fertigen Gruppe: sie haengt allein an den beiden Items und + # faellt damit gleich aus, egal wieviel Bestand geladen ist. + # Eine Regel auf Gruppenebene waere fensterabhaengig - und dann + # lieferte ein Slot-Lauf andere Gruppen als ein Backfill. + if a.quelle == b.quelle and a.datum != b.datum: + continue + if hamming(a.simhash, b.simhash) > hamming_max: + continue + # Bestaetigung. Auf dem Bestand vom 4.-7.9.2026 verwirft sie + # kein einziges Paar mehr, das die Hamming-Grenze passiert hat - + # sie ist trotzdem die eigentliche Praezisionszusage. SimHash + # kann kollidieren, Jaccard nicht. + j = jaccard(a.trigramme, b.trigramme) + if j < jaccard_min: + continue + yield a, b, j + + +def _zerlege_ketten(gruppe, fenster_tage): + """Gruppen aufbrechen, die nur ueber eine Kette zusammenhaengen. + + Union-Find ist transitiv, das Datumsfenster ist es nicht: A passt zu B, + B zu C, und schon haengt A mit C zusammen, obwohl mehr als ein Tag + dazwischen liegt. Eine Uebernahme ist ein Vorgang, der bei mehreren + Haeusern gleichzeitig erscheint - keine Schlagzeile, die drei Tage lang + weitergereicht wird. Eine Gruppe darf darum nicht mehr Tage umspannen + als das Fenster. + + Auf dem Bestand vom 4.-7.9.2026 trifft das genau eine von 135 + Komponenten: dieselbe Newsblog-Zeile bei handelsblatt (4.9.), sz (5.9.) + und tagesschau (6.9.). Der Schnitt ist also kein Massenphaenomen, + sondern verhindert, dass Phase 3 den Stand vom Mittwoch mit der + Kodierung vom Montag versieht. + + Getrennt wird nach Datum, damit das Ergebnis nicht von der + Ladereihenfolge abhaengt. + """ + spanne = dt.timedelta(days=fenster_tage) + geordnet = sorted(gruppe, key=lambda i: (i.datum, i.gesehen, i.id)) + teile, aktuell = [], [geordnet[0]] + for item in geordnet[1:]: + if item.datum - aktuell[0].datum > spanne: + teile.append(aktuell) + aktuell = [item] + else: + aktuell.append(item) + teile.append(aktuell) + return teile + + +def gruppiere(items, fenster_tage=EINSTELLUNGEN["fenster_tage"], **grenzen): + """Union-Find ueber die bestaetigten Paare. + + Liefert je Item die Gruppe, ihre Mitglieder, den Leitartikel und den + kleinsten Jaccard-Wert, der die Gruppe zusammenhaelt. + """ + uf = UnionFind() + schwaechste = {} + for a, b, j in paare(items, fenster_tage=fenster_tage, **grenzen): + uf.vereinige(a.id, b.id) + schwaechste[(a.id, b.id)] = j + + roh = {} + for item in items: + roh.setdefault(uf.finde(item.id), []).append(item) + + # Ketten aufbrechen und die Zugehoerigkeit neu festschreiben. Danach ist + # uf.finde() wieder die Wahrheit ueber die Gruppen. + mitglieder = {} + for gruppe in roh.values(): + for teil in _zerlege_ketten(gruppe, fenster_tage): + wurzel = min(i.id for i in teil) + for item in teil: + uf.eltern[item.id] = wurzel + mitglieder[wurzel] = teil + + ergebnis = {} + for wurzel, gruppe in mitglieder.items(): + # Leitartikel: aeltestes gesehenes Item, bei Gleichstand die kleinste + # id. Rein deterministisch, unabhaengig von der Ladereihenfolge. + leit = min(gruppe, key=lambda i: (i.gesehen, i.id)) + ids = {i.id for i in gruppe} + werte = [j for (x, y), j in schwaechste.items() if x in ids and y in ids] + ergebnis[wurzel] = { + "mitglieder": gruppe, + "leitartikel": leit, + "jaccard_min": round(min(werte), 4) if werte else None, + } + return uf, ergebnis + + +def nutzlast(item, gruppe): + leit = gruppe["leitartikel"] + groesse = len(gruppe["mitglieder"]) + return { + "gruppe": gruppen_id(leit.id), + "gruppengroesse": groesse, + "leitartikel": leit.id, + "simhash": f"{item.simhash:016x}", + "jaccard_min": gruppe["jaccard_min"], + "agentur_vermutet": agentur(item.titel, item.teaser), + # Basis und Stand des Texts, gegen den kodiert wurde. Ohne das laesst + # sich spaeter nicht sagen, ob eine Kodierung noch zum Item passt: + # raw_items haelt den *aktuellen* Titel, Redaktionen aendern ihn nach + # (in diesem Bestand 3 von 10 Items), und dann ist der Vergleich von + # gestern gegen einen Text gelaufen, den es so nicht mehr gibt. + "basis": item.basis, + "basis_hash": item.inhalt_hash, + "woerter": item.woerter, + "vergleichbar": item.vergleichbar, + # Eine Gruppe aus einem einzigen Haus ist keine Uebernahme, sondern + # eine Wiederholung im selben Feed. Fuer Phase 3 ist der Unterschied + # wesentlich: bei einer Uebernahme erben die Mitglieder die Kodierung + # des Leitartikels, eine Wiederholung ist derselbe Vorgang zweimal. + "haeuser": len({i.quelle for i in gruppe["mitglieder"]}), + } + + +# -------------------------------------------------------------------------- +# Schreiben +# -------------------------------------------------------------------------- +def schreibe(conn, items, ergebnis, uf, coder_version, trocken=False, schreibbereich=None): + """Codings ablegen, aber nur wo sich etwas geaendert hat. + + Jedes verarbeitete Item bekommt eine Zeile, auch ein Einzelstueck: sonst + liesse sich "keine Dublette" nicht von "nicht verarbeitet" unterscheiden. + + Unveraenderte Zeilen werden nicht angefasst. Ein Slot-Lauf rechnet ueber + das ganze Datumsfenster, und ohne diesen Vergleich schriebe er bei jedem + Durchgang einige tausend Zeilen neu - mit neuem kodiert_am, ohne dass + sich etwas geaendert haette. + """ + vorhanden = { + r[0]: r[1] + for r in conn.execute( + "select raw_item_id, nutzlast from codings" + " where coder_version = %s and ebene = %s", (coder_version, EBENE)) + } + + if schreibbereich is None: + zu_schreiben = items + else: + von, bis = schreibbereich + zu_schreiben = [i for i in items if von <= i.datum <= bis] + + neu = geaendert = unveraendert = 0 + stapel = [] + for item in zu_schreiben: + gruppe = ergebnis[uf.finde(item.id)] + last = nutzlast(item, gruppe) + alt = vorhanden.get(item.id) + if alt == last: + unveraendert += 1 + continue + if alt is None: + neu += 1 + else: + geaendert += 1 + stapel.append((item.id, coder_version, EBENE, last["jaccard_min"], Jsonb(last))) + + if stapel and not trocken: + with conn.cursor() as cur: + cur.executemany( + """insert into codings (raw_item_id, coder_version, ebene, konfidenz, nutzlast) + values (%s, %s, %s, %s, %s) + on conflict (raw_item_id, coder_version, ebene) + do update set nutzlast = excluded.nutzlast, + konfidenz = excluded.konfidenz, + kodiert_am = now()""", + stapel) + return {"neu": neu, "geaendert": geaendert, "unveraendert": unveraendert, + "ausserhalb": len(items) - len(zu_schreiben)} + + +def protokolliere(conn, slot, coder_version, dauer_ms, gesehen, kodiert, fehler=None): + conn.execute( + """insert into p2_laeufe (slot, modul, coder_version, dauer_ms, + items_gesehen, items_kodiert, items_fehler, fehler) + values (%s, %s, %s, %s, %s, %s, %s, %s)""", + (slot, EBENE, coder_version, dauer_ms, gesehen, kodiert, 0 if not fehler else 1, fehler)) + + +# -------------------------------------------------------------------------- +# Bericht und Stichprobe +# -------------------------------------------------------------------------- +def bericht(items, ergebnis): + gruppen = [g for g in ergebnis.values() if len(g["mitglieder"]) > 1] + in_gruppen = sum(len(g["mitglieder"]) for g in gruppen) + ueber = [g for g in gruppen if len({i.quelle for i in g["mitglieder"]}) > 1] + kurz = sum(1 for i in items if not i.vergleichbar) + mindest = MINDESTWOERTER[items[0].basis] if items else 0 + print(f" Items: {len(items)}") + print(f" zu kurz: {kurz} (< {mindest} Woerter, nicht verglichen)") + print(f" Uebernahmegruppen:{len(gruppen):>4} mit {in_gruppen} Mitgliedern") + print(f" davon ueber mehrere Haeuser: {len(ueber)} <- der eigentliche Zweck") + if gruppen: + groessen = sorted((len(g["mitglieder"]) for g in gruppen), reverse=True) + print(f" groesste Gruppe: {groessen[0]}") + for g in sorted(gruppen, key=lambda g: -len(g["mitglieder"]))[:5]: + haeuser = sorted({i.quelle for i in g["mitglieder"]}) + leit = g["leitartikel"] + print(f" [{len(g['mitglieder'])}] j={g['jaccard_min']} {', '.join(haeuser)}") + print(f" {leit.titel[:88]}") + + +def stichprobe(ergebnis, anzahl, pfad): + """Paare zum Nachprüfen von Hand ausschreiben. + + Die Spezifikation verlangt Praezision >= 0.98 auf 200 handgeprueften + Paaren. Pruefen kann das nur ein Mensch; diese Datei ist die Vorlage + dafuer. Sie enthaelt Titel im Klartext und gehoert deshalb nicht in die + Datenbank, sondern neben sie. + """ + zeilen = [] + for gruppe in ergebnis.values(): + m = gruppe["mitglieder"] + if len(m) < 2: + continue + for a, b in itertools.combinations(sorted(m, key=lambda i: i.id), 2): + zeilen.append((jaccard(a.trigramme, b.trigramme), + hamming(a.simhash, b.simhash), a, b)) + zeilen.sort(key=lambda z: (z[0], z[3].id)) # schwaechste zuerst - dort sitzen die Fehler + with open(pfad, "w", encoding="utf-8") as f: + f.write("# Dubletten-Stichprobe. Spalte 'urteil' von Hand fuellen: j = Dublette, n = keine.\n") + f.write("# Absteigend nach Zweifel sortiert: die obersten Paare sind die knappsten.\n\n") + for j, h, a, b in zeilen[:anzahl]: + f.write(f"urteil=_ jaccard={j:.3f} hamming={h}\n") + f.write(f" {a.id:>7} {a.quelle:<14}{a.titel}\n") + f.write(f" {b.id:>7} {b.quelle:<14}{b.titel}\n\n") + return min(len(zeilen), anzahl), len(zeilen) + + +def main(argv=None): + p = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + p.add_argument("--slot", help="ein Slot, z.B. 2026-09-07T08:00") + p.add_argument("--backfill", action="store_true", help="Gesamtbestand") + p.add_argument("--basis", choices=("titel", "titel+teaser"), default=BASIS_VORGABE, + help=f"Vergleichsbasis (Vorgabe {BASIS_VORGABE})") + p.add_argument("--jaccard", type=float, default=EINSTELLUNGEN["jaccard_min"], + help="Bestaetigungsschwelle (Vorgabe %(default)s)") + p.add_argument("--hamming", type=int, default=EINSTELLUNGEN["hamming_max"], + help="hoechste Hamming-Distanz fuer Kandidaten (Vorgabe %(default)s)") + p.add_argument("--fenster", type=int, default=1, help="Datumsfenster in Tagen (Vorgabe 1)") + p.add_argument("--trocken", action="store_true", help="rechnen, nichts schreiben") + p.add_argument("--stichprobe", type=int, metavar="N", help="N Paare zum Nachpruefen ausschreiben") + p.add_argument("--stichprobe-datei", default="stichprobe-dubletten.txt") + a = p.parse_args(argv) + + if not a.backfill and not a.slot: + p.error("--backfill oder --slot angeben") + + slot = dt.datetime.fromisoformat(a.slot) if a.slot else None + begonnen = time.monotonic() + + einstellungen = dict(EINSTELLUNGEN, + basis=a.basis, + jaccard_min=a.jaccard, + hamming_max=a.hamming, + fenster_tage=a.fenster, + mindestwoerter=MINDESTWOERTER[a.basis]) + + with verbindung() as conn: + # Die Einstellungen gehen in den Hash ein: eine andere Basis oder eine + # andere Schwelle ist eine andere Kodierung und darf die vorhandene + # nicht ueberschreiben. + v = versionsmodul.eintragen( + conn, komponenten=versionsmodul.komponenten(dubletten=einstellungen)) + print(f"coder_version {v} basis={a.basis} j>={a.jaccard} h<={a.hamming}" + f"{' (trocken)' if a.trocken else ''}") + + items, uf, ergebnis, schreibbereich = lade_und_gruppiere( + conn, slot=slot, fenster_tage=a.fenster, basis=a.basis, + jaccard_min=a.jaccard, hamming_max=a.hamming) + if not items: + print(" nichts zu tun") + return 0 + bericht(items, ergebnis) + + zahlen = schreibe(conn, items, ergebnis, uf, v, trocken=a.trocken, + schreibbereich=schreibbereich) + dauer = int((time.monotonic() - begonnen) * 1000) + print(f" codings: neu {zahlen['neu']}, geaendert {zahlen['geaendert']}," + f" unveraendert {zahlen['unveraendert']}" + + (f", Kranz uebergangen {zahlen['ausserhalb']}" if zahlen["ausserhalb"] else "")) + print(f" Dauer: {dauer} ms") + + if a.stichprobe: + n, gesamt = stichprobe(ergebnis, a.stichprobe, a.stichprobe_datei) + print(f" Stichprobe: {n} von {gesamt} Paaren nach {a.stichprobe_datei}") + + if not a.trocken: + protokolliere(conn, slot, v, dauer, len(items), + zahlen["neu"] + zahlen["geaendert"]) + conn.commit() + else: + conn.rollback() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/migrations/p2-002-embeddings.sql b/migrations/p2-002-embeddings.sql new file mode 100644 index 0000000..6c06b90 --- /dev/null +++ b/migrations/p2-002-embeddings.sql @@ -0,0 +1,87 @@ +-- Migration p2-002: pgvector und Ablage der Item-Embeddings +-- +-- psql "$ADMIN_DATABASE_URL" -v ON_ERROR_STOP=1 -f migrations/p2-002-embeddings.sql +-- +-- Setzt p2-001 voraus und ein Image, das pgvector mitbringt (siehe README): +-- docker.io/pgvector/pgvector:pg18-trixie. Ohne die Erweiterung bricht die +-- Migration in der ersten Zeile ab und hinterlaesst nichts. +-- +-- Die Embeddings sind Vorfilter fuer die Kandidatengenerierung in Phase 3b, +-- kein Selbstzweck. Phase 2 clustert nicht. + +begin; + +create extension if not exists vector; + +-- -------------------------------------------------------------------------- +-- Der Vektor liegt bewusst nicht in codings.nutzlast: JSONB-Vektoren sind +-- weder indizierbar noch platzsparend. Die Coding-Zeile der Schicht +-- 'embedding' haelt nur die Kennung, der Vektor steht hier. +-- +-- 1024 Dimensionen deckt beide in der Spezifikation genannten Modelle ab +-- (multilingual-e5-large, jina-embeddings-v3). Ein Modell mit anderer +-- Dimension braucht eine eigene Migration - das ist gewollt, denn es waere +-- ohnehin eine neue coder_version. +-- -------------------------------------------------------------------------- +create table if not exists item_embeddings ( + raw_item_id bigint not null references raw_items(id) on delete cascade, + coder_version text not null, + erzeugt_am timestamptz not null default now(), + vektor vector(1024) not null, + + primary key (raw_item_id, coder_version) +); + +comment on table item_embeddings is + 'Satz-Embedding je Item und Kodierer-Version. Vorfilter fuer Phase 3b'; +comment on column item_embeddings.coder_version is + 'Ohne Fremdschluessel auf coder_versionen, wie codings auch - die Herkunft ' + 'steht dort, erzwungen wird sie nicht'; + +-- Der Fremdschluessel auf raw_items ist nicht Kosmetik: ohne ihn bleiben +-- Vektoren als Waisen liegen, wenn ein Item aus raw_items verschwindet. + +-- -------------------------------------------------------------------------- +-- Kein globaler HNSW-Index. +-- +-- Ein Index ueber alle coder_version hinweg legt Vektoren verschiedener +-- Modelle in dieselbe Nachbarschaftsstruktur. Eine Suche muss dann nach +-- Version filtern, und pgvector filtert bei HNSW erst *nach* dem Durchlauf +-- der Kandidatenliste - das kostet Treffer, nicht nur Zeit. +-- +-- Stattdessen ein partieller Index je Version, angelegt wenn die Version in +-- Betrieb geht: +-- +-- select p2_embedding_index('p2-2026-09-07-a3f1b2c4'); +-- +-- Der Aufbau ist speicherhungrig; bei grossen Bestaenden vorher +-- set maintenance_work_mem = '2GB'; +-- Und die Version wieder loswerden (Invariante 4, verwerfbar): +-- drop index if exists item_embeddings_hnsw_; +-- delete from item_embeddings where coder_version = '...'; +-- -------------------------------------------------------------------------- +create or replace function p2_embedding_index(version text) +returns text language plpgsql as $$ +declare + idx text; +begin + -- Nicht-alphanumerisches im Versionsnamen wird zu _, damit der + -- Indexname ohne Anfuehrungszeichen auskommt. + idx := 'item_embeddings_hnsw_' || regexp_replace(version, '[^a-zA-Z0-9]+', '_', 'g'); + if to_regclass(idx) is not null then + return idx || ' (bestand bereits)'; + end if; + execute format( + 'create index %I on item_embeddings using hnsw (vektor vector_cosine_ops) where coder_version = %L', + idx, version); + return idx; +end; +$$; + +comment on function p2_embedding_index is + 'Legt den HNSW-Index fuer genau eine coder_version an. Idempotent'; + +insert into schema_migrationen (name) values ('p2-002-embeddings') +on conflict (name) do nothing; + +commit; diff --git a/migrations/p2-003-eigentuemer.sql b/migrations/p2-003-eigentuemer.sql new file mode 100644 index 0000000..280f68b --- /dev/null +++ b/migrations/p2-003-eigentuemer.sql @@ -0,0 +1,91 @@ +-- Migration p2-003: Schemaobjekte der Anwendungsrolle uebereignen +-- +-- psql "$ADMIN_DATABASE_URL" -v ON_ERROR_STOP=1 -f migrations/p2-003-eigentuemer.sql +-- +-- Die Datenbank wurzelwerk gehoert der Rolle wurzelwerk, ihre Tabellen aber +-- postgres: schema.sql wurde beim Ingest-Deploy als Superuser eingespielt. +-- Folge ist, dass die Anwendungsrolle an ihrem eigenen Schema nichts aendern +-- darf - jedes `alter table` scheitert mit "must be owner of table". +-- +-- Die uebrigen Dienste auf derselben Instanz halten es anders herum: gitea +-- und hedgedoc besitzen ihre Tabellen vollstaendig. Diese Migration stellt +-- denselben Zustand her. Der Radius bleibt auf die Datenbank wurzelwerk +-- beschraenkt; an gitea und hedgedoc wird nicht geruehrt. +-- +-- Danach laufen Migrationen ohne Superuser, also ueber DATABASE_URL statt +-- ueber den Socket im Container. Ausnahme bleibt `create extension` in +-- p2-002, sofern die Erweiterung nicht als trusted gefuehrt wird. +-- +-- Diese Migration ist die einzige, die zwingend als Superuser laeuft - sie +-- verschenkt Rechte, die sie selbst nicht mehr braeuchte. + +begin; + +do $$ +declare + ziel constant name := 'wurzelwerk'; + o record; + n integer := 0; +begin + if not exists (select 1 from pg_roles where rolname = ziel) then + raise exception 'Rolle % existiert nicht', ziel; + end if; + + -- Tabellen, Sichten und freistehende Sequenzen. + -- + -- Ausgenommen sind zwei Gruppen. Erstens Objekte einer Erweiterung + -- (pg_trgm, vector): die gehoeren dem Superuser und werden mit der + -- Erweiterung verwaltet, nicht von Hand. Zweitens Sequenzen, die an + -- einer Spalte haengen - alles aus bigserial. Die folgen dem Eigentuemer + -- ihrer Tabelle von selbst und lassen sich nicht einzeln uebereignen + -- ("is linked to table"). Deshalb erst die Tabellen, dann der Rest. + for o in + select c.oid::regclass::text as name, + case c.relkind when 'S' then 'sequence' + when 'v' then 'view' + when 'm' then 'materialized view' + else 'table' end as art + from pg_class c join pg_namespace n on n.oid = c.relnamespace + where n.nspname = 'public' + and c.relkind in ('r', 'p', 'S', 'v', 'm') + and pg_get_userbyid(c.relowner) <> ziel + and not exists (select 1 from pg_depend d + where d.classid = 'pg_class'::regclass + and d.objid = c.oid and d.deptype = 'e') + and not exists (select 1 from pg_depend d + where d.classid = 'pg_class'::regclass + and d.objid = c.oid and d.deptype = 'a' + and d.refclassid = 'pg_class'::regclass) + order by case c.relkind when 'r' then 0 when 'p' then 0 + when 'v' then 1 when 'm' then 1 else 2 end + loop + execute format('alter %s %s owner to %I', o.art, o.name, ziel); + n := n + 1; + end loop; + + -- Funktionen ebenso. Ohne sie duerfte die Rolle ihre eigenen Helfer + -- spaeter nicht mehr ersetzen. + for o in + select p.oid::regprocedure::text as name + from pg_proc p join pg_namespace n on n.oid = p.pronamespace + where n.nspname = 'public' + and pg_get_userbyid(p.proowner) <> ziel + and not exists (select 1 from pg_depend d + where d.classid = 'pg_proc'::regclass + and d.objid = p.oid and d.deptype = 'e') + loop + execute format('alter function %s owner to %I', o.name, ziel); + n := n + 1; + end loop; + + raise notice 'uebereignet: % Objekte an %', n, ziel; +end $$; + +-- Neue Objekte gehoeren ohnehin dem, der sie anlegt; das Recht dazu im +-- Schema public ist seit PostgreSQL 15 nicht mehr selbstverstaendlich. +grant create, usage on schema public to wurzelwerk; + +insert into schema_migrationen (name) values ('p2-003-eigentuemer') +on conflict (name) do nothing; + +commit; diff --git a/normalisierung.py b/normalisierung.py new file mode 100644 index 0000000..fe9a9bc --- /dev/null +++ b/normalisierung.py @@ -0,0 +1,105 @@ +"""Textnormalisierung fuer Phase 2. + +Gemeinsame Vorstufe der Module. Jede Aenderung hier aendert die +coder_version - siehe version.py. + +Grundsatz: die Normalisierung ist rein und ohne Zustand. Gleiche Eingabe, +gleiche Ausgabe, auf jeder Maschine und in jedem Prozess. +""" + +import html +import re +import unicodedata + +# Was hier steht, geht in den Versions-Hash ein. +REGELWERK_VERSION = "norm-3" + +_TAGS = re.compile(r"<[^>]{1,200}>") +_MEHRFACH_LEER = re.compile(r"\s+") + +# Anfuehrungszeichen und Striche vereinheitlichen. Verschiedene Haeuser setzen +# fuer dieselbe Agenturmeldung verschiedene Zeichen; ohne das scheitert schon +# die Trigrammbildung. +_ZEICHEN = str.maketrans({ + "„": '"', "“": '"', "”": '"', "»": '"', "«": '"', "‟": '"', + "‚": "'", "‘": "'", "’": "'", "‹": "'", "›": "'", "´": "'", "`": "'", + "–": "-", "—": "-", "‑": "-", "‒": "-", "−": "-", + "…": "...", " ": " ", " ": " ", " ": " ", "": " ", +}) + +# "+++ Newsblog +++" und Verwandtes: redaktioneller Aufmacher, der nichts +# ueber den Inhalt sagt. Dieselbe Meldung traegt ihn bei einem Haus und beim +# naechsten nicht. +_PLUSKASTEN = re.compile(r"\+{2,}[^+]{0,120}?\+{2,}") + +# Satzzeichen fallen weg: sie sind der haeufigste Unterschied zwischen zwei +# Fassungen derselben Meldung und tragen nichts zur Unterscheidung bei. +_KEIN_WORT = re.compile(r"[^\w\s]", re.UNICODE) + +# Ressortkuerzel, das Haeuser ihren Titeln voranstellen: "Raumfahrt: Deutsche +# Rakete ...", "Notfaelle: Fuenf Tote ...", "Fussball: FIFA-Praesident ...". +# Dieselbe Agenturzeile bekommt bei jedem Haus ein anderes - oder gar keins. +# +# Bewusst eng gefasst: hoechstens vier Woerter vor dem Doppelpunkt, und was +# danach kommt, muss fuer sich stehen koennen. Sonst frisst das Muster +# Titel, bei denen der Teil vor dem Doppelpunkt die eigentliche Aussage +# traegt ("Habeck: Wir haben uns geirrt"). +_RESSORT = re.compile(r"^\s*(?P[^:]{2,34}):\s+(?P.+)$", re.DOTALL) +_RESSORT_MAX_WOERTER = 4 +_REST_MIN_WOERTER = 5 + + +def _ressort_abtrennen(text): + treffer = _RESSORT.match(text) + if not treffer: + return text + kopf, rest = treffer.group("kopf"), treffer.group("rest") + if len(kopf.split()) > _RESSORT_MAX_WOERTER: + return text + if len(rest.split()) < _REST_MIN_WOERTER: + return text + return rest + + +def normalisiere(*teile, ressort_abtrennen=False): + """Vergleichstext aus einem oder mehreren Feldern. + + Reihenfolge ist bedeutsam: Tags vor Entities, NFKC vor der + Zeichentabelle, Kleinschreibung vor den Mustern, und das Ressortkuerzel + vor dem Entfernen der Satzzeichen - danach gibt es keinen Doppelpunkt + mehr, an dem es zu erkennen waere. + + ressort_abtrennen gilt nur fuer Titel. Im Teaser steht der Doppelpunkt + fuer gewoehnlich mitten im Satz. + """ + text = " ".join(t for t in teile if t) + if not text: + return "" + text = _TAGS.sub(" ", text) + text = html.unescape(text) + text = unicodedata.normalize("NFKC", text) + text = text.translate(_ZEICHEN) + text = text.casefold() + text = _PLUSKASTEN.sub(" ", text) + if ressort_abtrennen: + text = _ressort_abtrennen(text) + text = _KEIN_WORT.sub(" ", text) + return _MEHRFACH_LEER.sub(" ", text).strip() + + +def woerter(text): + return text.split() + + +def trigramme(text): + """Wort-Trigramme als Menge. + + Menge, nicht Liste: der Jaccard-Wert braucht sie so, und Wiederholungen + innerhalb einer kurzen Meldung sind kein Signal. + """ + w = woerter(text) + if len(w) < 3: + # Zu kurz fuer Trigramme. Das Wort selbst ist besser als nichts, + # aber der Aufrufer prueft ohnehin auf Mindestlaenge. + return {(x,) for x in w} + return {tuple(w[i:i + 3]) for i in range(len(w) - 2)} diff --git a/test_dubletten.py b/test_dubletten.py new file mode 100644 index 0000000..ae14ccd --- /dev/null +++ b/test_dubletten.py @@ -0,0 +1,349 @@ +#!/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("Habeck & 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() diff --git a/version.py b/version.py new file mode 100644 index 0000000..5b02b19 --- /dev/null +++ b/version.py @@ -0,0 +1,69 @@ +"""coder_version: was hinter dem Hash steht. + +Format aus der Spezifikation: p2--. + +Der Hash bildet alle Bestandteile, die das Ergebnis beeinflussen. Aendert +sich einer, aendert sich die Version, und alte Codings bleiben unangetastet +liegen. Das Datum ist das der letzten Aenderung an den Komponenten, nicht das +des Laufs - sonst waere die Version taeglich neu. + +Anmerkung zur Spezifikation: sie sieht *eine* Version ueber alle fuenf Module +vor. Ein neues Sentimentlexikon entwertet damit auch die Embeddings, obwohl +es sie nicht beruehrt. Das ist teuer, aber es ist die festgelegte Semantik; +wer das aendern will, aendert die Spezifikation, nicht diese Datei. +""" + +import hashlib +import json + +import normalisierung + +# Stand der Komponenten. Module, die es noch nicht gibt, stehen auf null - +# sie tragen dann nichts zum Hash bei, ausser dass sie erwaehnt sind. +KOMPONENTEN = { + "stand": "2026-09-07", + "normalisierung": normalisierung.REGELWERK_VERSION, + # Wird vom jeweiligen Modul beim Lauf eingesetzt (siehe komponenten()). + # Die Einstellungen stehen dort, wo sie wirken, nicht hier - sonst + # gerieten die beiden auseinander, und der Hash behauptete eine + # Gleichheit, die es nicht gibt. + "dubletten": None, + "akteure": None, + "geo": None, + "tonalitaet": None, + "embedding": None, +} + + +def komponenten(**module): + """Kopie der Komponenten mit eingesetzten Modulangaben. + + Ein Modul kennt seine Einstellungen selbst und reicht sie hier herein. + Aendert es eine Schwelle, aendert sich der Hash - genau so ist es + gemeint. + """ + k = dict(KOMPONENTEN) + k.update(module) + return k + + +def coder_version(komponenten=None): + k = KOMPONENTEN if komponenten is None else komponenten + # sort_keys: die Reihenfolge im Quelltext darf den Hash nicht bewegen. + roh = json.dumps(k, sort_keys=True, ensure_ascii=False, separators=(",", ":")) + hash8 = hashlib.blake2b(roh.encode("utf-8"), digest_size=4).hexdigest() + return f"p2-{k['stand']}-{hash8}" + + +def eintragen(conn, komponenten=None): + """Version in coder_versionen festhalten. Ohne das sagt der Hash nichts.""" + from psycopg.types.json import Jsonb + + k = KOMPONENTEN if komponenten is None else komponenten + v = coder_version(k) + conn.execute( + "insert into coder_versionen (version, komponenten) values (%s, %s)" + " on conflict (version) do nothing", + (v, Jsonb(k)), + ) + return v