""" Erzeugt aufgeteilte, gzip-komprimierte JSON-Dateien aus den Wahlergebnis-Daten. Liest das amtliche Ergebnis-XLSX (Amt für Statistik Berlin-Brandenburg) und aggregiert die Prozentwerte je Bezirk, Wahlkreis und Briefwahlbezirk – für AGH (Abgeordnetenhaus, Erst-/Zweitstimme) und BVV (Bezirksverordnetenversammlung). Ausgabedateien (alle unter json/, gzip-komprimiert): ERGEBNISSE_agh_bezirke_{JAHR}.json.gz – meta, gesamt, Bezirke, Wahlkreise, Erststimmen ERGEBNISSE_agh_wahlbezirke_{JAHR}.json.gz – AGH-Urnenwahlbezirke + Erststimmen je WB ERGEBNISSE_agh_bwb_{JAHR}.json.gz – AGH-Briefwahlbezirke + Erststimmen je BWB ERGEBNISSE_bvv_bezirke_{JAHR}.json.gz – BVV-Bezirke ERGEBNISSE_bvv_wahlbezirke_{JAHR}.json.gz – BVV-Urnenwahlbezirke ERGEBNISSE_bvv_bwb_{JAHR}.json.gz – BVV-Briefwahlbezirke Bei einer neuen Wahl: XLSX-Datei in csv/ ablegen, Konfiguration unten (Dateiname, WAHL_NAME, JAHR, ggf. PARTEIEN) anpassen und Skript erneut ausführen: python build_results.py Wahlkreise/Bezirke, für die keine Daten vorliegen (z.B. neu zugeschnittene Wahlkreise), tauchen im Ergebnis schlicht nicht auf – die Karte zeigt für sie dann keine Ergebnisdaten an (siehe getRes() in wahlkarte-embed.js). """ import gzip import json from collections import Counter from pathlib import Path import openpyxl BASE = Path(__file__).parent # ── Konfiguration: bei neuen Daten hier anpassen ─────────────────────────── XLSX = BASE / "csv" / "DL_BE_AGHBVV2023.xlsx" SHEET_AGH_ERST = "AGH_W1" # Erststimme (Direktkandidat:innen) SHEET_AGH_ZWEIT = "AGH_W2" # Zweitstimme (Parteiergebnis AGH) SHEET_BVV = "BVV" # BVV-Ergebnis WAHL_NAME = "AH 2023 (Wiederholungswahl)" JAHR = 2023 JSON_DIR = BASE / "json" DIREKTKANDIDATEN_JSON = JSON_DIR / "Direktkandidaten.json" # CSV-Spaltenname -> Anzeigename in 'ergebnis'. Diese etablierten Parteien # stehen in fester Reihenfolge zuerst; alle übrigen Parteien/Listen aus dem # XLSX folgen danach in Original-Reihenfolge. PARTEIEN = { "SPD": "SPD", "CDU": "CDU", "GRÜNE": "Grüne", "DIE LINKE": "Linke", "AfD": "AfD", "FDP": "FDP", "BSW": "BSW", } # Spalten ab 'Ungültige Stimmen', die keine Parteien/Listen sind IGNORE_SPALTEN: set = set() ERSTSTIMME_IGNORE: set = set() WK_SPALTE = "Abgeordneten-\nhauswahlkreis" # ── Hilfsfunktionen ──────────────────────────────────────────────────────── def _disambiguiere_header(header, raw_rows): """Benennt mehrfach vorkommende Spaltennamen eindeutig um (z.B. 'WB WsB').""" counts = Counter(header) dupes = {name for name, n in counts.items() if n > 1 and name and name not in IGNORE_SPALTEN} if not dupes: return header bez_idx = header.index("Bezirksnummer") bezname_idx = header.index("Bezirksname") new_header = list(header) for col_idx, name in enumerate(header): if name not in dupes: continue bez_namen = sorted({ row[bezname_idx] for row in raw_rows if row[bez_idx] and to_int(row[col_idx]) }) if bez_namen: new_header[col_idx] = f"{name} {' / '.join(bez_namen)}" return new_header def _str_bez(v): """Bezirksnummer zuverlässig als zweistelligen String (XLSX liefert teils int, teils str).""" if v is None: return None try: return str(int(str(v).strip())).zfill(2) except (ValueError, TypeError): return str(v) def load_sheet(sheet_name): """Liest ein XLSX-Tabellenblatt und gibt (header, rows) zurück.""" wb = openpyxl.load_workbook(XLSX, read_only=True, data_only=True) ws = wb[sheet_name] all_rows = list(ws.iter_rows(values_only=True)) wb.close() header = [str(h).strip() if h is not None else "" for h in all_rows[0]] raw_rows = [list(r) for r in all_rows[1:] if any(c is not None for c in r)] header = _disambiguiere_header(header, raw_rows) rows = [dict(zip(header, r)) for r in raw_rows] # Bezirksnummer einheitlich als zweistelliger String (XLSX liefert gemischte Typen) for r in rows: r["Bezirksnummer"] = _str_bez(r.get("Bezirksnummer")) return header, rows def to_int(value): if value is None: return 0 if isinstance(value, int): return value try: return int(str(value).strip()) except (ValueError, TypeError): return 0 def bwb_id(row): """Vollständiger BWB-Code: BEZ (2 Ziffern) + Briefwahlbezirk-Kurzform.""" return str(row["Bezirksnummer"]) + str(row["Briefwahlbezirk"]) def wk_id(row): """Vollständiger AWK-Schlüssel: BEZ + AWK zweistellig (z.B. '0101').""" return str(row["Bezirksnummer"]) + str(int(row[WK_SPALTE])).zfill(2) def partei_spalten(header): """Alle Partei-/Listen-Spaltennamen in Original-Reihenfolge.""" start = header.index("Ungültige Stimmen") + 1 return [h for h in header[start:] if h and h not in IGNORE_SPALTEN] def ergebnis(rows, header): """Prozentanteile je Partei/Liste (gerundet auf 1 Nachkommastelle).""" gueltig = sum(to_int(r["Gültige Stimmen"]) for r in rows) erg = {} for spalte in partei_spalten(header): label = PARTEIEN.get(spalte, spalte) stimmen = sum(to_int(r.get(spalte)) for r in rows) erg[label] = round(stimmen / gueltig * 100, 1) if gueltig else 0.0 if "BSW" not in erg: erg["BSW"] = 0.0 geordnet = {p: erg.pop(p) for p in PARTEIEN.values() if p in erg} geordnet.update(erg) return geordnet def ergebnis_absolut(rows, header): """Absolute Stimmenzahlen je Partei/Liste.""" erg = {} for spalte in partei_spalten(header): label = PARTEIEN.get(spalte, spalte) erg[label] = sum(to_int(r.get(spalte)) for r in rows) if "BSW" not in erg: erg["BSW"] = 0 geordnet = {p: erg.pop(p) for p in PARTEIEN.values() if p in erg} geordnet.update(erg) return geordnet def beteiligung(rows): """Wahlbeteiligung in %; auf 100 % gekappt (Datenproblem einiger BWBs).""" wahlberechtigte = sum(to_int(r["Wahlberechtigte insgesamt"]) for r in rows) waehlende = sum(to_int(r["Wählende"]) for r in rows) if wahlberechtigte == 0: return 0.0 return min(100.0, round(waehlende / wahlberechtigte * 100, 1)) def waehler_absolut(rows): """Wahlberechtigte und abgegebene Stimmen als absolute Zahlen.""" return { "wahlberechtigte": sum(to_int(r["Wahlberechtigte insgesamt"]) for r in rows), "abgegeben": sum(to_int(r["Wählende"]) for r in rows), } def briefwahlanteil(rows): """Anteil der Briefwahlstimmen an allen Wählenden (in %).""" gesamt = sum(to_int(r["Wählende"]) for r in rows) brief = sum(to_int(r["Wählende"]) for r in rows if r.get("Wahlbezirksart") == "B") return round(brief / gesamt * 100, 1) if gesamt else 0.0 def mehrheits_ostwest(rows): werte = Counter(r["OstWest"] for r in rows if r.get("OstWest")) return werte.most_common(1)[0][0] if werte else None def erststimmen_anteile(rows, header): """Erststimmen-Prozentanteile aller Parteien/Kandidat:innen (1 Nachkommastelle).""" gueltig = sum(to_int(r["Gültige Stimmen"]) for r in rows) if gueltig == 0: return None return { p: round(sum(to_int(r.get(p)) for r in rows) / gueltig * 100, 1) for p in partei_spalten(header) } def erststimmen_absolut(rows, header): """Absolute Erststimmenzahlen aller Parteien/Kandidat:innen.""" gueltig = sum(to_int(r["Gültige Stimmen"]) for r in rows) if gueltig == 0: return None return { p: sum(to_int(r.get(p)) for r in rows) for p in partei_spalten(header) } def lade_direktkandidaten(): """Lädt Kandidat:innennamen je Wahlkreis aus json/Direktkandidaten.json (falls vorhanden).""" if not DIREKTKANDIDATEN_JSON.exists(): return {} with open(DIREKTKANDIDATEN_JSON, encoding="utf-8") as f: return json.load(f) def direktkandidat_partei(rows, header): """Partei mit den meisten Erststimmen in den übergebenen Zeilen.""" summen = Counter() for r in rows: for spalte in partei_spalten(header): if spalte in ERSTSTIMME_IGNORE: continue summen[spalte] += to_int(r.get(spalte)) if not summen: return None sieger = summen.most_common(1)[0][0] return PARTEIEN.get(sieger, sieger) # ── AGH: Wahlbezirke, Wahlkreise, Bezirke ────────────────────────────────── def build_agh(header_zweit, rows_zweit, header_erst, rows_erst): direktkandidaten = lade_direktkandidaten() bezirke, wahlkreise, wahlbezirke = {}, {}, {} erststimmen, erststimmen_wahlbezirke = {}, {} bezirke_abs, wahlkreise_abs, wahlbezirke_abs = {}, {}, {} erststimmen_abs, erststimmen_wahlbezirke_abs = {}, {} bezirk_ids = sorted({r["Bezirksnummer"] for r in rows_zweit}) for bid in bezirk_ids: rows_b = [r for r in rows_zweit if r["Bezirksnummer"] == bid] rows_bu = [r for r in rows_b if r["Wahlbezirksart"] == "W"] bezirksname = rows_b[0]["Bezirksname"] bezirke[bid] = { "name": bezirksname, "beteiligung": beteiligung(rows_b), "beteiligung_vw": None, "vorwahl": None, "ergebnis": ergebnis(rows_b, header_zweit), "waehler": waehler_absolut(rows_b), "ostwest": mehrheits_ostwest(rows_bu), "briefwahlanteil": briefwahlanteil(rows_b), } bezirke_abs[bid] = {"ergebnis": ergebnis_absolut(rows_b, header_zweit)} # Wahlkreise innerhalb des Bezirks for wk in sorted({r[WK_SPALTE] for r in rows_b}): wkid = bid + str(int(wk)).zfill(2) rows_wk = [r for r in rows_b if r[WK_SPALTE] == wk] rows_wku = [r for r in rows_wk if r["Wahlbezirksart"] == "W"] rows_we = [ r for r in rows_erst if r["Bezirksnummer"] == bid and r[WK_SPALTE] == wk ] rows_weu = [r for r in rows_we if r["Wahlbezirksart"] == "W"] partei = direktkandidat_partei(rows_weu, header_erst) kandidat_name = ( direktkandidaten.get(wkid, {}).get("kandidaten", {}).get(partei) if partei else None ) anteile = erststimmen_anteile(rows_we, header_erst) if anteile is not None: erststimmen[wkid] = anteile abs_erst = erststimmen_absolut(rows_we, header_erst) if abs_erst is not None: erststimmen_abs[wkid] = abs_erst wahlkreise[wkid] = { "name": f"{bezirksname} {int(wk)}", "bezirk": bid, "beteiligung": beteiligung(rows_wk), "beteiligung_vw": None, "vorwahl": None, "ergebnis": ergebnis(rows_wk, header_zweit), "kandidat": {"name": kandidat_name, "partei": partei} if partei else None, "waehler": waehler_absolut(rows_wk), "ostwest": mehrheits_ostwest(rows_wku), "briefwahlanteil": briefwahlanteil(rows_wk), } wahlkreise_abs[wkid] = {"ergebnis": ergebnis_absolut(rows_wk, header_zweit)} # Urnenwahlbezirke (für Fallback/Archiv) for wbnr in sorted({r["Wahlbezirk"] for r in rows_bu}): wbid = bid + str(wbnr) rows_wb = [r for r in rows_bu if r["Wahlbezirk"] == wbnr] wahlbezirke[wbid] = { "name": f"{bezirksname} {bid}{int(rows_wb[0][WK_SPALTE]):02d}{str(wbnr).zfill(3)}", "beteiligung": beteiligung(rows_wb), "beteiligung_label": "Urnenbeteiligung", "beteiligung_vw": None, "vorwahl": None, "ergebnis": ergebnis(rows_wb, header_zweit), "waehler": waehler_absolut(rows_wb), "ostwest": mehrheits_ostwest(rows_wb), } wahlbezirke_abs[wbid] = {"ergebnis": ergebnis_absolut(rows_wb, header_zweit)} # Erststimmen per Urnenwahlbezirk urne_erst = [r for r in rows_erst if r["Wahlbezirksart"] == "W"] wb_eg: dict = {} for r in urne_erst: wbid = r["Bezirksnummer"] + str(r["Wahlbezirk"]) wb_eg.setdefault(wbid, []).append(r) for wbid, wb_rows in wb_eg.items(): a = erststimmen_anteile(wb_rows, header_erst) if a is not None: erststimmen_wahlbezirke[wbid] = a a_abs = erststimmen_absolut(wb_rows, header_erst) if a_abs is not None: erststimmen_wahlbezirke_abs[wbid] = a_abs # Gesamt-Berlin rows_gu = [r for r in rows_zweit if r["Wahlbezirksart"] == "W"] gesamt = { "name": "Berlin", "beteiligung": beteiligung(rows_zweit), "beteiligung_vw": None, "vorwahl": None, "ergebnis": ergebnis(rows_gu, header_zweit), "erststimmen": erststimmen_anteile(urne_erst, header_erst), "waehler": waehler_absolut(rows_zweit), "briefwahlanteil": briefwahlanteil(rows_zweit), } gesamt_abs = { "ergebnis": ergebnis_absolut(rows_gu, header_zweit), "erststimmen": erststimmen_absolut(urne_erst, header_erst), } return ( bezirke, wahlkreise, wahlbezirke, erststimmen, erststimmen_wahlbezirke, gesamt, bezirke_abs, wahlkreise_abs, wahlbezirke_abs, erststimmen_abs, erststimmen_wahlbezirke_abs, gesamt_abs, ) # ── AGH: Briefwahlbezirke ────────────────────────────────────────────────── def build_bwb(header_zweit, rows_zweit, header_erst, rows_erst): """Ergebnisse je Briefwahlbezirk (BWB): Urnen- + Briefwahlstimmen kombiniert.""" briefwahlbezirke, erststimmen_bwb = {}, {} briefwahlbezirke_abs, erststimmen_bwb_abs = {}, {} for bwb in sorted({bwb_id(r) for r in rows_zweit}): rows_bwb = [r for r in rows_zweit if bwb_id(r) == bwb] bezirksname = rows_bwb[0]["Bezirksname"] briefwahlbezirke[bwb] = { "name": f"{bezirksname} {bwb[:2]}{int(rows_bwb[0][WK_SPALTE]):02d}{bwb[2:]}", "beteiligung": beteiligung(rows_bwb), "beteiligung_vw": None, "vorwahl": None, "ergebnis": ergebnis(rows_bwb, header_zweit), "waehler": waehler_absolut(rows_bwb), "ostwest": mehrheits_ostwest(rows_bwb), "briefwahlanteil": briefwahlanteil(rows_bwb), } briefwahlbezirke_abs[bwb] = {"ergebnis": ergebnis_absolut(rows_bwb, header_zweit)} for bwb in sorted({bwb_id(r) for r in rows_erst}): rows_bwb_e = [r for r in rows_erst if bwb_id(r) == bwb] a = erststimmen_anteile(rows_bwb_e, header_erst) if a is not None: erststimmen_bwb[bwb] = a a_abs = erststimmen_absolut(rows_bwb_e, header_erst) if a_abs is not None: erststimmen_bwb_abs[bwb] = a_abs return briefwahlbezirke, erststimmen_bwb, briefwahlbezirke_abs, erststimmen_bwb_abs # ── BVV: Wahlbezirke, Bezirke ────────────────────────────────────────────── def build_bvv(header_bvv, rows_bvv): bezirke, wahlbezirke = {}, {} bezirke_abs, wahlbezirke_abs = {}, {} for bid in sorted({r["Bezirksnummer"] for r in rows_bvv}): rows_b = [r for r in rows_bvv if r["Bezirksnummer"] == bid] rows_bu = [r for r in rows_b if r["Wahlbezirksart"] == "W"] bezirksname = rows_b[0]["Bezirksname"] bezirke[bid] = { "name": bezirksname, "beteiligung": beteiligung(rows_b), "beteiligung_vw": None, "vorwahl": None, "ergebnis": ergebnis(rows_b, header_bvv), "briefwahlanteil": briefwahlanteil(rows_b), } bezirke_abs[bid] = {"ergebnis": ergebnis_absolut(rows_b, header_bvv)} for wbnr in sorted({r["Wahlbezirk"] for r in rows_bu}): wbid = bid + str(wbnr) rows_wb = [r for r in rows_bu if r["Wahlbezirk"] == wbnr] wahlbezirke[wbid] = { "name": f"{bezirksname} {bid}{int(rows_wb[0][WK_SPALTE]):02d}{str(wbnr).zfill(3)}", "beteiligung": beteiligung(rows_wb), "beteiligung_label": "Urnenbeteiligung", "beteiligung_vw": None, "vorwahl": None, "ergebnis": ergebnis(rows_wb, header_bvv), } wahlbezirke_abs[wbid] = {"ergebnis": ergebnis_absolut(rows_wb, header_bvv)} return bezirke, wahlbezirke, bezirke_abs, wahlbezirke_abs # ── BVV: Briefwahlbezirke ────────────────────────────────────────────────── def build_bvv_bwb(header_bvv, rows_bvv): bvv_bwb, bvv_bwb_abs = {}, {} for bwb in sorted({bwb_id(r) for r in rows_bvv}): rows_bwb = [r for r in rows_bvv if bwb_id(r) == bwb] bezirksname = rows_bwb[0]["Bezirksname"] bvv_bwb[bwb] = { "name": f"{bezirksname} {bwb[:2]}{int(rows_bwb[0][WK_SPALTE]):02d}{bwb[2:]}", "beteiligung": beteiligung(rows_bwb), "beteiligung_vw": None, "vorwahl": None, "ergebnis": ergebnis(rows_bwb, header_bvv), "briefwahlanteil": briefwahlanteil(rows_bwb), } bvv_bwb_abs[bwb] = {"ergebnis": ergebnis_absolut(rows_bwb, header_bvv)} return bvv_bwb, bvv_bwb_abs def write_gz(path: Path, data: dict) -> int: """Schreibt data als komprimiertes JSON; gibt die Dateigröße in Bytes zurück.""" raw = json.dumps(data, ensure_ascii=False, separators=(",", ":")).encode() with gzip.open(path, "wb", compresslevel=9) as f: f.write(raw) return path.stat().st_size def main(): print(f"Lese {XLSX} …") header_zweit, rows_zweit = load_sheet(SHEET_AGH_ZWEIT) header_erst, rows_erst = load_sheet(SHEET_AGH_ERST) header_bvv, rows_bvv = load_sheet(SHEET_BVV) (bezirke, wahlkreise, wahlbezirke, erststimmen, erststimmen_wahlbezirke, gesamt, bezirke_abs, wahlkreise_abs, wahlbezirke_abs, erststimmen_abs, erststimmen_wahlbezirke_abs, gesamt_abs) = \ build_agh(header_zweit, rows_zweit, header_erst, rows_erst) bvv_bezirke, bvv_wahlbezirke, bvv_bezirke_abs, bvv_wahlbezirke_abs = build_bvv(header_bvv, rows_bvv) bvv_briefwahlbezirke, bvv_briefwahlbezirke_abs = build_bvv_bwb(header_bvv, rows_bvv) briefwahlbezirke, erststimmen_briefwahlbezirke, briefwahlbezirke_abs, erststimmen_briefwahlbezirke_abs = \ build_bwb(header_zweit, rows_zweit, header_erst, rows_erst) meta = { "wahl": WAHL_NAME, "jahr": JAHR, "auszaehlung": {"prozent": None, "uhrzeit": None, "ausgezaehlt": None, "gesamt": None}, } files = { f"ERGEBNISSE_agh_bezirke_{JAHR}.json.gz": { "meta": meta, "gesamt": gesamt, "bezirke": bezirke, "wahlkreise": wahlkreise, "erststimmen": erststimmen, }, f"ERGEBNISSE_agh_wahlbezirke_{JAHR}.json.gz": { "wahlbezirke": wahlbezirke, "erststimmen_wahlbezirke": erststimmen_wahlbezirke, }, f"ERGEBNISSE_agh_bwb_{JAHR}.json.gz": { "briefwahlbezirke": briefwahlbezirke, "erststimmen_briefwahlbezirke": erststimmen_briefwahlbezirke, }, f"ERGEBNISSE_bvv_bezirke_{JAHR}.json.gz": { "bvv_bezirke": bvv_bezirke, }, f"ERGEBNISSE_bvv_wahlbezirke_{JAHR}.json.gz": { "bvv_wahlbezirke": bvv_wahlbezirke, }, f"ERGEBNISSE_bvv_bwb_{JAHR}.json.gz": { "bvv_briefwahlbezirke": bvv_briefwahlbezirke, }, # ── Absolut-Varianten (Keys mit _abs-Suffix, damit Object.assign() die # Prozentwerte nicht überschreibt wenn beide Dateien in RESULTS landen) f"ERGEBNISSE_agh_bezirke_absolut_{JAHR}.json.gz": { "gesamt_abs": gesamt_abs, "bezirke_abs": bezirke_abs, "wahlkreise_abs": wahlkreise_abs, "erststimmen_abs": erststimmen_abs, }, f"ERGEBNISSE_agh_wahlbezirke_absolut_{JAHR}.json.gz": { "wahlbezirke_abs": wahlbezirke_abs, "erststimmen_wahlbezirke_abs": erststimmen_wahlbezirke_abs, }, f"ERGEBNISSE_agh_bwb_absolut_{JAHR}.json.gz": { "briefwahlbezirke_abs": briefwahlbezirke_abs, "erststimmen_briefwahlbezirke_abs": erststimmen_briefwahlbezirke_abs, }, f"ERGEBNISSE_bvv_bezirke_absolut_{JAHR}.json.gz": { "bvv_bezirke_abs": bvv_bezirke_abs, }, f"ERGEBNISSE_bvv_wahlbezirke_absolut_{JAHR}.json.gz": { "bvv_wahlbezirke_abs": bvv_wahlbezirke_abs, }, f"ERGEBNISSE_bvv_bwb_absolut_{JAHR}.json.gz": { "bvv_briefwahlbezirke_abs": bvv_briefwahlbezirke_abs, }, } JSON_DIR.mkdir(parents=True, exist_ok=True) total_gz = 0 for filename, data in files.items(): size = write_gz(JSON_DIR / filename, data) total_gz += size print(f" {filename}: {size / 1024:.1f} KB") print(f"\nGesamt komprimiert: {total_gz / 1024:.1f} KB") print(f"Gesamt-Berlin: Beteiligung {gesamt['beteiligung']} %, Briefwahl {gesamt['briefwahlanteil']} %") print(f"Bezirke: {len(bezirke)}, Wahlkreise: {len(wahlkreise)}, Wahlbezirke: {len(wahlbezirke)}") print(f"Briefwahlbezirke: {len(briefwahlbezirke)}, BVV-BWBs: {len(bvv_briefwahlbezirke)}") over100 = [k for k, v in briefwahlbezirke.items() if v["beteiligung"] >= 100] if over100: print(f"\u26a0 BWBs mit 100% (gekappt): {len(over100)}") if __name__ == "__main__": main()