swp-02-aufbereitung/akteure.py

"""Modul 4: Akteursaufloesung (ebene = 'akteure').

Drei Schritte, streng getrennt:

    NER            liefert Kandidatenspannen - nur wo, nicht was
    Lexikon        entscheidet, was eine Spanne ist
    Amtstabelle    loest Aemter gegen das Datum des Items auf

Die Trennung ist gemessen begruendet: de_core_news_lg haelt "Merz" fuer
einen Ort, "Landtagswahl" fuer einen Ort und "Ukraine-Krieg" fuer eine
Person (siehe NOTES.md). Der Typ einer Nennung kommt deshalb aus dem
Lexikon, nie aus dem Modell. Der NER-Typ wird trotzdem mitgeschrieben -
die Geokodierung braucht ihn als Vorfilter, und er dokumentiert, woher
eine Spanne stammt.

Nicht aufloesbare Nennungen bleiben mit id null stehen. Sie sind kein
Ausschuss, sondern das Rohmaterial der Lexikonpflege; die Sicht
akteure_unaufgeloest zaehlt sie aus.
"""

import argparse
import collections
import datetime
import hashlib
import re
import sys
import time
import unicodedata

import db
import normalisierung
import version

EBENE = "akteure"

NER_MODELL = "de_core_news_lg"
STAPEL = 64

# Obergrenze fuer eine Oberflaechenform in der Nutzlast.
FORM_MAX = 100

# Nur diese NER-Typen werden als Kandidaten betrachtet. MISC bleibt drin,
# weil beide Modelle Parteien regelmaessig dort einsortieren - AfD steht
# 226-mal als MISC im sm-Lauf.
TYPEN = ("PER", "ORG", "LOC", "MISC")

# Fuzzy-Match: erlaubte Editierdistanz nach Laenge der Oberflaechenform.
# Kurze Formen vertragen keinen Fehler - aus "Wolf" wuerde sonst "Wolff",
# "Rolf" und "Golf".
def fuzzy_grenze(laenge):
    if laenge < 6:
        return 0
    if laenge < 12:
        return 1
    return 2


# Woerter, die als Nachname im Lexikon stehen, in Schlagzeilen aber fast
# immer etwas anderes meinen. Handpflege, mit Begruendung je Eintrag.
NACHNAMEN_SPERRE = {
    "wolf",     # Katja Wolf, aber auch das Tier - in dieser Saison haeufig
    "merkel",   # nicht im Lexikon, aber als Nachname mehrdeutig belegt
    "bund",
    "land",
    "recht",
    "kraft",
    "post",
    "stark",
    "reich",
    "gross",
    "klein",
    "lang",
    "kurz",
    "jung",
    "alt",
    "neu",
    "weiss",
    "schwarz",
    "braun",
    "gruen",
    "hacker",   # Thomas Hacker, MdB - in Schlagzeilen fast nie gemeint
    "un",       # Nachname von Kim Jong-un, in Schlagzeilen die UN
    "partei",
    "post",
    "bahn",
    "welt",
    "zeit",
    "spiegel",
    "focus",
    # Staedtenamen, die zugleich Nachnamen von Abgeordneten sind. Erst mit
    # den Landtagen kamen sie ins Lexikon, und sofort wurde aus jedem
    # "Hamburg" eine niedersaechsische Kultusministerin - 22-mal im Korpus
    # vom 08.09.2026.
    "hamburg",   # Julia Willie Hamburg, MdL Niedersachsen
    "rostock",   # Clemens Rostock, MdL Brandenburg
    "oldenburg", # Simone Oldenburg, MdL Mecklenburg-Vorpommern
    "ki",        # Ki-hong Nam - in deutschen Schlagzeilen die KI
    "nothing",   # Volker Nothing, MdL - sonst das englische Wort
    "andrew",    # Stuart Andrew, MP - sonst ein Vorname
    "champagne", # Francois-Philippe Champagne - sonst das Getraenk
}

# Aliase, die Wikidata fuehrt, die in Schlagzeilen aber regelmaessig etwas
# anderes meinen. Wirkt auf den Alias-Match, nicht nur auf Nachnamen.
FORMEN_SPERRE = {
    "europa",   # Wikidata fuehrt es als Alias der EU. Meist ist der
                # Kontinent gemeint, und der ist Sache der Geokodierung.
    "europas",
    "union",    # CDU/CSU oder Europaeische Union - ohne Kontext nicht zu
                # entscheiden, und beide Lesarten sind haeufig.
}

