diff --git a/app/auswertungen.py b/app/auswertungen.py index 68035a5..6643ad6 100644 --- a/app/auswertungen.py +++ b/app/auswertungen.py @@ -242,3 +242,416 @@ def export_long_format(db_path: Optional[Path] = None) -> str: f"{r['gwoe_score']:.2f}", ]) return buf.getvalue() + + +# ───────────────────────────────────────────────────────────────────────────── +# 4. Stimmverhalten × Gemeinwohl-Orientierung (#106 + #145 Folge) +# +# JOIN ueber assessments.gwoe_score × plenum_vote_results.fraktionen_ja|_nein| +# _enthaltung. Vier Sichten: +# - aggregate_stimm_index: pro Fraktion JA-Mean minus NEIN-Mean von gwoe_score +# - aggregate_heuchelei: pro Fraktion % der Antraege mit +# wahlprogramm_score>=7 wo Vote=NEIN +# - aggregate_stimm_index_pro_wert: stimm_index pro (Fraktion, GWOe-Wert) +# aus gwoe_matrix-Spalten +# - aggregate_stimm_index_cross_bl: stimm_index pro (Fraktion, BL) +# +# Sparse-Data-Realitaet: 35 Assessments × 7000 Votes → meiste Anträge mit Vote +# haben kein Assessment. min_n-Cutoff filtert Fraktionen mit zu wenig Daten. +# ───────────────────────────────────────────────────────────────────────────── + + +# Spalten der GWOe-Matrix: {field-suffix → Wert-Name}. Die Field-IDs in der DB +# sind ``A1..E5`` mit Reihe = Beruehrungsgruppe (A-E), Spalte = Wert (1-5). +GWOE_WERTE = { + "1": "Menschenwürde", + "2": "Solidarität", + "3": "Ökologische Nachhaltigkeit", + "4": "Soziale Gerechtigkeit", + "5": "Transparenz & Demokratie", +} + + +def _load_assessments_with_votes( + filter_bl: Optional[str] = None, + filter_wp: Optional[str] = None, + db_path: Optional[Path] = None, +) -> list[dict]: + """JOIN assessments × plenum_vote_results auf (bundesland, drucksache). + + Liefert pro Match-Zeile alle relevanten Felder fuer die Stimmverhalten- + Aggregationen — nur Assessments mit gwoe_score und Vote-Result. + Mehrfach-Votes pro Drucksache (Compound-PK ueber quelle_protokoll) + erzeugen entsprechend mehrere Zeilen. + """ + 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 a.drucksache, a.bundesland, a.datum, + a.fraktionen, a.gwoe_score, a.gwoe_matrix, + a.gwoe_schwerpunkt, a.wahlprogramm_scores, + p.fraktionen_ja, p.fraktionen_nein, p.fraktionen_enthaltung, + p.quelle_protokoll + FROM assessments a + INNER JOIN plenum_vote_results p + ON a.bundesland = p.bundesland + AND a.drucksache = p.drucksache + WHERE a.gwoe_score IS NOT NULL + """ + ).fetchall() + finally: + conn.close() + + out: list[dict] = [] + for r in rows: + bl = r["bundesland"] or "" + if filter_bl is not None and bl != filter_bl: + continue + if filter_wp is not None: + wp = wahlperiode_for(r["datum"], bl) + if wp != filter_wp: + continue + + def _parse_json_list(raw): + try: + return json.loads(raw) if raw else [] + except (json.JSONDecodeError, TypeError): + return [] + + def _norm_set(raw_list): + """Normalisiere eine Fraktionsliste auf kanonische Namen.""" + return { + normalize_partei(p, bundesland=bl) or p + for p in raw_list if p + } + + antragsteller = _norm_set(_parse_json_list(r["fraktionen"])) + ja = _norm_set(_parse_json_list(r["fraktionen_ja"])) + nein = _norm_set(_parse_json_list(r["fraktionen_nein"])) + enth = _norm_set(_parse_json_list(r["fraktionen_enthaltung"])) + + out.append({ + "drucksache": r["drucksache"], + "bundesland": bl, + "datum": r["datum"] or "", + "gwoe_score": r["gwoe_score"], + "gwoe_matrix": _parse_json_list(r["gwoe_matrix"]), + "gwoe_schwerpunkt": _parse_json_list(r["gwoe_schwerpunkt"]), + "wahlprogramm_scores": _parse_json_list(r["wahlprogramm_scores"]), + "antragsteller": antragsteller, + "ja": ja, + "nein": nein, + "enthaltung": enth, + "quelle_protokoll": r["quelle_protokoll"], + }) + return out + + +def _avg(values: list[float]) -> Optional[float]: + return round(sum(values) / len(values), 2) if values else None + + +def aggregate_stimm_index( + filter_bl: Optional[str] = None, + filter_wp: Optional[str] = None, + exclude_antragsteller: bool = True, + min_n: int = 5, + db_path: Optional[Path] = None, +) -> dict: + """Pro Fraktion: Ø-GWÖ-Score der JA-Antraege minus Ø-GWÖ-Score der + NEIN-Antraege. Antragsteller-Bias optional rausgerechnet. + + ``stimm_index`` ist None wenn n_ja oder n_nein = 0; ``ausreichend`` + ist True wenn n_ja >= min_n und n_nein >= min_n (Domain-Heuristik: + Aussage erst belastbar bei beidseitigem Mindest-N). + """ + rows = _load_assessments_with_votes(filter_bl, filter_wp, db_path) + + ja_scores: defaultdict[str, list[float]] = defaultdict(list) + nein_scores: defaultdict[str, list[float]] = defaultdict(list) + enth_scores: defaultdict[str, list[float]] = defaultdict(list) + + for row in rows: + score = row["gwoe_score"] + skip = row["antragsteller"] if exclude_antragsteller else set() + for f in row["ja"] - skip: + ja_scores[f].append(score) + for f in row["nein"] - skip: + nein_scores[f].append(score) + for f in row["enthaltung"] - skip: + enth_scores[f].append(score) + + parteien = sorted(set(ja_scores) | set(nein_scores) | set(enth_scores)) + fraktionen_out = [] + for p in parteien: + n_ja = len(ja_scores[p]) + n_nein = len(nein_scores[p]) + n_enth = len(enth_scores[p]) + avg_ja = _avg(ja_scores[p]) + avg_nein = _avg(nein_scores[p]) + idx = (round(avg_ja - avg_nein, 2) + if avg_ja is not None and avg_nein is not None else None) + fraktionen_out.append({ + "partei": p, + "n_ja": n_ja, + "n_nein": n_nein, + "n_enth": n_enth, + "avg_gwoe_ja": avg_ja, + "avg_gwoe_nein": avg_nein, + "avg_gwoe_enth": _avg(enth_scores[p]), + "stimm_index": idx, + "ausreichend": n_ja >= min_n and n_nein >= min_n, + }) + + fraktionen_out.sort( + key=lambda f: (f["stimm_index"] if f["stimm_index"] is not None else -999), + reverse=True, + ) + return { + "fraktionen": fraktionen_out, + "n_assessments_matched": len({r["drucksache"] for r in rows}), + "n_votes_matched": len(rows), + "filter": { + "bundesland": filter_bl, + "wahlperiode": filter_wp, + "exclude_antragsteller": exclude_antragsteller, + "min_n": min_n, + }, + } + + +def aggregate_heuchelei( + filter_bl: Optional[str] = None, + filter_wp: Optional[str] = None, + score_threshold: float = 7.0, + min_n: int = 5, + db_path: Optional[Path] = None, +) -> dict: + """Pro Fraktion: Anteil der Antraege mit wahlprogramm_score >= 7 + (Antrag passt zum Wahlprogramm), bei denen die Fraktion trotzdem + NEIN gestimmt hat. + + Misst Inkonsistenz zwischen Wahlversprechen und Stimmverhalten. + """ + rows = _load_assessments_with_votes(filter_bl, filter_wp, db_path) + + n_passt: defaultdict[str, int] = defaultdict(int) + n_passt_nein: defaultdict[str, int] = defaultdict(int) + n_passt_ja: defaultdict[str, int] = defaultdict(int) + n_passt_enth: defaultdict[str, int] = defaultdict(int) + + for row in rows: + wp_scores = row["wahlprogramm_scores"] or [] + for entry in wp_scores: + if not isinstance(entry, dict): + continue + raw_partei = entry.get("fraktion") or "" + partei = normalize_partei(raw_partei, bundesland=row["bundesland"]) or raw_partei + if not partei: + continue + wp_block = entry.get("wahlprogramm") or {} + score = wp_block.get("score") + if score is None or score < score_threshold: + continue + n_passt[partei] += 1 + if partei in row["nein"]: + n_passt_nein[partei] += 1 + elif partei in row["ja"]: + n_passt_ja[partei] += 1 + elif partei in row["enthaltung"]: + n_passt_enth[partei] += 1 + + fraktionen_out = [] + for partei in sorted(n_passt): + total = n_passt[partei] + nein = n_passt_nein[partei] + ja = n_passt_ja[partei] + enth = n_passt_enth[partei] + quote = round(nein / total, 3) if total else None + fraktionen_out.append({ + "partei": partei, + "n_im_programm": total, + "n_nein_trotz_programm": nein, + "n_ja_passt": ja, + "n_enth_passt": enth, + "heuchelei_quote": quote, + "ausreichend": total >= min_n, + }) + fraktionen_out.sort( + key=lambda f: (f["heuchelei_quote"] or 0), reverse=True, + ) + return { + "fraktionen": fraktionen_out, + "n_assessments_matched": len({r["drucksache"] for r in rows}), + "filter": { + "bundesland": filter_bl, + "wahlperiode": filter_wp, + "score_threshold": score_threshold, + "min_n": min_n, + }, + } + + +def _wert_score_for_assessment(matrix: list[dict]) -> dict[str, float]: + """Aus der gwoe_matrix-Liste den Mittelwert pro Wert-Spalte (1..5) + berechnen. + + matrix-Eintraege haben ``field`` wie ``A1, B3, E5`` und ``rating`` -5..+5. + Pro Spalten-Suffix wird Ø-rating berechnet — fehlende Felder werden + ignoriert (LLM lässt nicht-relevante Zellen oft weg). + """ + by_col: defaultdict[str, list[float]] = defaultdict(list) + for entry in matrix or []: + if not isinstance(entry, dict): + continue + field = entry.get("field") or "" + rating = entry.get("rating") + if not field or rating is None: + continue + if len(field) >= 2 and field[-1] in GWOE_WERTE: + by_col[field[-1]].append(float(rating)) + return {col: sum(vals) / len(vals) for col, vals in by_col.items()} + + +def aggregate_stimm_index_pro_wert( + filter_bl: Optional[str] = None, + filter_wp: Optional[str] = None, + exclude_antragsteller: bool = True, + min_n: int = 5, + db_path: Optional[Path] = None, +) -> dict: + """Pro (Fraktion, GWÖ-Wert) ein Stimm-Index-Analog: Ø-Wert-Score der + JA-Antraege minus Ø-Wert-Score der NEIN-Antraege. + + Wert-Score eines Antrags = Ø(rating der gwoe_matrix-Felder mit dem + entsprechenden Spalten-Suffix). Domain: -5..+5 pro Wert. + """ + rows = _load_assessments_with_votes(filter_bl, filter_wp, db_path) + + werte_namen = list(GWOE_WERTE.values()) # in Reihenfolge 1..5 + werte_keys = list(GWOE_WERTE.keys()) + + # Pro (partei, wert_key) → list[wert_score] fuer JA / NEIN + ja: defaultdict[tuple[str, str], list[float]] = defaultdict(list) + nein: defaultdict[tuple[str, str], list[float]] = defaultdict(list) + + parteien_seen: set[str] = set() + + for row in rows: + wert_scores = _wert_score_for_assessment(row["gwoe_matrix"]) + if not wert_scores: + continue + skip = row["antragsteller"] if exclude_antragsteller else set() + for f in row["ja"] - skip: + parteien_seen.add(f) + for col, sc in wert_scores.items(): + ja[(f, col)].append(sc) + for f in row["nein"] - skip: + parteien_seen.add(f) + for col, sc in wert_scores.items(): + nein[(f, col)].append(sc) + + parteien = sorted(parteien_seen) + cells: dict[str, dict[str, dict]] = {} + for p in parteien: + cells[p] = {} + for col, wert_name in zip(werte_keys, werte_namen): + n_ja = len(ja[(p, col)]) + n_nein = len(nein[(p, col)]) + avg_ja = _avg(ja[(p, col)]) + avg_nein = _avg(nein[(p, col)]) + idx = (round(avg_ja - avg_nein, 2) + if avg_ja is not None and avg_nein is not None else None) + cells[p][wert_name] = { + "stimm_index": idx, + "n_ja": n_ja, + "n_nein": n_nein, + "ausreichend": n_ja >= min_n and n_nein >= min_n, + } + + return { + "fraktionen": parteien, + "werte": werte_namen, + "cells": cells, + "n_assessments_matched": len({r["drucksache"] for r in rows}), + "filter": { + "bundesland": filter_bl, + "wahlperiode": filter_wp, + "exclude_antragsteller": exclude_antragsteller, + "min_n": min_n, + }, + } + + +def aggregate_stimm_index_cross_bl( + filter_wp: Optional[str] = None, + exclude_antragsteller: bool = True, + min_n: int = 5, + db_path: Optional[Path] = None, +) -> dict: + """Stimm-Index pro (Fraktion, Bundesland) — macht regionale Drift + sichtbar fuer bundesweit aktive Fraktionen (CDU, SPD, GRÜNE, AfD, + FDP, LINKE, BSW).""" + rows = _load_assessments_with_votes(None, filter_wp, db_path) + + ja: defaultdict[tuple[str, str], list[float]] = defaultdict(list) + nein: defaultdict[tuple[str, str], list[float]] = defaultdict(list) + bl_seen: set[str] = set() + parteien_seen: set[str] = set() + + for row in rows: + bl = row["bundesland"] + if not bl: + continue + bl_seen.add(bl) + score = row["gwoe_score"] + skip = row["antragsteller"] if exclude_antragsteller else set() + for f in row["ja"] - skip: + parteien_seen.add(f) + ja[(f, bl)].append(score) + for f in row["nein"] - skip: + parteien_seen.add(f) + nein[(f, bl)].append(score) + + bundeslaender = sorted(bl_seen) + parteien = sorted(parteien_seen) + cells: dict[str, dict[str, dict]] = {} + for p in parteien: + cells[p] = {} + for bl in bundeslaender: + n_ja = len(ja[(p, bl)]) + n_nein = len(nein[(p, bl)]) + avg_ja = _avg(ja[(p, bl)]) + avg_nein = _avg(nein[(p, bl)]) + idx = (round(avg_ja - avg_nein, 2) + if avg_ja is not None and avg_nein is not None else None) + cells[p][bl] = { + "stimm_index": idx, + "n_ja": n_ja, + "n_nein": n_nein, + "ausreichend": n_ja >= min_n and n_nein >= min_n, + } + + # Filter Fraktionen mit nur 1 BL → in Phase 1 nicht aufschlussreich + parteien_multi_bl = [ + p for p in parteien + if sum(1 for bl in bundeslaender if cells[p][bl]["ausreichend"]) >= 2 + ] + + return { + "fraktionen": parteien_multi_bl, + "fraktionen_alle": parteien, + "bundeslaender": bundeslaender, + "cells": cells, + "n_assessments_matched": len({r["drucksache"] for r in rows}), + "filter": { + "wahlperiode": filter_wp, + "exclude_antragsteller": exclude_antragsteller, + "min_n": min_n, + }, + } diff --git a/app/main.py b/app/main.py index c731c20..64bfa0c 100644 --- a/app/main.py +++ b/app/main.py @@ -2292,6 +2292,84 @@ async def auswertungen_export_csv(): ) +# ─── Stimmverhalten × Gemeinwohl-Orientierung (Issue: Auswertungen-Erweiterung) ─ + +@app.get("/api/auswertungen/stimm-index") +async def auswertungen_stimm_index( + bundesland: Optional[str] = None, + wahlperiode: Optional[str] = None, + exclude_antragsteller: bool = True, + min_n: int = 5, +): + """Pro Fraktion Ø-GWÖ der JA-Anträge MINUS Ø-GWÖ der NEIN-Anträge. + + Antwort auf "Welche Fraktionen stimmen häufiger gemeinwohl- + orientierten Anträgen zu?". Antragsteller-Bias per Default raus. + """ + from .auswertungen import aggregate_stimm_index + return aggregate_stimm_index( + filter_bl=bundesland, + filter_wp=wahlperiode, + exclude_antragsteller=exclude_antragsteller, + min_n=min_n, + ) + + +@app.get("/api/auswertungen/heuchelei") +async def auswertungen_heuchelei( + bundesland: Optional[str] = None, + wahlperiode: Optional[str] = None, + score_threshold: float = 7.0, + min_n: int = 5, +): + """Pro Fraktion: Anteil der Anträge mit wahlprogramm_score>=threshold, + bei denen die Fraktion trotzdem NEIN gestimmt hat. + + Macht sichtbar, wer gegen die eigenen Wahlversprechen stimmt. + """ + from .auswertungen import aggregate_heuchelei + return aggregate_heuchelei( + filter_bl=bundesland, + filter_wp=wahlperiode, + score_threshold=score_threshold, + min_n=min_n, + ) + + +@app.get("/api/auswertungen/stimm-index-pro-wert") +async def auswertungen_stimm_index_pro_wert( + bundesland: Optional[str] = None, + wahlperiode: Optional[str] = None, + exclude_antragsteller: bool = True, + min_n: int = 5, +): + """Stimm-Index aufgeschlüsselt pro GWÖ-Wert (5 Spalten: + Würde/Solidarität/Nachhaltigkeit/Gerechtigkeit/Demokratie).""" + from .auswertungen import aggregate_stimm_index_pro_wert + return aggregate_stimm_index_pro_wert( + filter_bl=bundesland, + filter_wp=wahlperiode, + exclude_antragsteller=exclude_antragsteller, + min_n=min_n, + ) + + +@app.get("/api/auswertungen/stimm-index-cross-bl") +async def auswertungen_stimm_index_cross_bl( + wahlperiode: Optional[str] = None, + exclude_antragsteller: bool = True, + min_n: int = 5, +): + """Stimm-Index pro (Fraktion, Bundesland) — regionale Drift sichtbar + fuer bundesweit aktive Fraktionen.""" + from .auswertungen import aggregate_stimm_index_cross_bl + return aggregate_stimm_index_cross_bl( + filter_wp=wahlperiode, + exclude_antragsteller=exclude_antragsteller, + min_n=min_n, + ) + + # ─── v2 Frontend (#139 Phase 2 + Phase 3) ─────────────────────────────────── # / ist jetzt Default-v2. /v2 leitet auf / weiter; /v2/antrag/* auf /antrag/*. # /classic ist die alte Ansicht (index.html unverändert). diff --git a/app/templates/v2/screens/auswertungen.html b/app/templates/v2/screens/auswertungen.html index 3bf90c0..6a1dfe2 100644 --- a/app/templates/v2/screens/auswertungen.html +++ b/app/templates/v2/screens/auswertungen.html @@ -160,7 +160,7 @@ table.modal-table th { background: var(--ecg-bg-subtle); font-weight: 700; }

