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",
}

# 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.
}

EINSTELLUNGEN_FEST = {
    "ner_modell": NER_MODELL,
    "basis": "titel+teaser",
    "text": "bereinigt",
    "ner_typen": list(TYPEN),
    "typ_aus": "lexikon",
    "fuzzy": "editierdistanz_nach_laenge",
    "nachnamen": "eindeutig_oder_kontext",
}

_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 from akteure")
        akteure = [
            {"id": r[0], "typ": r[1], "name": r[2], "aliase": r[3],
             "land": r[4], "rolle": r[5]}
            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. 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"
    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]
        if grund == "kontext_partei":
            konfidenz -= 0.1
        eintrag.update({"id": gewaehlt, "typ": a["typ"], "rolle": a["rolle"],
                        "methode": methode, "konfidenz": round(konfidenz, 2)})

        # 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())