# Rangordnung der Spiegelabfragen fuer den Gleichstandsfall. Sie ist eine
# Aussage darueber, wen eine deutsche Schlagzeile mit blossem Nachnamen
# meint, nicht darueber, wer wichtiger ist. Der Anlass ist messbar: mit den
# Landtagen kamen Janine und Heiner Merz ins Lexikon, und "Merz" war
# schlagartig mehrdeutig - 125 Nennungen des Bundeskanzlers fielen aus.
#
# Ein Rang entscheidet nur, wenn genau ein Kandidat den niedrigsten traegt.
# Bei Gleichstand innerhalb eines Rangs bleibt es beim Muenzwurfverbot.
RANG = {
    "hand": 0,
    "staatsfuehrung": 1,
    "regierung": 1,
    "sondergesandte": 1,
    "bundestag": 2,
    "landtage": 3,
    # Die Amtstabelle reicht bis 1794 zurueck und besteht ueberwiegend aus
    # Verstorbenen. Sie darf einen Gleichstand nicht gewinnen: sonst wird
    # aus "Teufel" ein Ministerpraesident a.D. und aus "Otto" ein Bremer
    # Buergermeister des 18. Jahrhunderts. Wer heute im Amt ist, steht
    # ohnehin in einer der Personenabfragen.
    "aemter": 5,
}
RANG_SONST = 4


def rang(quelle):
    """'wikidata:landtage@2026-09-08' -> 3."""
    name = (quelle or "").split(":")[-1].split("@")[0]
    return RANG.get(name, RANG_SONST)


EINSTELLUNGEN_FEST = {
    "ner_modell": NER_MODELL,
    "basis": "titel+teaser",
    "text": "bereinigt",
    "ner_typen": list(TYPEN),
    "typ_aus": "lexikon",
    "fuzzy": "editierdistanz_nach_laenge",
    "nachnamen": "eindeutig_kontext_rang",
    "rang": "|".join(f"{k}={v}" for k, v in sorted(RANG.items())),
    # Die Sperrliste gehoert in die Version, nicht nur in den Quelltext.
    # Sie aendert das Ergebnis - ein Eintrag mehr, und aus 22 Nennungen
    # der Hansestadt wird keine Ministerin mehr. Ohne sie im Fingerabdruck
    # traegt dieselbe coder_version zwei verschiedene Auszaehlungen, und
    # genau das soll die Version verhindern.
    "nachnamen_sperre": hashlib.blake2b(
        "|".join(sorted(NACHNAMEN_SPERRE)).encode(), digest_size=8
    ).hexdigest(),
}

_UMLAUTE = str.maketrans({"ä": "ae", "ö": "oe", "ü": "ue", "ß": "ss"})
_UNRAT = re.compile(r"[^a-z0-9 ]+")
_MEHRFACH = re.compile(r"\s+")

# Titel und Anreden, die vor einem Namen stehen und nicht zu ihm gehoeren.
_ANREDE = re.compile(
    r"^(?:der|die|das|dem|den|des|ex|alt)?[- ]?"
    r"(?:bundes|landes|us|eu|nato|un|uno)?[- ]?"
    r"(?:kanzler(?:in)?|praesident(?:in)?|minister(?:in)?|ministerpraesident(?:in)?|"
    r"aussenminister(?:in)?|innenminister(?:in)?|chef(?:in)?|sprecher(?:in)?|"
    r"vorsitzende[rn]?|abgeordnete[rn]?|spitzenkandidat(?:in)?|kandidat(?:in)?|"
    r"parteichef(?:in)?|fraktionschef(?:in)?|gesandte[rn]?|premier(?:minister)?)"
    r"[- ]",
)


def vergleichsform(text):
    """Form, in der Oberflaeche und Lexikon miteinander verglichen werden."""
    text = unicodedata.normalize("NFKC", text).casefold().translate(_UMLAUTE)
    text = unicodedata.normalize("NFKD", text)
    text = "".join(z for z in text if not unicodedata.combining(z))
    return _MEHRFACH.sub(" ", _UNRAT.sub(" ", text)).strip()


def ohne_anrede(form):
    """'us-praesident trump' -> 'trump'. Mehrfach, fuer Ketten."""
    vorher = None
    while vorher != form:
        vorher = form
        form = _ANREDE.sub("", form).strip()
    return form or vorher


