swp-02-aufbereitung/lexikon_laden.py

"""Spiegel in die Tabellen akteure und amtsinhaber laden.

Liest ausschliesslich lexikon/*.json. Kein Netzzugriff - wer den Spiegel
erneuern will, nimmt lexikon.py --spiegeln.

Hier und nur hier geschieht die Auslegung: aus Wikidata-Labels werden IDs,
aus Abfragenamen werden Rollen, aus Zeitangaben werden Amtszeiten. Der
Spiegel bleibt unberuehrt, damit sich jede dieser Ableitungen spaeter
aendern laesst, ohne Wikidata erneut zu behelligen.
"""

import json
import re
import unicodedata

import lexikon

VERZEICHNIS = lexikon.VERZEICHNIS

# Welche Abfrage welchen Typ und welche Rolle traegt. Die Rolle ist eine
# grobe Zuordnung nach CAMEO-Manier; sie beschreibt, in welcher Eigenschaft
# jemand ueblicherweise in Nachrichten auftaucht, nicht seine Biographie.
HERKUNFT = {
    "bundestag": ("person", "LEG"),
    "landtage": ("person", "LEG"),
    "staatsfuehrung": ("person", "GOV"),
    "regierung": ("person", "GOV"),
    "sondergesandte": ("person", "GOV"),
    "unternehmen": ("organisation", "BUS"),
    "parteien": ("partei", None),
    "staaten": ("staat", None),
    "organisationen": ("organisation", "INT"),
}

PRAEFIX = {
    "person": "per",
    "partei": "par",
    "staat": "sta",
    "organisation": "org",
    "amt": "amt",
}

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


def schluessel(text):
    """Namensbestandteil zu einem ID-tauglichen Wort."""
    text = text.casefold().translate(_UMLAUTE)
    text = unicodedata.normalize("NFKD", text)
    text = "".join(z for z in text if not unicodedata.combining(z))
    return _UNRAT.sub("_", text).strip("_")


def personen_id(name):
    """'Robert Habeck' -> 'per:habeck_robert', wie in der Spezifikation."""
    teile = name.split()
    if len(teile) < 2:
        return f"per:{schluessel(name)}"
    nachname, vornamen = teile[-1], " ".join(teile[:-1])
    return f"per:{schluessel(nachname)}_{schluessel(vornamen)}"


def qid(uri):
    return uri.rsplit("/", 1)[-1]


def aliase(zeile, name):
    """Aliase aus dem Spiegel, ohne den kanonischen Namen selbst."""
    roh = zeile.get("aliase") or ""
    menge = {a.strip() for a in roh.split("|") if a.strip()}
    menge.discard(name)
    return sorted(menge)


_ISO = re.compile(r"^-?\d{4}-\d{2}-\d{2}")


def _datum(wert):
    """Nur echte Datumsangaben.

    Wikidata kennt "unbekannter Wert" und liefert dafuer einen anonymen
    Knoten - im JSON eine URI. Als Datum ist der wertlos, und als Text
    faengt er sich sonst bis in die Tabelle durch.
    """
    if not wert or not _ISO.match(wert):
        return None
    return wert[:10]


def _stand(manifest, name):
    return (manifest.get(name) or {}).get("stand", "unbekannt")


