swp-02-aufbereitung/lexikon.py

"""Akteurslexikon: Wikidata spiegeln, Spiegel laden.

Zwei streng getrennte Betriebsarten, und die Trennung ist der Zweck der
Datei:

    lexikon.py --spiegeln    fragt Wikidata ab und schreibt lexikon/*.json
    lexikon.py --laden       liest lexikon/*.json und fuellt die Tabellen

Nur --spiegeln fasst das Netz an, und es laeuft von Hand, nicht im Takt.
Der Slot-Lauf der Phase 2 sieht Wikidata nie. Das ist keine Feinheit: die
Wikidata Query Service ist ein von Spenden getragener oeffentlicher Dienst,
und eine Pipeline, die alle 15 Minuten dagegenlaeuft, ist Schmarotzertum.

Der Spiegel ist eine getreue Abschrift, keine Auslegung. Was in
lexikon/*.json steht, steht so in Wikidata; jede Ableitung - Kurzformen,
Rollenzuordnung, Normalisierung - geschieht beim Laden und ist damit
nachvollziehbar und wiederholbar, ohne erneut abzufragen.
"""

import argparse
import datetime
import hashlib
import json
import os
import pathlib
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

VERZEICHNIS = pathlib.Path(__file__).parent / "lexikon"
ENDPUNKT = "https://query.wikidata.org/sparql"

# Die Nutzungsrichtlinie der WDQS bittet um einen sprechenden User-Agent mit
# Kontaktangabe. Hier steht die Impressumsadresse des Projekts - eine
# oeffentliche Adresse, die genau dafuer da ist.
KENNUNG = os.environ.get(
    "WDQS_USER_AGENT",
    "wurzelwerk-p2/0.1 (Datenjournalismus-Pipeline Wurzelwerkt; stress@bnd.wtf)",
)

# Pause zwischen zwei Abfragen. Die WDQS erlaubt mehr, aber wir haben es
# nicht eilig - der Spiegel entsteht einmal und liegt dann Monate.
PAUSE = 2.0
VERSUCHE = 3
ZEITGRENZE = 120

# Stichtag fuer Personen: wer vor 2017 aus allen Aemtern ausgeschieden ist,
# taucht in Nachrichten von 2026 hoechstens als Rueckblick auf. Die Grenze
# ist willkuerlich, aber sie steht hier und nicht verstreut in Abfragen.
STICHTAG = "2017-01-01"