def distanz(a, b, grenze):
    """Levenshtein, abgebrochen sobald die Grenze ueberschritten ist."""
    if abs(len(a) - len(b)) > grenze:
        return grenze + 1
    vorige = list(range(len(b) + 1))
    for i, za in enumerate(a, 1):
        zeile = [i]
        for j, zb in enumerate(b, 1):
            zeile.append(min(vorige[j] + 1, zeile[j - 1] + 1,
                             vorige[j - 1] + (za != zb)))
        if min(zeile) > grenze:
            return grenze + 1
        vorige = zeile
    return vorige[-1]


class Lexikon:
    """Nachschlagewerk im Speicher. Wird einmal je Lauf gebaut."""

    def __init__(self, akteure, amtszeiten):
        self.akteure = {a["id"]: a for a in akteure}
        self.formen = collections.defaultdict(set)
        self.nachnamen = collections.defaultdict(set)
        self.amtszeiten = collections.defaultdict(list)

        for a in akteure:
            for form in [a["name"], *a["aliase"]]:
                v = vergleichsform(form)
                if v:
                    self.formen[v].add(a["id"])
            if a["typ"] == "person":
                teile = vergleichsform(a["name"]).split()
                if len(teile) > 1 and teile[-1] not in NACHNAMEN_SPERRE:
                    self.nachnamen[teile[-1]].add(a["id"])

        for z in amtszeiten:
            self.amtszeiten[z["amt_id"]].append(z)
        for eintraege in self.amtszeiten.values():
            eintraege.sort(key=lambda z: z["von"])

        # Fuzzy-Suche nur innerhalb eines Laengenfensters, sonst waere sie
        # ein Kreuzprodukt aus 14000 Spannen und 12000 Formen.
        self.nach_laenge = collections.defaultdict(list)
        for v in self.formen:
            self.nach_laenge[len(v)].append(v)

    def inhaber(self, amt_id, datum):
        """Wer hatte dieses Amt an diesem Tag."""
        treffer = []
        for z in self.amtszeiten.get(amt_id, ()):
            # Halboffen, wie in der Sicht amtsinhaber_ueberschneidungen:
            # am Tag der Uebergabe zaehlt der Nachfolger.
            if z["von"] <= datum and (z["bis"] is None or datum < z["bis"]):
                treffer.append(z["person_id"])
        return treffer

    def schlage_nach(self, form):
        """(ids, methode) fuer eine Vergleichsform."""
        if form in FORMEN_SPERRE:
            return set(), None
        if form in self.formen:
            return self.formen[form], "alias"
        kurz = ohne_anrede(form)
        if kurz in FORMEN_SPERRE:
            return set(), None
        if kurz != form and kurz in self.formen:
            return self.formen[kurz], "alias_ohne_anrede"
        if kurz in self.nachnamen:
            return self.nachnamen[kurz], "nachname"

        grenze = fuzzy_grenze(len(kurz))
        if grenze:
            beste, bester_abstand = set(), grenze + 1
            for laenge in range(len(kurz) - grenze, len(kurz) + grenze + 1):
                for kandidat in self.nach_laenge.get(laenge, ()):
                    if kandidat[0] != kurz[0]:
                        continue
                    d = distanz(kurz, kandidat, grenze)
                    if d > grenze:
                        # distanz() liefert bei Abbruch grenze+1. Ohne diese
                        # Zeile zaehlt jeder Abbruch als Gleichstand mit
                        # jedem anderen Abbruch, und die Kandidatenliste
                        # fuellt sich mit allem, was zufaellig gleich lang
                        # ist und mit demselben Buchstaben beginnt.
                        continue
                    if d < bester_abstand:
                        beste, bester_abstand = set(self.formen[kandidat]), d
                    elif d == bester_abstand:
                        beste |= self.formen[kandidat]
            if beste:
                return beste, "fuzzy"
        return set(), None


def lade_lexikon(conn):
    with conn.cursor() as k:
        k.execute("select id, typ, name, aliase, land, rolle, quelle from akteure")
        akteure = [
            {"id": r[0], "typ": r[1], "name": r[2], "aliase": r[3],
             "land": r[4], "rolle": r[5], "rang": rang(r[6])}
            for r in k.fetchall()
        ]
        k.execute("select amt_id, person_id, von, bis from amtsinhaber")
        amtszeiten = [
            {"amt_id": r[0], "person_id": r[1], "von": r[2], "bis": r[3]}
            for r in k.fetchall()
        ]
    return Lexikon(akteure, amtszeiten)