Auswertungen

- Bundesland × Partei · Thema × Fraktion · Cluster + Bundesland × Partei · Thema × Fraktion · Stimmverhalten · Cluster

@@ -168,6 +168,7 @@ table.modal-table th { background: var(--ecg-bg-subtle); font-weight: 700; }
+
@@ -198,7 +199,87 @@ table.modal-table th { background: var(--ecg-bg-subtle); font-weight: 700; } - + +
+
+

Stimmverhalten × Gemeinwohl-Orientierung

+

+ Verschneidung von GWÖ-Bewertung pro Antrag mit dem + tatsächlichen Plenum-Stimmverhalten der Fraktionen. + Beantwortet die Frage: Welche Fraktion stimmt häufiger Anträgen mit + hoher Gemeinwohl-Bewertung zu, welche lehnt sie eher ab? +

+

+ Datenbasis: Anträge, die sowohl eine GWÖ-Bewertung + als auch einen Plenum-Vote haben. Heute noch dünn — wächst + mit der Anzahl Bewertungen. + Eigene Anträge sind per Default ausgeschlossen, weil + Antragsteller-Fraktionen quasi immer „ja" stimmen — das würde den Index + verzerren. +

+ +
+ + +

1. Gemeinwohl-Stimm-Index pro Fraktion