# Aemter, die in deutschen Schlagzeilen ohne Namen auftauchen ("der
# Kanzler", "die Aussenministerin"). Handverlesen, weil eine generische
# Abfrage hier tausende Aemter liefert, von denen keines je in einer
# Schlagzeile steht. Das ist die "manuelle Ergaenzung" aus der Spezifikation.
AEMTER = {
    "Q4970706": ("amt:bundeskanzler", "Bundeskanzler", "GOV"),
    "Q25223": ("amt:bundespraesident", "Bundespraesident", "GOV"),
    "Q56024": ("amt:bundestagspraesident", "Praesident des Deutschen Bundestages", "LEG"),
    "Q568605": ("amt:aussenminister", "Bundesminister des Auswaertigen", "GOV"),
    "Q22575628": ("amt:innenminister", "Bundesminister des Innern", "GOV"),
    "Q22575092": ("amt:finanzminister", "Bundesminister der Finanzen", "GOV"),
    "Q23016036": ("amt:verteidigungsminister", "Bundesminister der Verteidigung", "GOV"),
    "Q30533368": ("amt:gesundheitsminister", "Bundesminister fuer Gesundheit", "GOV"),
    "Q30543924": ("amt:verkehrsminister", "Bundesminister fuer Verkehr", "GOV"),
    "Q30533351": ("amt:arbeitsminister", "Bundesminister fuer Arbeit und Soziales", "GOV"),
    "Q30544097": ("amt:umweltminister", "Bundesminister fuer Umwelt", "GOV"),
    # Justiz und Wirtschaft fuehrt Wikidata je Zuschnitt als eigenes Amt.
    # Fuer uns ist es dasselbe Amt: die Ressortnamen wechseln, die Rolle in
    # der Schlagzeile nicht. Mehrere QIDs zeigen deshalb auf eine amt_id.
    "Q30533247": ("amt:justizminister", "Bundesminister der Justiz", "GOV"),
    "Q30542236": ("amt:justizminister", "Bundesminister der Justiz", "GOV"),
    "Q30543192": ("amt:wirtschaftsminister", "Bundesminister fuer Wirtschaft", "GOV"),
    "Q30533307": ("amt:wirtschaftsminister", "Bundesminister fuer Wirtschaft", "GOV"),
    "Q30319173": ("amt:wirtschaftsminister", "Bundesminister fuer Wirtschaft", "GOV"),
    # Regierungschefs der Laender. Jedes Land fuehrt sein eigenes Amt als
    # eigenes Wikidata-Objekt; einen gemeinsamen Oberbegriff, ueber den sich
    # alle sechzehn einsammeln liessen, gibt es nicht. Also stehen sie hier.
    "Q30968305": ("amt:mp_sachsen_anhalt", "Ministerpraesident Sachsen-Anhalts", "GOV"),
    "Q30569887": ("amt:mp_sachsen", "Ministerpraesident Sachsens", "GOV"),
    "Q30584228": ("amt:mp_thueringen", "Ministerpraesident Thueringens", "GOV"),
    "Q30562601": ("amt:mp_brandenburg", "Ministerpraesident Brandenburgs", "GOV"),
    "Q14917672": ("amt:mp_mecklenburg_vorpommern", "Ministerpraesident Mecklenburg-Vorpommerns", "GOV"),
    "Q15303555": ("amt:mp_bayern", "Ministerpraesident Bayerns", "GOV"),
    "Q31096505": ("amt:mp_baden_wuerttemberg", "Ministerpraesident Baden-Wuerttembergs", "GOV"),
    "Q30534793": ("amt:mp_nordrhein_westfalen", "Ministerpraesident Nordrhein-Westfalens", "GOV"),
    "Q31524732": ("amt:mp_niedersachsen", "Niedersaechsischer Ministerpraesident", "GOV"),
    "Q1615935":  ("amt:mp_hessen", "Ministerpraesident Hessens", "GOV"),
    "Q15720142": ("amt:mp_rheinland_pfalz", "Ministerpraesident von Rheinland-Pfalz", "GOV"),
    "Q31068298": ("amt:mp_saarland", "Ministerpraesident des Saarlandes", "GOV"),
    "Q30263013": ("amt:mp_schleswig_holstein", "Ministerpraesident Schleswig-Holsteins", "GOV"),
    "Q641159":   ("amt:rb_berlin", "Regierender Buergermeister von Berlin", "GOV"),
    "Q1362118":  ("amt:eb_hamburg", "Erster Buergermeister von Hamburg", "GOV"),
    "Q31388880": ("amt:b_bremen", "Bremer Buergermeister", "GOV"),
}

_PERSONEN_RUMPF = """
  ?p rdfs:label ?name . FILTER (lang(?name) = "de")
  OPTIONAL { ?p skos:altLabel ?alias . FILTER (lang(?alias) = "de") }
  OPTIONAL { ?p wdt:P102 ?pt . ?pt rdfs:label ?ptl . FILTER (lang(?ptl) = "de") }
"""