def waehle(lexikon, ids, kontext):
    """Aus mehreren Kandidaten einen machen - oder keinen.

    Regelbasiert und in dieser Reihenfolge, wie die Spezifikation es
    verlangt:

    1. Ist genau einer gemeint, ist die Sache erledigt.
    2. Wird die Partei eines Kandidaten im selben Item genannt, gewinnt er.
    3. Traegt genau einer den niedrigsten Rang, gewinnt er - siehe RANG.
    4. Sonst bleibt die Nennung unaufgeloest. Ein Muenzwurf waere hier der
       schlimmere Fehler, weil er in der Zeitreihe nicht auffaellt.
    """
    if len(ids) == 1:
        return next(iter(ids)), "eindeutig"
    mit_partei = [i for i in ids
                  if (lexikon.akteure[i].get("notiz") or "") and
                  vergleichsform(lexikon.akteure[i]["notiz"]) in kontext]
    if len(mit_partei) == 1:
        return mit_partei[0], "kontext_partei"
    beste = min(lexikon.akteure[i].get("rang", RANG_SONST) for i in ids)
    vorn = [i for i in ids
            if lexikon.akteure[i].get("rang", RANG_SONST) == beste]
    if len(vorn) == 1:
        return vorn[0], "rang"
    return None, "mehrdeutig"


def erkennungen(nlp, titel, teaser):
    """Kandidatenspannen je Feld, mit Offsets im bereinigten Text."""
    felder = [("titel", normalisierung.bereinige(titel)),
              ("teaser", normalisierung.bereinige(teaser))]
    ergebnis = []
    for feld, text in felder:
        if not text:
            continue
        for ent in nlp(text).ents:
            if ent.label_ in TYPEN:
                ergebnis.append({"feld": feld,
                                 # Die Nutzlast darf keinen Volltext tragen
                                 # (Constraint p2_nutzlast_ohne_text). Eine
                                 # Nennung ist nie so lang; start/ende machen
                                 # den Urtext ohnehin auffindbar.
                                 "form": ent.text[:FORM_MAX],
                                 "start": ent.start_char, "ende": ent.end_char,
                                 "ner_typ": ent.label_})
    return ergebnis


def aufloesen(lexikon, spannen, datum):
    """Spannen gegen das Lexikon aufloesen. Reine Funktion, testbar."""
    kontext = " ".join(vergleichsform(s["form"]) for s in spannen)
    erkannt = []
    for s in spannen:
        form = vergleichsform(s["form"])
        ids, methode = lexikon.schlage_nach(form)
        eintrag = dict(s)
        if not ids:
            eintrag.update({"id": None, "typ": None, "rolle": None,
                            "methode": "ner_unaufgeloest", "konfidenz": 0.4})
            erkannt.append(eintrag)
            continue

        gewaehlt, grund = waehle(lexikon, ids, kontext)
        if gewaehlt is None:
            eintrag.update({"id": None, "typ": None, "rolle": None,
                            "methode": f"{methode}_{grund}", "konfidenz": 0.3,
                            "kandidaten": sorted(ids)[:5]})
            erkannt.append(eintrag)
            continue

        a = lexikon.akteure[gewaehlt]
        konfidenz = {"alias": 0.95, "alias_ohne_anrede": 0.9,
                     "nachname": 0.8, "fuzzy": 0.6}[methode]
        # Abschlaege fuer die Gleichstandsregeln. Der Rang ist die
        # schwaechere der beiden: die Partei steht im Text, der Rang ist
        # nur eine Erwartung darueber, wer gemeint sein duerfte.
        if grund == "kontext_partei":
            konfidenz -= 0.1
        elif grund == "rang":
            konfidenz -= 0.15
        eintrag.update({"id": gewaehlt, "typ": a["typ"], "rolle": a["rolle"],
                        "methode": methode, "konfidenz": round(konfidenz, 2)})
        if grund != "eindeutig":
            eintrag["grund"] = grund

        # Aemter gegen das Datum des Items aufloesen. Das Amt bleibt stehen,
        # die Person kommt daneben - beides ist eine Aussage ueber den Text.
        if a["typ"] == "amt" and datum:
            inhaber = lexikon.inhaber(gewaehlt, datum)
            if len(inhaber) == 1:
                eintrag["inhaber"] = inhaber[0]
                eintrag["methode"] = "amt"
            elif inhaber:
                eintrag["inhaber_mehrdeutig"] = sorted(inhaber)
        erkannt.append(eintrag)
    return erkannt


def nutzlast(erkannt, lexikon_stand):
    aufgeloest = [e for e in erkannt if e["id"]]
    return {
        "erkannt": erkannt,
        "ner_modell": NER_MODELL,
        "lexikon_version": lexikon_stand,
        "nennungen": len(erkannt),
        "aufgeloest": len(aufgeloest),
        "typen": dict(collections.Counter(e["typ"] for e in aufgeloest)),
    }