+

+ Ø-GWÖ-Score der Ja-Stimmen minus Ø-GWÖ-Score der + Nein-Stimmen. Positiv = stimmt eher Gemeinwohl-affinen Anträgen zu. + Domain: −10..+10. +

+
+ +
+
+
+ + +

2. Heuchelei-Quote pro Fraktion

+

+ Anteil der Anträge mit Wahlprogramm-Treue ≥ 7/10 + (passt inhaltlich zum Wahlprogramm der Fraktion), bei denen die Fraktion + trotzdem Nein gestimmt hat. Hohe Werte = häufige Inkonsistenz + zwischen Wahlversprechen und Stimmverhalten. +

+
+ +
+
+ + +

3. Stimm-Index pro GWÖ-Wert

+

+ Aufschlüsselung nach den fünf GWÖ-Werten (Würde, Solidarität, + Nachhaltigkeit, Gerechtigkeit, Demokratie). Pro Zelle: Stimm-Index analog + Aussage 1, aber gegen den Wert-Score statt den Gesamt-Score. Domain + pro Zelle: −5..+5. +

+
+ + +

4. Stimm-Index pro Bundesland (Cross-BL)

+

+ Gleiche Fraktion in mehreren Ländern. Macht regionale Drift sichtbar + — gleiche Partei, unterschiedlicher Stimm-Index? Nur Fraktionen in ≥2 BL + mit ausreichender Datenbasis. +

