From 3631e5418c9c64ff5394a9864e2bc7967356bf53 Mon Sep 17 00:00:00 2001 From: Dotty Dotter Date: Thu, 9 Apr 2026 11:25:57 +0200 Subject: [PATCH] Phase C: Auswertungen-Dashboard #58 + CSV-Export #45 (Roadmap #59) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drei-dimensionale Aggregations-Sicht über Bundesland × Partei × Wahlperiode mit minimalem Frontend. Backend (`app/auswertungen.py`): - `aggregate_matrix(filter_wp=None)` — 2D-Matrix Bundesland × Partei mit (n, Ø-Score) pro Zelle, optional gefiltert nach Wahlperiode - `aggregate_zeitreihe(bundesland, partei)` — Score-Verlauf einer (BL, Partei)-Kombination über alle bekannten WPs - `export_long_format()` — Long-Format-CSV-Export für externe Tools (deckt #45 vollständig ab) - Partei-Auflösung läuft strikt durch `normalize_partei()` aus #55 — damit wird BB-`FREIE WÄHLER` korrekt als `BVB-FW` aggregiert und NICHT mit dem RP-FW zusammengezählt Wahlperioden-Helper (`app/wahlperioden.py`): - `wahlperiode_for(datum, bundesland)` mappt ein ISO-Datum + BL auf eine Kennung wie `"NRW-WP18"` oder `"MV-WP7"` (Vorgänger-WP). Single Source of Truth ist `BUNDESLAENDER[bl].wahlperiode_start` - `all_wahlperioden()` für UI-Filter-Dropdowns Endpoints in `app/main.py`: - `GET /auswertungen` — HTML-Seite (neues Template) - `GET /api/auswertungen/matrix?wahlperiode=NRW-WP18` — JSON-Matrix - `GET /api/auswertungen/zeitreihe?bundesland=MV&partei=CDU` — JSON-Verlauf - `GET /api/auswertungen/export.csv` — CSV-Download Frontend (`app/templates/auswertungen.html`): - Statisches Template mit Vanilla-JS, kein Build-Step - Wahlperioden-Dropdown + Reload-Button + CSV-Export-Button - Matrix-Tabelle mit Score-Color-Coding (rot ≤ 3, gelb 3-6, grün > 6) - Sticky-Bundesland-Spalte für horizontales Scrolling Tests (`tests/test_auswertungen.py`): - 19 Cases mit in-memory SQLite-Fixture - Verifiziert WP-Mapping, Matrix-Aggregation, Koalitions-Counting, WP-Filter-Korrektheit, BVB-FW-Disambiguierung in der Matrix, CSV-Long-Format - 176 Unit-Tests grün (157 alt + 19 neu) Refs: #58, #45, #59 (Phase C) Co-Authored-By: Claude Opus 4.6 (1M context) --- app/auswertungen.py | 219 +++++++++++++++++++++++++++++++ app/main.py | 42 ++++++ app/templates/auswertungen.html | 201 ++++++++++++++++++++++++++++ app/wahlperioden.py | 52 ++++++++ tests/test_auswertungen.py | 225 ++++++++++++++++++++++++++++++++ 5 files changed, 739 insertions(+) create mode 100644 app/auswertungen.py create mode 100644 app/templates/auswertungen.html create mode 100644 app/wahlperioden.py create mode 100644 tests/test_auswertungen.py diff --git a/app/auswertungen.py b/app/auswertungen.py new file mode 100644 index 0000000..e48191a --- /dev/null +++ b/app/auswertungen.py @@ -0,0 +1,219 @@ +"""Aggregations-Funktionen für die Auswertungen-Seite (#58). + +Liest direkt aus ``data/gwoe-antraege.db`` (assessments-Tabelle) und baut +drei Sichten: + +1. ``aggregate_matrix(filter_wp=None)`` — 2D-Matrix Bundesland × Partei + mit (n, Ø-GWÖ-Score). Filterbar nach Wahlperiode. +2. ``aggregate_zeitreihe(bundesland, partei)`` — Score-Verlauf einer + (BL, Partei)-Kombination über alle bekannten WPs. +3. ``export_long_format()`` — Long-Format-Tabelle für CSV-Export + (deckt zusätzlich Issue #45 ab). + +Partei-Auflösung läuft strikt über ``app.parteien.normalize_partei`` — +ohne den Mapper aus #55 würde z.B. BB-FW mit RP-FW in einen Topf +gerührt. +""" +from __future__ import annotations + +import csv +import io +import json +import sqlite3 +from collections import defaultdict +from pathlib import Path +from typing import Optional + +from .config import settings +from .parteien import normalize_partei +from .wahlperioden import wahlperiode_for + + +# ───────────────────────────────────────────────────────────────────────────── +# Datenstrukturen +# ───────────────────────────────────────────────────────────────────────────── + + +def _load_assessments(db_path: Optional[Path] = None) -> list[dict]: + """Lese alle Assessments aus der SQLite-DB. Kein Filter — die + Aggregations-Funktionen filtern selbst. Kein async, weil die + Sicht synchron berechnet werden kann.""" + path = db_path or settings.db_path + if not Path(path).exists(): + return [] + conn = sqlite3.connect(str(path)) + try: + conn.row_factory = sqlite3.Row + rows = conn.execute( + """ + SELECT drucksache, bundesland, datum, fraktionen, gwoe_score + FROM assessments + WHERE gwoe_score IS NOT NULL + """ + ).fetchall() + finally: + conn.close() + + out: list[dict] = [] + for r in rows: + try: + fraktionen = json.loads(r["fraktionen"]) if r["fraktionen"] else [] + except (json.JSONDecodeError, TypeError): + fraktionen = [] + out.append({ + "drucksache": r["drucksache"], + "bundesland": r["bundesland"], + "datum": r["datum"] or "", + "fraktionen": fraktionen, + "gwoe_score": r["gwoe_score"], + }) + return out + + +# ───────────────────────────────────────────────────────────────────────────── +# 1. Matrix Bundesland × Partei +# ───────────────────────────────────────────────────────────────────────────── + + +def aggregate_matrix( + filter_wp: Optional[str] = None, + db_path: Optional[Path] = None, +) -> dict: + """Aggregate assessments to a 2D matrix. + + Returns: + ``{ + "bundeslaender": [...], + "parteien": [...], + "cells": { + "": {"": {"n": int, "avg": float}} + }, + "filter_wp": | None, + "total": int, + }`` + + ``filter_wp`` ist eine ``"-WP"``-Kennung wie ``"NRW-WP18"``; + nur Assessments dieser Wahlperiode fließen ein. ``None`` = keine + WP-Einschränkung (alle WPs zusammen). + """ + rows = _load_assessments(db_path) + + bundeslaender: set[str] = set() + parteien: set[str] = set() + sums: defaultdict[tuple[str, str], float] = defaultdict(float) + counts: defaultdict[tuple[str, str], int] = defaultdict(int) + total = 0 + + for row in rows: + bl = row["bundesland"] + if not bl: + continue + if filter_wp is not None: + wp = wahlperiode_for(row["datum"], bl) + if wp != filter_wp: + continue + bundeslaender.add(bl) + for raw_partei in row["fraktionen"]: + canonical = normalize_partei(raw_partei, bundesland=bl) or raw_partei + parteien.add(canonical) + key = (bl, canonical) + sums[key] += row["gwoe_score"] + counts[key] += 1 + total += 1 + + cells: dict[str, dict[str, dict]] = {} + for (bl, partei), s in sums.items(): + n = counts[(bl, partei)] + cells.setdefault(bl, {})[partei] = { + "n": n, + "avg": round(s / n, 2) if n else None, + } + + return { + "bundeslaender": sorted(bundeslaender), + "parteien": sorted(parteien), + "cells": cells, + "filter_wp": filter_wp, + "total": total, + } + + +# ───────────────────────────────────────────────────────────────────────────── +# 2. Zeitreihe pro (BL, Partei) über alle Wahlperioden +# ───────────────────────────────────────────────────────────────────────────── + + +def aggregate_zeitreihe( + bundesland: str, + partei: str, + db_path: Optional[Path] = None, +) -> dict: + """Score-Verlauf einer (BL, Partei)-Kombination über alle WPs. + + Returns: + ``{ + "bundesland": str, + "partei": str, + "wahlperioden": [ + {"wp": "-WP", "n": int, "avg": float}, + ... + ] + }`` + """ + rows = _load_assessments(db_path) + sums: defaultdict[str, float] = defaultdict(float) + counts: defaultdict[str, int] = defaultdict(int) + + for row in rows: + if row["bundesland"] != bundesland: + continue + canonical_partei_in_row = { + normalize_partei(p, bundesland=bundesland) or p + for p in row["fraktionen"] + } + if partei not in canonical_partei_in_row: + continue + wp = wahlperiode_for(row["datum"], bundesland) + if wp is None: + continue + sums[wp] += row["gwoe_score"] + counts[wp] += 1 + + wps = sorted(sums.keys()) + return { + "bundesland": bundesland, + "partei": partei, + "wahlperioden": [ + {"wp": wp, "n": counts[wp], "avg": round(sums[wp] / counts[wp], 2)} + for wp in wps + ], + } + + +# ───────────────────────────────────────────────────────────────────────────── +# 3. Long-Format-Export für CSV (deckt #45 mit ab) +# ───────────────────────────────────────────────────────────────────────────── + + +def export_long_format(db_path: Optional[Path] = None) -> str: + """Long-Format-CSV-Export aller Assessments für externe Auswertung. + + Spalten: ``drucksache,bundesland,wahlperiode,datum,partei,gwoe_score``. + Eine Zeile pro (drucksache, partei) — wenn ein Antrag mehrere + Fraktionen hat (Koalitionsanträge), erscheinen entsprechend mehrere + Zeilen mit identischer Drucksache. + """ + rows = _load_assessments(db_path) + buf = io.StringIO() + writer = csv.writer(buf, dialect="excel") + writer.writerow(["drucksache", "bundesland", "wahlperiode", "datum", "partei", "gwoe_score"]) + for r in rows: + bl = r["bundesland"] or "" + wp = wahlperiode_for(r["datum"], bl) if bl else "" + for raw_partei in r["fraktionen"]: + canonical = normalize_partei(raw_partei, bundesland=bl) or raw_partei + writer.writerow([ + r["drucksache"], bl, wp or "", r["datum"], canonical, + f"{r['gwoe_score']:.2f}", + ]) + return buf.getvalue() diff --git a/app/main.py b/app/main.py index c8a675e..7d5aa15 100644 --- a/app/main.py +++ b/app/main.py @@ -636,6 +636,48 @@ async def index_programme( raise HTTPException(status_code=400, detail="Ungültiges Programm") +# ───────────────────────────────────────────────────────────────────────────── +# Auswertungen #58 — Bundesland × Partei × Wahlperiode Aggregations-Sicht +# ───────────────────────────────────────────────────────────────────────────── + + +@app.get("/auswertungen", response_class=HTMLResponse) +async def auswertungen_page(request: Request): + """Statische Seite, die die Matrix-Endpoints per fetch() lädt.""" + from .wahlperioden import all_wahlperioden + return templates.TemplateResponse("auswertungen.html", { + "request": request, + "app_name": settings.app_name, + "wahlperioden": sorted(all_wahlperioden()), + }) + + +@app.get("/api/auswertungen/matrix") +async def auswertungen_matrix(wahlperiode: Optional[str] = None): + """2D-Matrix Bundesland × Partei mit Anzahl + Ø-GWÖ-Score.""" + from .auswertungen import aggregate_matrix + return aggregate_matrix(filter_wp=wahlperiode) + + +@app.get("/api/auswertungen/zeitreihe") +async def auswertungen_zeitreihe(bundesland: str, partei: str): + """Score-Verlauf einer (BL, Partei)-Kombination über alle WPs.""" + from .auswertungen import aggregate_zeitreihe + return aggregate_zeitreihe(bundesland, partei) + + +@app.get("/api/auswertungen/export.csv") +async def auswertungen_export_csv(): + """Long-Format-CSV-Export aller Assessments. Deckt #45 mit ab.""" + from .auswertungen import export_long_format + csv_text = export_long_format() + return Response( + content=csv_text, + media_type="text/csv", + headers={"Content-Disposition": 'attachment; filename="gwoe-assessments.csv"'}, + ) + + # Health check @app.get("/health") async def health(): diff --git a/app/templates/auswertungen.html b/app/templates/auswertungen.html new file mode 100644 index 0000000..28a912b --- /dev/null +++ b/app/templates/auswertungen.html @@ -0,0 +1,201 @@ + + + + + + Auswertungen — {{ app_name }} + + + +
+