_SPALTEN = "select id, titel, teaser, pubdate from raw_items"


def lade(conn, slot=None):
    with conn.cursor() as k:
        if slot:
            k.execute(_SPALTEN + " where slot = %s order by id", (slot,))
        else:
            k.execute(_SPALTEN + " order by id")
        return k.fetchall()


def lexikon_stand(conn):
    """Fingerabdruck des Lexikons.

    Geht in die coder_version ein, und zwar inhaltlich: ein einziger neuer
    Alias aendert den Hash und damit die Version. Eine blosse Zeilenzahl
    taete das nicht - dann waere eine Korrektur am Lexikon von aussen
    unsichtbar, und alte Codings behaupteten eine Grundlage, die es nicht
    mehr gibt.
    """
    with conn.cursor() as k:
        k.execute("""select id, typ, name, aliase, land, rolle
                     from akteure order by id""")
        akteure_zeilen = k.fetchall()
        k.execute("""select amt_id, person_id, von, bis
                     from amtsinhaber order by amt_id, person_id, von""")
        amtszeilen = k.fetchall()
    roh = repr(akteure_zeilen) + repr(amtszeilen)
    kurz = hashlib.blake2b(roh.encode("utf-8"), digest_size=6).hexdigest()
    return f"{len(akteure_zeilen)}+{len(amtszeilen)}/{kurz}"


def einstellungen(stand):
    return dict(EINSTELLUNGEN_FEST, lexikon=stand)


def verarbeite(conn, slot=None, trocken=False):
    import spacy

    beginn = time.time()
    lexikon = lade_lexikon(conn)
    stand = lexikon_stand(conn)
    v = version.coder_version(version.komponenten(akteure=einstellungen(stand)))
    if not trocken:
        version.eintragen(conn, version.komponenten(akteure=einstellungen(stand)))

    nlp = spacy.load(NER_MODELL, exclude=["lemmatizer"])
    zeilen = lade(conn, slot)
    print(f"{EBENE}: {len(zeilen)} Items, Version {v}, Lexikon {stand}")

    saetze, zahlen = [], collections.Counter()
    for item_id, titel, teaser, pubdate in zeilen:
        datum = pubdate.date() if isinstance(pubdate, datetime.datetime) else pubdate
        spannen = erkennungen(nlp, titel, teaser)
        erkannt = aufloesen(lexikon, spannen, datum)
        zahlen["nennungen"] += len(erkannt)
        zahlen["aufgeloest"] += sum(1 for e in erkannt if e["id"])
        saetze.append((item_id, v, nutzlast(erkannt, stand)))

    if trocken:
        print(f"  trocken, nichts geschrieben: {dict(zahlen)}")
        return zahlen

    from psycopg.types.json import Jsonb

    with conn.cursor() as k:
        k.executemany(
            """insert into codings (raw_item_id, coder_version, ebene, nutzlast)
               values (%s, %s, %s, %s)
               on conflict (raw_item_id, coder_version, ebene) do update
                   set nutzlast = excluded.nutzlast, kodiert_am = now()""",
            [(i, ver, EBENE, Jsonb(n)) for i, ver, n in saetze],
        )
        k.execute(
            """insert into p2_laeufe (slot, modul, coder_version, dauer_ms,
                                      items_gesehen, items_kodiert, items_fehler)
               values (%s, %s, %s, %s, %s, %s, 0)""",
            (slot, EBENE, v, int((time.time() - beginn) * 1000),
             len(zeilen), len(saetze)),
        )
    conn.commit()
    quote = zahlen["aufgeloest"] / zahlen["nennungen"] if zahlen["nennungen"] else 0
    print(f"  {len(saetze)} Codings, {zahlen['nennungen']} Nennungen, "
          f"{zahlen['aufgeloest']} aufgeloest ({quote:.1%}), "
          f"{time.time() - beginn:.1f}s")
    return zahlen


def main(argumente=None):
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    g = p.add_mutually_exclusive_group(required=True)
    g.add_argument("--backfill", action="store_true")
    g.add_argument("--slot")
    p.add_argument("--trocken", action="store_true")
    a = p.parse_args(argumente)

    with db.verbindung() as conn:
        verarbeite(conn, slot=a.slot, trocken=a.trocken)
    return 0


if __name__ == "__main__":
    sys.exit(main())