ABFRAGEN = {
    # Bundestagsabgeordnete, alle seit dem Stichtag amtierenden.
    "bundestag": """
SELECT ?p ?name (GROUP_CONCAT(DISTINCT ?alias; separator="|") AS ?aliase)
       (SAMPLE(?ptl) AS ?partei)
WHERE {
  ?p p:P39 ?s . ?s ps:P39 ?a . ?a wdt:P279* wd:Q1939555 .
  OPTIONAL { ?s pq:P582 ?ende . }
  FILTER (!BOUND(?ende) || ?ende >= "%(stichtag)s"^^xsd:dateTime)
  %(rumpf)s
} GROUP BY ?p ?name
""",
    # Parteien in Deutschland, Oesterreich und der Schweiz.
    "parteien": """
SELECT ?p ?name (GROUP_CONCAT(DISTINCT ?alias; separator="|") AS ?aliase)
       (SAMPLE(?landc) AS ?land)
WHERE {
  ?p wdt:P31/wdt:P279* wd:Q7278 .
  ?p wdt:P17 ?l . VALUES ?l { wd:Q183 wd:Q40 wd:Q39 }
  OPTIONAL { ?l wdt:P298 ?landc . }
  ?p rdfs:label ?name . FILTER (lang(?name) = "de")
  OPTIONAL { ?p skos:altLabel ?alias . FILTER (lang(?alias) = "de") }
  FILTER NOT EXISTS { ?p wdt:P576 ?aufgeloest . FILTER (?aufgeloest < "%(stichtag)s"^^xsd:dateTime) }
} GROUP BY ?p ?name
""",
    # Souveraene Staaten mit ISO-3166-1 alpha-3.
    "staaten": """
SELECT ?p ?name (GROUP_CONCAT(DISTINCT ?alias; separator="|") AS ?aliase) ?land
WHERE {
  ?p wdt:P31 wd:Q3624078 ; wdt:P298 ?land .
  FILTER NOT EXISTS { ?p wdt:P576 ?ende . }
  ?p rdfs:label ?name . FILTER (lang(?name) = "de")
  OPTIONAL { ?p skos:altLabel ?alias . FILTER (lang(?alias) = "de") }
} GROUP BY ?p ?name ?land
""",
    # Amtierende Staats- und Regierungschefs aller Staaten. Ohne sie fehlen
    # die haeufigsten Personen des Korpus, denn im Bundestag sass keiner von
    # ihnen. Nur der aktuelle Stand - Vorgaenger holt die Amtstabelle.
    "staatsfuehrung": """
SELECT ?p ?name (GROUP_CONCAT(DISTINCT ?alias; separator="|") AS ?aliase) ?land
WHERE {
  ?l wdt:P31 wd:Q3624078 ; wdt:P298 ?land .
  { ?l wdt:P35 ?p } UNION { ?l wdt:P6 ?p }
  ?p rdfs:label ?name . FILTER (lang(?name) = "de")
  OPTIONAL { ?p skos:altLabel ?alias . FILTER (lang(?alias) = "de") }
} GROUP BY ?p ?name ?land
""",
    # Boersennotierte Unternehmen aus den Leitindizes. Sie sind Akteure wie
    # Parteien es sind - "VW streicht Stellen" ist eine Handlung. Ueber die
    # Indexmitgliedschaft statt ueber "Unternehmen in Deutschland", weil
    # letzteres Hunderttausende liefert und keine Auswahl trifft.
    "unternehmen": """
SELECT ?p ?name (GROUP_CONCAT(DISTINCT ?alias; separator="|") AS ?aliase) (SAMPLE(?landc) AS ?land)
WHERE {
  ?p wdt:P361 ?idx .
  VALUES ?idx { wd:Q155718 wd:Q457539 wd:Q457540 wd:Q642856 wd:Q319415
                wd:Q180816 wd:Q242345 }
  OPTIONAL { ?p wdt:P17 ?l . ?l wdt:P298 ?landc . }
  ?p rdfs:label ?name . FILTER (lang(?name) = "de")
  OPTIONAL { ?p skos:altLabel ?alias . FILTER (lang(?alias) = "de") }
} GROUP BY ?p ?name
""",
    # Internationale Organisationen.
    "organisationen": """
SELECT ?p ?name (GROUP_CONCAT(DISTINCT ?alias; separator="|") AS ?aliase)
WHERE {
  ?p wdt:P31/wdt:P279* wd:Q484652 .
  FILTER NOT EXISTS { ?p wdt:P576 ?ende . }
  ?p rdfs:label ?name . FILTER (lang(?name) = "de")
  OPTIONAL { ?p skos:altLabel ?alias . FILTER (lang(?alias) = "de") }
} GROUP BY ?p ?name
""",
}

# Amtsinhaber je Amt, mit Anfang und Ende. Ein Amt je Abfrage - das ist
# langsamer, aber jede Antwort bleibt klein und nachvollziehbar.
ABFRAGE_INHABER = """
SELECT ?p ?name ?von ?bis WHERE {
  ?p p:P39 ?s . ?s ps:P39 wd:%(qid)s .
  OPTIONAL { ?s pq:P580 ?von . }
  OPTIONAL { ?s pq:P582 ?bis . }
  ?p rdfs:label ?name . FILTER (lang(?name) = "de")
}
"""