def _handpflege(akteure):
    """lexikon/hand.json einarbeiten.

    Zweierlei: eigene Eintraege fuer Akteure, die durch jede Spiegelabfrage
    fallen (die NATO ist bei Wikidata ein Militaerbuendnis, kein
    internationale Organisation), und zusaetzliche Aliase fuer Abkuerzungen,
    die in deutschen Schlagzeilen selbstverstaendlich sind, in den deutschen
    altLabels aber fehlen - "USA", "EU", "UN".

    Ergaenzt, ueberschreibt nicht: was der Spiegel liefert, bleibt stehen.
    """
    datei = VERZEICHNIS / "hand.json"
    if not datei.exists():
        return
    hand = json.loads(datei.read_text())
    nach_qid = {a["wikidata_qid"]: a["id"] for a in akteure.values()
                if a.get("wikidata_qid")}
    for eintrag in hand.get("akteure", []):
        eintrag = dict(eintrag)
        eintrag.setdefault("aliase", [])
        eintrag["quelle"] = "hand"
        # Dasselbe Wikidata-Objekt kann im Spiegel unter anderem Namen
        # stecken: die NATO steht dort als
        # "Nordatlantikvertragsorganisation" - ohne den Alias "NATO".
        # Dann gehoeren nur die Aliase dazu, kein zweiter Eintrag.
        vorhanden = nach_qid.get(eintrag.get("wikidata_qid"))
        if vorhanden:
            ziel = akteure[vorhanden]
            ziel["aliase"] = sorted(set(ziel["aliase"]) | set(eintrag["aliase"])
                                    | {eintrag["name"]} - {ziel["name"]})
            continue
        akteure.setdefault(eintrag["id"], eintrag)
    for kennung, zusatz in hand.get("aliase", {}).items():
        if kennung in akteure:
            akteure[kennung]["aliase"] = sorted(
                set(akteure[kennung]["aliase"]) | set(zusatz))
        else:
            print(f"  hand.json: {kennung} gibt es nicht, Aliase ignoriert")


def zeilen_aus_spiegel(nur=None):
    """(akteure, amtsinhaber) aus den Spiegeldateien ableiten."""
    manifest = json.loads((VERZEICHNIS / "MANIFEST.json").read_text())
    akteure, gesehen = {}, {}
    # QID -> ID. Ein Wikidata-Objekt darf nur einen Akteur ergeben, sonst
    # zerfaellt es in zwei Zeitreihen. Australien etwa kommt sowohl in der
    # Staaten- als auch in der Organisationsabfrage vor.
    nach_qid, doppelt = {}, []

    for name, (typ, rolle) in HERKUNFT.items():
        if nur and name not in nur:
            continue
        datei = VERZEICHNIS / f"{name}.json"
        if not datei.exists():
            continue
        quelle = f"wikidata:{name}@{_stand(manifest, name)}"
        for z in json.loads(datei.read_text()):
            bezeichnung = z["name"]
            if typ == "person":
                kennung = personen_id(bezeichnung)
            else:
                kennung = f"{PRAEFIX[typ]}:{schluessel(bezeichnung)}"
            q = qid(z["p"])

            # Schon unter anderem Namen gesehen: Aliase dazu, sonst nichts.
            # Die erste Abfrage gewinnt, und die Reihenfolge von HERKUNFT
            # ist deshalb eine Aussage - Staat vor Organisation.
            if q in nach_qid:
                vorhanden = akteure[nach_qid[q]]
                vorhanden["aliase"] = sorted(
                    set(vorhanden["aliase"]) | set(aliase(z, bezeichnung))
                    | {bezeichnung} - {vorhanden["name"]})
                doppelt.append((q, nach_qid[q], name))
                continue

            # Namensgleichheit ist bei Abgeordneten keine Seltenheit. Die
            # zweite Person bekommt das QID angehaengt, statt die erste zu
            # ueberschreiben - stumm zusammenfallen waere der schlimmere
            # Fehler, weil er als Zeitreihe unauffaellig bleibt.
            if kennung in gesehen and gesehen[kennung] != q:
                kennung = f"{kennung}_{q.lower()}"
            gesehen[kennung] = q
            nach_qid[q] = kennung

            eintrag = akteure.setdefault(
                kennung,
                {
                    "id": kennung,
                    "typ": typ,
                    "name": bezeichnung,
                    "aliase": [],
                    "land": z.get("land"),
                    "rolle": rolle,
                    "wikidata_qid": q,
                    "quelle": quelle,
                    "notiz": z.get("partei"),
                },
            )
            eintrag["aliase"] = sorted(set(eintrag["aliase"]) | set(aliase(z, bezeichnung)))

    _handpflege(akteure)

    if doppelt:
        print(f"  {len(doppelt)} Objekte in mehreren Abfragen, erste gewinnt "
              f"(z.B. {doppelt[0][0]} -> {doppelt[0][1]} aus {doppelt[0][2]})")

    inhaber, verworfen = [], []
    if (not nur or "aemter" in nur) and (VERZEICHNIS / "aemter.json").exists():
        quelle = f"wikidata:aemter@{_stand(manifest, 'aemter')}"
        for amt_qid, (amt_id, amt_name, amt_rolle) in lexikon.AEMTER.items():
            akteure.setdefault(
                amt_id,
                {
                    "id": amt_id,
                    "typ": "amt",
                    "name": amt_name,
                    "aliase": [],
                    "land": "DEU",
                    "rolle": amt_rolle,
                    "wikidata_qid": amt_qid,
                    "quelle": quelle,
                    "notiz": None,
                },
            )
        for z in json.loads((VERZEICHNIS / "aemter.json").read_text()):
            amt_id = lexikon.AEMTER[z["amt_qid"]][0]
            person = nach_qid.get(qid(z["p"])) or personen_id(z["name"])
            # Amtsinhaber, die in keiner Personenabfrage vorkamen - etwa
            # Bundespraesidenten, die nie im Bundestag sassen.
            nach_qid.setdefault(qid(z["p"]), person)
            akteure.setdefault(
                person,
                {
                    "id": person,
                    "typ": "person",
                    "name": z["name"],
                    "aliase": [],
                    "land": None,
                    "rolle": "GOV",
                    "wikidata_qid": qid(z["p"]),
                    "quelle": quelle,
                    "notiz": None,
                },
            )
            von, bis = _datum(z.get("von")), _datum(z.get("bis"))
            if not von:
                # Ohne Anfang ist die Zeile fuer eine Amtsaufloesung wertlos.
                continue
            if bis and bis < von:
                # Kommt vor, wenn jemand dasselbe Amt zweimal innehatte und
                # Wikidata die Zeitangaben an einer Aussage buendelt: Max
                # Brauer, Erster Buergermeister von Hamburg 1946-1953 und
                # 1957-1960. Die Paarung ist dann nicht rekonstruierbar,
                # und geraten wird hier nicht.
                verworfen.append((amt_id, person, von, bis))
                continue
            inhaber.append(
                {
                    "amt_id": amt_id,
                    "person_id": person,
                    "von": von,
                    "bis": bis,
                    "quelle": quelle,
                    "notiz": None,
                }
            )
    if verworfen:
        print(f"  {len(verworfen)} Amtszeiten mit Ende vor Anfang verworfen "
              f"(z.B. {verworfen[0][1]} in {verworfen[0][0]})")
    return list(akteure.values()), inhaber