+
+ +
+
+
+ +

Cluster-Ansicht

@@ -229,7 +310,8 @@ table.modal-table th { background: var(--ecg-bg-subtle); font-weight: 700; } {% block body_scripts %} diff --git a/tests/test_auswertungen_stimmverhalten.py b/tests/test_auswertungen_stimmverhalten.py new file mode 100644 index 0000000..34c32c7 --- /dev/null +++ b/tests/test_auswertungen_stimmverhalten.py @@ -0,0 +1,401 @@ +"""Tests fuer app.auswertungen.aggregate_stimm_index/heuchelei/pro_wert/cross_bl. + +Verifiziert die JOIN-Aggregations-Logik gegen eine in-memory SQLite-DB +mit kontrollierten Sample-Assessments und Sample-Vote-Results. +""" +from __future__ import annotations + +import json +import sqlite3 +from datetime import datetime +from pathlib import Path + +import pytest + +from app.auswertungen import ( + aggregate_heuchelei, + aggregate_stimm_index, + aggregate_stimm_index_cross_bl, + aggregate_stimm_index_pro_wert, + _wert_score_for_assessment, +) + + +# ───────────────────────────────────────────────────────────────────────────── +# Fixture: assessments + plenum_vote_results +# ───────────────────────────────────────────────────────────────────────────── + + +def _matrix(*ratings: tuple[str, int]) -> str: + """Build a gwoe_matrix JSON-Array. Args: (field, rating) tuples.""" + return json.dumps([ + {"field": f, "label": f, "aspect": f[-1], "rating": r, + "symbol": "++" if r >= 4 else ("+" if r >= 1 else "○")} + for f, r in ratings + ]) + + +def _wp_scores(*items: tuple[str, int, bool]) -> str: + """Build a wahlprogramm_scores JSON-Array. Args: (fraktion, score, is_antrag).""" + return json.dumps([ + { + "fraktion": fr, + "ist_antragsteller": ist, + "wahlprogramm": {"score": sc, "begründung": "stub"}, + } + for fr, sc, ist in items + ]) + + +@pytest.fixture +def sample_db(tmp_path: Path) -> Path: + """Lege eine Mini-DB mit assessments + plenum_vote_results an, die typische + Faelle abdeckt: + + - NRW Antrag mit hohem GWOe-Score, GRÜNE/SPD ja, CDU/AfD nein + - NRW Antrag mit niedrigem GWOe-Score, AfD ja, GRÜNE nein (Anti-Pattern) + - NRW Antrag mit hohem Score, CDU Antragsteller (sollte bei Default + excluded werden) + - MV Anträge fuer Cross-BL-Test + - Heuchelei-Sample: GRÜNE-Antrag mit GRÜNE-wahlprogramm_score=9 aber + GRÜNE stimmt NEIN (synthetisch konstruiert) + """ + db = tmp_path / "test_stimmverhalten.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 + ) + """) + conn.execute(""" + CREATE TABLE plenum_vote_results ( + bundesland TEXT NOT NULL, + drucksache TEXT NOT NULL, + ergebnis TEXT NOT NULL, + einstimmig INTEGER NOT NULL DEFAULT 0, + fraktionen_ja TEXT NOT NULL DEFAULT '[]', + fraktionen_nein TEXT NOT NULL DEFAULT '[]', + fraktionen_enthaltung TEXT NOT NULL DEFAULT '[]', + quelle_protokoll TEXT NOT NULL, + quelle_url TEXT, + parsed_at TEXT NOT NULL DEFAULT (datetime('now')), + PRIMARY KEY (bundesland, drucksache, quelle_protokoll) + ) + """) + + # ─── Assessments ─── + now = datetime.utcnow().isoformat() + assessments = [ + # NRW WP18 + ("18/A", "NRW", "2024-01-15", '["GRÜNE"]', 8.5, + _matrix(("A1", 4), ("B2", 3), ("C3", 5), ("D4", 4), ("E5", 5)), + _wp_scores(("GRÜNE", 9, True), ("CDU", 3, False), + ("SPD", 7, False), ("AfD", 1, False))), + ("18/B", "NRW", "2024-02-15", '["GRÜNE"]', 7.5, + _matrix(("A2", 3), ("C3", 4), ("D4", 3), ("E5", 4)), + _wp_scores(("GRÜNE", 8, True), ("CDU", 2, False), + ("SPD", 6, False), ("AfD", 1, False))), + ("18/C", "NRW", "2024-03-15", '["GRÜNE"]', 9.0, + _matrix(("A3", 5), ("D4", 4), ("E5", 5)), + _wp_scores(("GRÜNE", 9, True), ("CDU", 3, False), + ("SPD", 7, False), ("AfD", 1, False))), + ("18/D", "NRW", "2024-04-15", '["GRÜNE"]', 7.0, + _matrix(("B2", 3), ("C2", 4), ("D2", 3)), + _wp_scores(("GRÜNE", 8, True), ("CDU", 4, False), + ("SPD", 7, False), ("AfD", 2, False))), + ("18/E", "NRW", "2024-05-15", '["GRÜNE"]', 8.0, + _matrix(("A1", 4), ("B1", 3), ("E5", 4)), + _wp_scores(("GRÜNE", 9, True), ("CDU", 3, False), + ("SPD", 7, False), ("AfD", 1, False))), + # AfD-Antrag mit niedrigem Score (Anti-Pattern) + ("18/F", "NRW", "2024-06-15", '["AfD"]', 2.0, + _matrix(("A1", -3), ("B2", -2), ("E5", -4)), + _wp_scores(("AfD", 8, True), ("GRÜNE", 1, False), + ("CDU", 4, False), ("SPD", 2, False))), + ("18/G", "NRW", "2024-07-15", '["AfD"]', 1.5, + _matrix(("A1", -4), ("E5", -5)), + _wp_scores(("AfD", 9, True), ("GRÜNE", 1, False), + ("CDU", 3, False), ("SPD", 1, False))), + ("18/H", "NRW", "2024-08-15", '["CDU"]', 5.0, + _matrix(("D4", 1), ("D3", 0)), + _wp_scores(("CDU", 7, True), ("GRÜNE", 4, False), + ("SPD", 5, False), ("AfD", 3, False))), + ("18/I", "NRW", "2024-09-15", '["SPD"]', 6.5, + _matrix(("B2", 3), ("D2", 2)), + _wp_scores(("SPD", 8, True), ("GRÜNE", 6, False), + ("CDU", 4, False), ("AfD", 1, False))), + ("18/J", "NRW", "2024-10-15", '["SPD"]', 4.0, + _matrix(("D4", 1), ("E5", 0)), + _wp_scores(("SPD", 5, True), ("GRÜNE", 3, False), + ("CDU", 5, False), ("AfD", 2, False))), + # MV WP8 fuer Cross-BL + ("8/A", "MV", "2024-04-01", '["GRÜNE"]', 7.0, + _matrix(("A1", 3), ("D4", 4)), + _wp_scores(("GRÜNE", 8, True), ("CDU", 4, False), + ("SPD", 6, False), ("AfD", 2, False))), + ("8/B", "MV", "2024-05-01", '["GRÜNE"]', 8.0, + _matrix(("B2", 4), ("D4", 5)), + _wp_scores(("GRÜNE", 9, True), ("CDU", 3, False), + ("SPD", 7, False), ("AfD", 1, False))), + ("8/C", "MV", "2024-06-01", '["AfD"]', 2.0, + _matrix(("A1", -3), ("E5", -4)), + _wp_scores(("AfD", 9, True), ("GRÜNE", 1, False), + ("CDU", 3, False), ("SPD", 2, False))), + ] + for ds, bl, dat, fr, sc, mat, wps in assessments: + conn.execute( + "INSERT INTO assessments (drucksache, title, fraktionen, datum, " + "bundesland, gwoe_score, gwoe_matrix, wahlprogramm_scores, " + "source, model, created_at, updated_at) VALUES " + "(?, ?, ?, ?, ?, ?, ?, ?, 'test', 'test', ?, ?)", + (ds, f"Test {ds}", fr, dat, bl, sc, mat, wps, now, now), + ) + + # ─── Vote-Results ─── + # Jeder NRW-Hoch-Antrag (A-E,I): GRÜNE+SPD ja, CDU+AfD nein + # AfD-Antrag F,G: AfD ja, GRÜNE+SPD+CDU nein + # CDU-Antrag H: CDU ja, GRÜNE+SPD+AfD nein + # SPD-Antrag J: SPD ja, GRÜNE+CDU+AfD nein + # MV-Hoch-Anträge: GRÜNE+SPD ja, CDU+AfD nein + # MV-AfD-Antrag: AfD ja, andere nein + votes = [ + ("NRW", "18/A", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "MMP18-A"), + ("NRW", "18/B", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "MMP18-B"), + ("NRW", "18/C", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "MMP18-C"), + ("NRW", "18/D", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "MMP18-D"), + ("NRW", "18/E", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "MMP18-E"), + ("NRW", "18/F", "abgelehnt", '["AfD"]', '["GRÜNE","SPD","CDU"]', "MMP18-F"), + ("NRW", "18/G", "abgelehnt", '["AfD"]', '["GRÜNE","SPD","CDU"]', "MMP18-G"), + ("NRW", "18/H", "angenommen", '["CDU","AfD"]', '["GRÜNE","SPD"]', "MMP18-H"), + ("NRW", "18/I", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "MMP18-I"), + ("NRW", "18/J", "abgelehnt", '["SPD"]', '["GRÜNE","CDU","AfD"]', "MMP18-J"), + ("MV", "8/A", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "PlPr8-A"), + ("MV", "8/B", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "PlPr8-B"), + ("MV", "8/C", "abgelehnt", '["AfD"]', '["GRÜNE","SPD","CDU"]', "PlPr8-C"), + ] + for bl, ds, erg, ja, nein, prot in votes: + conn.execute( + "INSERT INTO plenum_vote_results " + "(bundesland, drucksache, ergebnis, fraktionen_ja, fraktionen_nein, " + " quelle_protokoll) VALUES (?, ?, ?, ?, ?, ?)", + (bl, ds, erg, ja, nein, prot), + ) + + conn.commit() + conn.close() + return db + + +# ───────────────────────────────────────────────────────────────────────────── +# aggregate_stimm_index +# ───────────────────────────────────────────────────────────────────────────── + + +class TestAggregateStimmIndex: + def test_grüne_high_index(self, sample_db): + """GRÜNE: stimmt JA bei hohen GWÖ-Anträgen, NEIN bei niedrigen → hoher Index.""" + out = aggregate_stimm_index(db_path=sample_db, min_n=2) + gruene = next(f for f in out["fraktionen"] if f["partei"] == "GRÜNE") + # JA: 18/A,B,D,E,I + 8/A,B (NRW+MV high score). Antragsteller-exclude + # entfernt 18/A-E + 8/A,B fuer GRÜNE → bleibt 18/I als JA. NEIN: 18/H + 18/J + 18/F + 18/G + 8/C. + assert gruene["n_ja"] >= 1 + assert gruene["n_nein"] >= 4 + # Wenn beide N>0: stimm_index sollte positiv sein + if gruene["stimm_index"] is not None: + assert gruene["stimm_index"] > 0 + + def test_afd_low_index(self, sample_db): + """AfD: stimmt NEIN bei hohen GWÖ-Anträgen, JA bei niedrigen → negativer Index.""" + out = aggregate_stimm_index(db_path=sample_db, min_n=2) + afd = next(f for f in out["fraktionen"] if f["partei"] == "AfD") + # AfD ist Antragsteller bei 18/F,G,8/C → diese werden ausgeschlossen. + # Bleibt: NEIN bei 18/A-E, 18/I, 18/J, 8/A, 8/B. AfD JA bei 18/H (CDU-Antrag). + assert afd["n_nein"] >= 4 + # Wenn JA und NEIN beide gefuellt: Index negativ + if afd["stimm_index"] is not None: + assert afd["stimm_index"] < 0 + + def test_exclude_antragsteller_default(self, sample_db): + """Default schliesst eigene Antraege aus.""" + with_excl = aggregate_stimm_index( + db_path=sample_db, exclude_antragsteller=True, min_n=2, + ) + without_excl = aggregate_stimm_index( + db_path=sample_db, exclude_antragsteller=False, min_n=2, + ) + # GRÜNE hat 7 eigene Anträge → ohne Exclude mehr n_ja + gr_with = next(f for f in with_excl["fraktionen"] if f["partei"] == "GRÜNE") + gr_without = next(f for f in without_excl["fraktionen"] if f["partei"] == "GRÜNE") + assert gr_without["n_ja"] > gr_with["n_ja"] + + def test_filter_by_bundesland(self, sample_db): + nrw = aggregate_stimm_index(db_path=sample_db, filter_bl="NRW", min_n=2) + assert nrw["n_assessments_matched"] == 10 + + mv = aggregate_stimm_index(db_path=sample_db, filter_bl="MV", min_n=2) + assert mv["n_assessments_matched"] == 3 + + def test_min_n_threshold(self, sample_db): + """Fraktion mit n_ja=2, min_n=5 → ausreichend=False.""" + out = aggregate_stimm_index(db_path=sample_db, min_n=5) + for f in out["fraktionen"]: + if f["n_ja"] < 5 or f["n_nein"] < 5: + assert f["ausreichend"] is False + + def test_empty_db(self, tmp_path): + out = aggregate_stimm_index(db_path=tmp_path / "missing.db") + assert out["n_assessments_matched"] == 0 + assert out["fraktionen"] == [] + + def test_sorted_by_index_desc(self, sample_db): + """Output sorted by stimm_index descending — None at end.""" + out = aggregate_stimm_index(db_path=sample_db, min_n=1) + indices = [f["stimm_index"] for f in out["fraktionen"]] + # All not-None values should be in descending order + not_none = [v for v in indices if v is not None] + assert not_none == sorted(not_none, reverse=True) + + +# ───────────────────────────────────────────────────────────────────────────── +# aggregate_heuchelei +# ───────────────────────────────────────────────────────────────────────────── + + +class TestAggregateHeuchelei: + def test_returns_structure(self, sample_db): + out = aggregate_heuchelei(db_path=sample_db, min_n=1) + assert "fraktionen" in out + assert "n_assessments_matched" in out + + def test_heuchelei_quote_calculation(self, sample_db): + out = aggregate_heuchelei( + db_path=sample_db, score_threshold=7.0, min_n=1, + ) + # SPD: wahlprogramm_score>=7 in 18/A-E (5x). SPD stimmt JA in allen + # → heuchelei_quote = 0/5 = 0. + spd = next(f for f in out["fraktionen"] if f["partei"] == "SPD") + assert spd["n_im_programm"] >= 5 + assert spd["heuchelei_quote"] == 0 or spd["heuchelei_quote"] is None + + def test_threshold_filter(self, sample_db): + """Hoher Threshold reduziert n_im_programm.""" + low = aggregate_heuchelei(db_path=sample_db, score_threshold=1.0, min_n=1) + high = aggregate_heuchelei(db_path=sample_db, score_threshold=9.5, min_n=1) + # Bei threshold=1 sind quasi alle Fraktionen drin, bei 9.5 sehr wenige + low_total = sum(f["n_im_programm"] for f in low["fraktionen"]) + high_total = sum(f["n_im_programm"] for f in high["fraktionen"]) + assert low_total > high_total + + +# ───────────────────────────────────────────────────────────────────────────── +# _wert_score_for_assessment helper +# ───────────────────────────────────────────────────────────────────────────── + + +class TestWertScoreHelper: + def test_extracts_per_column(self): + matrix = [ + {"field": "A1", "rating": 3}, + {"field": "B1", "rating": 5}, + {"field": "C2", "rating": 4}, + ] + result = _wert_score_for_assessment(matrix) + # Spalte 1 (Würde): A1=3 + B1=5 → Ø=4 + assert result["1"] == 4.0 + # Spalte 2 (Solidarität): C2=4 → Ø=4 + assert result["2"] == 4.0 + + def test_empty_matrix(self): + assert _wert_score_for_assessment([]) == {} + + def test_invalid_entries_skipped(self): + matrix = [ + {"field": "A1", "rating": 3}, + {"field": "", "rating": 5}, # empty field skipped + {"field": "X9", "rating": 2}, # invalid column suffix + "not a dict", # type error + ] + result = _wert_score_for_assessment(matrix) + assert result == {"1": 3.0} + + +# ───────────────────────────────────────────────────────────────────────────── +# aggregate_stimm_index_pro_wert +# ───────────────────────────────────────────────────────────────────────────── + + +class TestAggregateProWert: + def test_structure(self, sample_db): + out = aggregate_stimm_index_pro_wert(db_path=sample_db, min_n=1) + assert out["werte"] == [ + "Menschenwürde", "Solidarität", "Ökologische Nachhaltigkeit", + "Soziale Gerechtigkeit", "Transparenz & Demokratie", + ] + assert "cells" in out + for partei in out["fraktionen"]: + assert set(out["cells"][partei].keys()) == set(out["werte"]) + + def test_cell_format(self, sample_db): + out = aggregate_stimm_index_pro_wert(db_path=sample_db, min_n=1) + if not out["fraktionen"]: + pytest.skip("no parteien — DB empty") + first_partei = out["fraktionen"][0] + cell = out["cells"][first_partei]["Menschenwürde"] + assert "stimm_index" in cell + assert "n_ja" in cell + assert "n_nein" in cell + assert "ausreichend" in cell + + +# ───────────────────────────────────────────────────────────────────────────── +# aggregate_stimm_index_cross_bl +# ───────────────────────────────────────────────────────────────────────────── + + +class TestAggregateCrossBl: + def test_structure(self, sample_db): + out = aggregate_stimm_index_cross_bl(db_path=sample_db, min_n=1) + assert "NRW" in out["bundeslaender"] + assert "MV" in out["bundeslaender"] + assert "fraktionen" in out + assert "fraktionen_alle" in out + + def test_only_multi_bl_in_fraktionen(self, sample_db): + """Fraktionen-Liste enthält nur Parteien mit ≥2 BL ausreichend-erfüllt.""" + out = aggregate_stimm_index_cross_bl(db_path=sample_db, min_n=2) + # Bei min_n=2 sind die meisten Kombinationen nicht ausreichend. + # Test prueft nur Struktur: fraktionen ⊆ fraktionen_alle. + assert set(out["fraktionen"]).issubset(set(out["fraktionen_alle"])) + + def test_cell_format(self, sample_db): + out = aggregate_stimm_index_cross_bl(db_path=sample_db, min_n=1) + if out["fraktionen_alle"]: + partei = out["fraktionen_alle"][0] + for bl in out["bundeslaender"]: + cell = out["cells"][partei][bl] + assert "stimm_index" in cell + assert "n_ja" in cell + assert "n_nein" in cell