def _abfragetext(name):
    if name in ABFRAGEN:
        return ABFRAGEN[name] % {"stichtag": STICHTAG, "rumpf": _PERSONEN_RUMPF}
    raise KeyError(name)


def frage(sparql):
    """Eine SPARQL-Abfrage, mit Wiederholung bei 429 und 5xx."""
    ziel = ENDPUNKT + "?" + urllib.parse.urlencode({"query": sparql, "format": "json"})
    anfrage = urllib.request.Request(
        ziel,
        headers={"Accept": "application/sparql-results+json", "User-Agent": KENNUNG},
    )
    for versuch in range(1, VERSUCHE + 1):
        try:
            with urllib.request.urlopen(anfrage, timeout=ZEITGRENZE) as antwort:
                return json.load(antwort)["results"]["bindings"]
        except urllib.error.HTTPError as fehler:
            if fehler.code not in (429, 500, 502, 503, 504) or versuch == VERSUCHE:
                raise
            # Retry-After ernst nehmen, sonst geometrisch zurueckweichen.
            warte = float(fehler.headers.get("Retry-After") or 0) or PAUSE * 2**versuch
            print(f"    HTTP {fehler.code}, warte {warte:.0f}s", file=sys.stderr)
            time.sleep(warte)
    raise RuntimeError("unerreichbar")


def _werte(bindungen):
    """SPARQL-Bindungen auf schlichte Dicts eindampfen."""
    return [{k: v["value"] for k, v in b.items()} for b in bindungen]


def spiegeln(nur=None, erneuern=False):
    VERZEICHNIS.mkdir(exist_ok=True)
    stand = datetime.date.today().isoformat()
    manifest = {}
    manifestdatei = VERZEICHNIS / "MANIFEST.json"
    if manifestdatei.exists():
        manifest = json.loads(manifestdatei.read_text())

    auftraege = [(n, _abfragetext(n)) for n in ABFRAGEN if not nur or n in nur]
    if not nur or "aemter" in nur:
        auftraege.append(("aemter", None))

    for name, sparql in auftraege:
        datei = VERZEICHNIS / f"{name}.json"
        if datei.exists() and not erneuern:
            print(f"{name}: Spiegel vorhanden, uebersprungen (--erneuern erzwingt)")
            continue

        if name == "aemter":
            zeilen = []
            for qid, (_id, _n, _r) in AEMTER.items():
                zeilen_amt = _werte(frage(ABFRAGE_INHABER % {"qid": qid}))
                for z in zeilen_amt:
                    z["amt_qid"] = qid
                zeilen.extend(zeilen_amt)
                print(f"  {qid}: {len(zeilen_amt)} Inhaber")
                time.sleep(PAUSE)
            fingerabdruck = hashlib.blake2b(
                json.dumps(sorted(AEMTER), ensure_ascii=False).encode(), digest_size=8
            ).hexdigest()
        else:
            zeilen = _werte(frage(sparql))
            fingerabdruck = hashlib.blake2b(sparql.encode(), digest_size=8).hexdigest()
            time.sleep(PAUSE)

        datei.write_text(
            json.dumps(zeilen, ensure_ascii=False, indent=1, sort_keys=True) + "\n"
        )
        manifest[name] = {
            "stand": stand,
            "zeilen": len(zeilen),
            "abfrage_hash": fingerabdruck,
            "endpunkt": ENDPUNKT,
        }
        print(f"{name}: {len(zeilen)} Zeilen -> {datei}")

    manifestdatei.write_text(
        json.dumps(manifest, ensure_ascii=False, indent=1, sort_keys=True) + "\n"
    )
    return manifest


def main(argumente=None):
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    p.add_argument("--spiegeln", action="store_true", help="Wikidata abfragen")
    p.add_argument("--laden", action="store_true", help="Spiegel in die Tabellen laden")
    p.add_argument("--nur", nargs="*", help="nur diese Abfragen")
    p.add_argument("--erneuern", action="store_true", help="vorhandenen Spiegel ersetzen")
    a = p.parse_args(argumente)

    if a.spiegeln:
        spiegeln(nur=a.nur, erneuern=a.erneuern)
    if a.laden:
        import lexikon_laden

        lexikon_laden.laden(nur=a.nur)
    if not (a.spiegeln or a.laden):
        p.error("--spiegeln oder --laden angeben")


if __name__ == "__main__":
    main()