Auswertungen — Bundesland × Partei × Wahlperiode

+ +
+ +
+
+ + + +
+ +
+
Lade Matrix …
+
+ +
+
+ + + + diff --git a/app/wahlperioden.py b/app/wahlperioden.py new file mode 100644 index 0000000..84c7574 --- /dev/null +++ b/app/wahlperioden.py @@ -0,0 +1,52 @@ +"""Wahlperioden-Helper für die Auswertungen. + +Maps a Drucksache-Datum + Bundesland auf eine Wahlperioden-Kennung. +Single Source of Truth ist ``BUNDESLAENDER[bl].wahlperiode_start`` — +alles vor diesem Datum gehört zur Vorgänger-WP, alles ab dem Datum zur +aktuellen. + +Granularität: pro Bundesland kennen wir genau zwei WP — die laufende und +die direkt davor. Das reicht für die Aggregations-Sicht (#58), denn der +gesamte Antragsbestand stammt aus den letzten 5–10 Jahren. Sollte später +eine echte Multi-WP-Historie nötig werden, ist der Erweiterungspunkt +``_wp_calendar`` als Liste von ``(start, ende, wp_id)``-Tupeln pro BL. +""" +from __future__ import annotations + +from typing import Optional + +from .bundeslaender import BUNDESLAENDER + + +def wahlperiode_for(datum: str, bundesland: str) -> Optional[str]: + """Liefert die Wahlperioden-Kennung für ein Datum + Bundesland. + + Format der Kennung: ``"-WP"``, z.B. ``"NRW-WP18"``, + ``"MV-WP7"`` (Vorgänger-WP). Returns ``None`` wenn das Bundesland + unbekannt oder das Datum nicht parsbar ist. + + >>> wahlperiode_for("2026-03-18", "MV") + 'MV-WP8' + >>> wahlperiode_for("2020-01-01", "MV") + 'MV-WP7' + """ + bl = BUNDESLAENDER.get(bundesland) + if bl is None: + return None + if not datum: + return f"{bundesland}-WP{bl.wahlperiode}" # default: aktuelle WP + # ISO-Datum ist lexikographisch vergleichbar — keine datetime-Parsing + # nötig, solange das Format YYYY-MM-DD eingehalten wird. + if datum >= bl.wahlperiode_start: + return f"{bundesland}-WP{bl.wahlperiode}" + return f"{bundesland}-WP{bl.wahlperiode - 1}" + + +def all_wahlperioden() -> list[str]: + """Liste aller bekannten Wahlperioden-Kennungen (aktuelle + Vorgänger + pro Bundesland). Nützlich für UI-Filter-Dropdowns.""" + out: list[str] = [] + for code, bl in BUNDESLAENDER.items(): + out.append(f"{code}-WP{bl.wahlperiode}") + out.append(f"{code}-WP{bl.wahlperiode - 1}") + return out diff --git a/tests/test_auswertungen.py b/tests/test_auswertungen.py new file mode 100644 index 0000000..cc8f10c --- /dev/null +++ b/tests/test_auswertungen.py @@ -0,0 +1,225 @@ +"""Tests für app.wahlperioden und app.auswertungen. + +Issue #58 + Roadmap #59 Phase C. Verifiziert die Aggregations-Logik +gegen eine in-memory SQLite-DB mit kontrollierten Sample-Assessments. +""" +from __future__ import annotations + +import json +import sqlite3 +from datetime import datetime +from pathlib import Path + +import pytest + +from app.auswertungen import ( + aggregate_matrix, + aggregate_zeitreihe, + export_long_format, +) +from app.wahlperioden import all_wahlperioden, wahlperiode_for + + +# ───────────────────────────────────────────────────────────────────────────── +# wahlperioden helper +# ───────────────────────────────────────────────────────────────────────────── + + +class TestWahlperiodeFor: + def test_current_wp_for_recent_date(self): + assert wahlperiode_for("2026-03-18", "MV") == "MV-WP8" + + def test_previous_wp_for_old_date(self): + # MV WP8 startete am 26.10.2021 — alles davor ist WP7 + assert wahlperiode_for("2020-01-01", "MV") == "MV-WP7" + + def test_unknown_bl_returns_none(self): + assert wahlperiode_for("2026-01-01", "XX") is None + + def test_empty_datum_returns_current_wp(self): + # Wenn kein Datum bekannt → wir nehmen die aktuelle WP an, + # weil das die einzig sinnvolle Default-Annahme ist + assert wahlperiode_for("", "NRW") == "NRW-WP18" + + def test_all_wahlperioden_lists_each_bl_twice(self): + out = all_wahlperioden() + # 16 Bundesländer × 2 WPs = 32 Einträge + assert len(out) == 32 + # Aktuelle und vorherige WP für NRW + assert "NRW-WP18" in out + assert "NRW-WP17" in out + + +# ───────────────────────────────────────────────────────────────────────────── +# Test-DB-Fixture +# ───────────────────────────────────────────────────────────────────────────── + + +@pytest.fixture +def sample_db(tmp_path: Path) -> Path: + """Lege eine Mini-Assessments-DB an, die typische Fälle abdeckt.""" + db = tmp_path / "test_assessments.db" + conn = sqlite3.connect(str(db)) + conn.execute(""" + CREATE TABLE assessments ( + drucksache TEXT PRIMARY KEY, + title TEXT, + fraktionen TEXT, + datum TEXT, + bundesland TEXT, + gwoe_score REAL, + link TEXT, + gwoe_begruendung TEXT, + gwoe_matrix TEXT, + gwoe_schwerpunkt TEXT, + wahlprogramm_scores TEXT, + verbesserungen TEXT, + staerken TEXT, + schwaechen TEXT, + empfehlung TEXT, + empfehlung_symbol TEXT, + verbesserungspotenzial TEXT, + themen TEXT, + antrag_zusammenfassung TEXT, + antrag_kernpunkte TEXT, + source TEXT, + model TEXT, + created_at TEXT, + updated_at TEXT + ) + """) + samples = [ + # NRW WP18 — drei Anträge, zwei Parteien + ("18/100", "NRW", "2024-01-15", '["CDU"]', 7.0), + ("18/101", "NRW", "2024-02-15", '["SPD"]', 8.0), + ("18/102", "NRW", "2024-03-15", '["CDU"]', 5.0), + # MV WP8 — Koalitionsantrag (zwei Parteien zählen beide) + ("8/200", "MV", "2024-04-01", '["SPD","LINKE"]', 6.0), + ("8/201", "MV", "2025-01-10", '["AfD"]', 2.0), + # MV WP7 — historischer Antrag vor wahlperiode_start (2021-10-26) + ("7/100", "MV", "2020-05-01", '["CDU"]', 4.0), + # BB — FREIE WÄHLER soll als BVB-FW kanonisiert werden + ("8/2", "BB", "2024-10-17", '["FREIE WÄHLER"]', 6.5), + ] + now = datetime.utcnow().isoformat() + for ds, bl, dat, fr, sc in samples: + conn.execute( + "INSERT INTO assessments (drucksache, title, fraktionen, datum, bundesland, " + "gwoe_score, source, model, created_at, updated_at) VALUES " + "(?, ?, ?, ?, ?, ?, 'test', 'test', ?, ?)", + (ds, f"Test {ds}", fr, dat, bl, sc, now, now), + ) + conn.commit() + conn.close() + return db + + +# ───────────────────────────────────────────────────────────────────────────── +# aggregate_matrix +# ───────────────────────────────────────────────────────────────────────────── + + +class TestAggregateMatrix: + def test_total_count(self, sample_db): + m = aggregate_matrix(db_path=sample_db) + assert m["total"] == 7 + + def test_bundeslaender_listed(self, sample_db): + m = aggregate_matrix(db_path=sample_db) + assert set(m["bundeslaender"]) == {"NRW", "MV", "BB"} + + def test_nrw_cdu_average(self, sample_db): + # NRW-CDU: 7.0 + 5.0 → Avg 6.0, n=2 + m = aggregate_matrix(db_path=sample_db) + cell = m["cells"]["NRW"]["CDU"] + assert cell["n"] == 2 + assert cell["avg"] == 6.0 + + def test_koalition_counts_both_parties(self, sample_db): + # MV-SPD und MV-LINKE bekommen beide den Score 6.0 (n=1) + m = aggregate_matrix(db_path=sample_db) + assert m["cells"]["MV"]["SPD"]["n"] == 1 + assert m["cells"]["MV"]["LINKE"]["n"] == 1 + assert m["cells"]["MV"]["SPD"]["avg"] == 6.0 + + def test_filter_by_wahlperiode(self, sample_db): + # NRW-WP18-Filter → nur die 3 NRW-Anträge + m = aggregate_matrix(filter_wp="NRW-WP18", db_path=sample_db) + assert m["total"] == 3 + assert set(m["bundeslaender"]) == {"NRW"} + + def test_filter_excludes_old_wp(self, sample_db): + # MV-WP8 darf den 7/100-Antrag (datum=2020) NICHT enthalten + m = aggregate_matrix(filter_wp="MV-WP8", db_path=sample_db) + assert m["total"] == 2 # nur 8/200 und 8/201 + # CDU darf NICHT vorkommen, weil der CDU-Antrag in WP7 war + assert "CDU" not in m["cells"].get("MV", {}) + + def test_bb_freie_waehler_normalized_to_bvb(self, sample_db): + # Die BB-FW-Drucksache muss als BVB-FW gezählt werden, NICHT als + # generisches FREIE WÄHLER — das ist der eigentliche Mehrwert + # des Parteinamen-Mappers (#55) + m = aggregate_matrix(db_path=sample_db) + bb_cells = m["cells"]["BB"] + assert "BVB-FW" in bb_cells + assert bb_cells["BVB-FW"]["n"] == 1 + assert "FREIE WÄHLER" not in bb_cells + + def test_empty_db_returns_empty_matrix(self, tmp_path): + m = aggregate_matrix(db_path=tmp_path / "missing.db") + assert m["total"] == 0 + assert m["bundeslaender"] == [] + + +# ───────────────────────────────────────────────────────────────────────────── +# aggregate_zeitreihe +# ───────────────────────────────────────────────────────────────────────────── + + +class TestAggregateZeitreihe: + def test_mv_cdu_two_wps(self, sample_db): + # MV-CDU hat einen Eintrag in WP7 (4.0) und keinen in WP8 + z = aggregate_zeitreihe("MV", "CDU", db_path=sample_db) + wps = {entry["wp"]: entry for entry in z["wahlperioden"]} + assert "MV-WP7" in wps + assert wps["MV-WP7"]["avg"] == 4.0 + assert wps["MV-WP7"]["n"] == 1 + + def test_nrw_cdu_one_wp(self, sample_db): + z = aggregate_zeitreihe("NRW", "CDU", db_path=sample_db) + assert len(z["wahlperioden"]) == 1 + assert z["wahlperioden"][0]["avg"] == 6.0 + + def test_unknown_combination_empty(self, sample_db): + z = aggregate_zeitreihe("NRW", "AfD", db_path=sample_db) + assert z["wahlperioden"] == [] + + +# ───────────────────────────────────────────────────────────────────────────── +# export_long_format +# ───────────────────────────────────────────────────────────────────────────── + + +class TestExportLongFormat: + def test_csv_has_header(self, sample_db): + csv_text = export_long_format(db_path=sample_db) + first_line = csv_text.splitlines()[0] + assert "drucksache" in first_line + assert "bundesland" in first_line + assert "wahlperiode" in first_line + assert "partei" in first_line + assert "gwoe_score" in first_line + + def test_koalition_yields_two_rows(self, sample_db): + csv_text = export_long_format(db_path=sample_db) + lines = csv_text.splitlines()[1:] # ohne Header + # 8/200 ist Koalitionsantrag (SPD+LINKE) → 2 Zeilen + mv_8_200_lines = [l for l in lines if l.startswith("8/200,")] + assert len(mv_8_200_lines) == 2 + + def test_bb_fw_normalized_in_csv(self, sample_db): + csv_text = export_long_format(db_path=sample_db) + assert "BVB-FW" in csv_text + # Generic FREIE WÄHLER darf in der Zeile NICHT auftauchen + bb_lines = [l for l in csv_text.splitlines() if "BB" in l and "8/2," in l] + assert any("BVB-FW" in l for l in bb_lines)