def laden(nur=None, conn=None):
    import db

    akteure, inhaber = zeilen_aus_spiegel(nur=nur)
    eigen = conn is None
    verbindung = db.verbindung() if eigen else conn
    try:
        with verbindung.cursor() as k:
            k.executemany(
                """insert into akteure
                       (id, typ, name, aliase, land, rolle, wikidata_qid, quelle, notiz)
                   values (%(id)s, %(typ)s, %(name)s, %(aliase)s, %(land)s,
                           %(rolle)s, %(wikidata_qid)s, %(quelle)s, %(notiz)s)
                   on conflict (id) do update set
                       typ = excluded.typ, name = excluded.name,
                       aliase = excluded.aliase, land = excluded.land,
                       rolle = excluded.rolle, wikidata_qid = excluded.wikidata_qid,
                       quelle = excluded.quelle, notiz = excluded.notiz""",
                akteure,
            )
            if inhaber:
                k.executemany(
                    """insert into amtsinhaber (amt_id, person_id, von, bis, quelle, notiz)
                       values (%(amt_id)s, %(person_id)s, %(von)s, %(bis)s,
                               %(quelle)s, %(notiz)s)
                       on conflict (amt_id, person_id, von) do update set
                           bis = excluded.bis, quelle = excluded.quelle""",
                    inhaber,
                )
            verbindung.commit()
            k.execute("select typ, count(*) from akteure group by 1 order by 1")
            for typ, n in k.fetchall():
                print(f"  {typ:13} {n:6}")
            k.execute("select count(*) from amtsinhaber")
            print(f"  amtsinhaber   {k.fetchone()[0]:6}")
            k.execute("select * from amtsinhaber_ueberschneidungen")
            for r in k.fetchall():
                print(f"  Ueberschneidung: {r}")
    finally:
        if eigen:
            verbindung.close()
    return len(akteure), len(inhaber)


if __name__ == "__main__":
    laden()