From d81753c4fb6f912e04bd696761e7393a817654d6 Mon Sep 17 00:00:00 2001
From: Dotty Dotter
Date: Wed, 29 Apr 2026 22:56:35 +0200
Subject: [PATCH] feat(#167): Empfehlungs-Konsistenz + CSV-Export
Stimmverhalten
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
Phase-2-Erweiterungen des Stimmverhalten-Tabs:
**1. Empfehlungs-Konsistenz (#167):**
Pro Fraktion: Anteil der Anträge mit GWÖ-Empfehlung
"Uneingeschränkt unterstützen" oder "Unterstützen mit Änderungen",
bei denen die Fraktion trotzdem NEIN gestimmt hat. Orthogonal zur
Heuchelei-Quote — prüft NICHT gegen Wahlprogramm-Treue, sondern gegen
die GWÖ-Empfehlung des Systems.
- `aggregate_empfehlungs_konsistenz()` in app/auswertungen.py
- `GET /api/auswertungen/empfehlungs-konsistenz`
- 5. Chart-Sub-Section im Stimmverhalten-Tab (rote Bar Chart, 0..100%)
**2. CSV-Export (Phase-1-Querschnitts-TODO):**
Long-Format-CSV mit Spalten: drucksache, bundesland, wahlperiode, datum,
gwoe_score, empfehlung, partei, vote, ist_antragsteller. Macht alle
Stimmverhalten-Aussagen wissenschaftlich auswertbar (R/pandas/Excel).
- `export_stimmverhalten_csv()` in app/auswertungen.py
- `GET /api/auswertungen/stimmverhalten.csv` mit
Filter-Parametern bundesland/wahlperiode/exclude_antragsteller
- "CSV-Export"-Button im Stimmverhalten-Tab neben dem Toggle
**Tests:** 27 Stimmverhalten-Tests (war 18, +4 Empfehlungs-Konsistenz,
+5 CSV-Export). Fixture um `empfehlung`-Spalte erweitert.
Suite: 989 Tests grün (war 980).
Co-Authored-By: Claude Opus 4.7 (1M context)
---
app/auswertungen.py | 150 +++++++++++++++++++++
app/main.py | 38 ++++++
app/templates/v2/screens/auswertungen.html | 109 +++++++++++++--
tests/test_auswertungen_stimmverhalten.py | 131 +++++++++++++++---
4 files changed, 400 insertions(+), 28 deletions(-)
diff --git a/app/auswertungen.py b/app/auswertungen.py
index 6643ad6..f1093b9 100644
--- a/app/auswertungen.py
+++ b/app/auswertungen.py
@@ -588,6 +588,89 @@ def aggregate_stimm_index_pro_wert(
}
+def aggregate_empfehlungs_konsistenz(
+ filter_bl: Optional[str] = None,
+ filter_wp: Optional[str] = None,
+ min_n: int = 5,
+ db_path: Optional[Path] = None,
+) -> dict:
+ """Pro Fraktion: Anteil der Antraege mit GWÖ-Empfehlung "Uneingeschraenkt
+ unterstuetzen" oder "Unterstuetzen mit Aenderungen", bei denen die
+ Fraktion trotzdem NEIN gestimmt hat.
+
+ Orthogonal zu Heuchelei-Score: prueft NICHT gegen Wahlprogramm-Treue,
+ sondern gegen die GWÖ-Empfehlung des Systems. Misst Inkonsistenz
+ zwischen "GWÖ haelt Antrag fuer gut" und "Fraktion stimmt dagegen".
+ """
+ rows = _load_assessments_with_votes(filter_bl, filter_wp, db_path)
+ # Empfehlung ist im JOIN-Helper noch nicht — eigener Lookup pro Drucksache.
+ # Statt Helper umzubauen: zweite Query auf assessments fuer empfehlung.
+ path = db_path or settings.db_path
+ if not Path(path).exists():
+ return {"fraktionen": [], "n_assessments_matched": 0, "filter": {
+ "bundesland": filter_bl, "wahlperiode": filter_wp, "min_n": min_n,
+ }}
+ conn = sqlite3.connect(str(path))
+ try:
+ empfehlung_map = {
+ (r[0], r[1]): r[2] for r in conn.execute(
+ "SELECT bundesland, drucksache, empfehlung FROM assessments"
+ ).fetchall()
+ }
+ finally:
+ conn.close()
+
+ POSITIV = {"Uneingeschränkt unterstützen", "Unterstützen mit Änderungen"}
+
+ n_empfohlen: defaultdict[str, int] = defaultdict(int)
+ n_nein: defaultdict[str, int] = defaultdict(int)
+ n_ja: defaultdict[str, int] = defaultdict(int)
+ n_enth: defaultdict[str, int] = defaultdict(int)
+ seen_drucksachen = set()
+
+ for row in rows:
+ empfehlung = empfehlung_map.get((row["bundesland"], row["drucksache"]))
+ if empfehlung not in POSITIV:
+ continue
+ seen_drucksachen.add((row["bundesland"], row["drucksache"]))
+ all_voters = row["ja"] | row["nein"] | row["enthaltung"]
+ for f in all_voters:
+ n_empfohlen[f] += 1
+ if f in row["nein"]:
+ n_nein[f] += 1
+ elif f in row["ja"]:
+ n_ja[f] += 1
+ elif f in row["enthaltung"]:
+ n_enth[f] += 1
+
+ fraktionen_out = []
+ for partei in sorted(n_empfohlen):
+ total = n_empfohlen[partei]
+ nein = n_nein[partei]
+ quote = round(nein / total, 3) if total else None
+ fraktionen_out.append({
+ "partei": partei,
+ "n_empfohlen": total,
+ "n_nein_trotz_empfehlung": nein,
+ "n_ja": n_ja[partei],
+ "n_enth": n_enth[partei],
+ "konsistenz_quote": quote,
+ "ausreichend": total >= min_n,
+ })
+ fraktionen_out.sort(
+ key=lambda f: (f["konsistenz_quote"] or 0), reverse=True,
+ )
+ return {
+ "fraktionen": fraktionen_out,
+ "n_assessments_matched": len(seen_drucksachen),
+ "filter": {
+ "bundesland": filter_bl,
+ "wahlperiode": filter_wp,
+ "min_n": min_n,
+ },
+ }
+
+
def aggregate_stimm_index_cross_bl(
filter_wp: Optional[str] = None,
exclude_antragsteller: bool = True,
@@ -655,3 +738,70 @@ def aggregate_stimm_index_cross_bl(
"min_n": min_n,
},
}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 5. CSV-Export der Stimmverhalten-Aggregationen
+#
+# Long-Format-CSV pro Aggregation, analog zu export_long_format(). Macht die
+# Aussagen wissenschaftlich auswertbar (R/pandas/Excel) ohne JSON-Parsing.
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def export_stimmverhalten_csv(
+ filter_bl: Optional[str] = None,
+ filter_wp: Optional[str] = None,
+ exclude_antragsteller: bool = True,
+ db_path: Optional[Path] = None,
+) -> str:
+ """Long-Format-CSV: Eine Zeile pro (drucksache, partei, vote).
+
+ Spalten: drucksache, bundesland, wahlperiode, datum, gwoe_score,
+ empfehlung, partei, vote (ja|nein|enthaltung), ist_antragsteller.
+
+ Eine Zeile pro Fraktion-Stimme — wer also an N Anträgen mit Vote
+ teilgenommen hat, hat N Zeilen.
+ """
+ rows = _load_assessments_with_votes(filter_bl, filter_wp, db_path)
+ # Empfehlung-Map analog aggregate_empfehlungs_konsistenz
+ path = db_path or settings.db_path
+ empfehlung_map: dict[tuple[str, str], str] = {}
+ if Path(path).exists():
+ conn = sqlite3.connect(str(path))
+ try:
+ empfehlung_map = {
+ (r[0], r[1]): r[2] or "" for r in conn.execute(
+ "SELECT bundesland, drucksache, empfehlung FROM assessments"
+ ).fetchall()
+ }
+ finally:
+ conn.close()
+
+ buf = io.StringIO()
+ writer = csv.writer(buf, dialect="excel")
+ writer.writerow([
+ "drucksache", "bundesland", "wahlperiode", "datum",
+ "gwoe_score", "empfehlung", "partei", "vote", "ist_antragsteller",
+ ])
+
+ for row in rows:
+ bl = row["bundesland"]
+ wp = wahlperiode_for(row["datum"], bl) if bl else ""
+ empfehlung = empfehlung_map.get((bl, row["drucksache"]), "")
+ antragsteller = row["antragsteller"]
+ for vote_key, voters in [
+ ("ja", row["ja"]),
+ ("nein", row["nein"]),
+ ("enthaltung", row["enthaltung"]),
+ ]:
+ for partei in sorted(voters):
+ if exclude_antragsteller and partei in antragsteller:
+ continue
+ writer.writerow([
+ row["drucksache"], bl, wp or "", row["datum"],
+ f"{row['gwoe_score']:.2f}",
+ empfehlung,
+ partei, vote_key,
+ "1" if partei in antragsteller else "0",
+ ])
+ return buf.getvalue()
diff --git a/app/main.py b/app/main.py
index 64bfa0c..b00b0c2 100644
--- a/app/main.py
+++ b/app/main.py
@@ -2370,6 +2370,44 @@ async def auswertungen_stimm_index_cross_bl(
)
+@app.get("/api/auswertungen/empfehlungs-konsistenz")
+async def auswertungen_empfehlungs_konsistenz(
+ bundesland: Optional[str] = None,
+ wahlperiode: Optional[str] = None,
+ min_n: int = 5,
+):
+ """Pro Fraktion: Anteil der Anträge mit GWÖ-Empfehlung "Uneingeschränkt
+ unterstützen"/"Unterstützen mit Änderungen", bei denen die Fraktion
+ trotzdem NEIN gestimmt hat (#167)."""
+ from .auswertungen import aggregate_empfehlungs_konsistenz
+ return aggregate_empfehlungs_konsistenz(
+ filter_bl=bundesland,
+ filter_wp=wahlperiode,
+ min_n=min_n,
+ )
+
+
+@app.get("/api/auswertungen/stimmverhalten.csv")
+async def auswertungen_stimmverhalten_csv(
+ bundesland: Optional[str] = None,
+ wahlperiode: Optional[str] = None,
+ exclude_antragsteller: bool = True,
+):
+ """Long-Format-CSV: eine Zeile pro (drucksache, partei, vote). Macht die
+ Stimmverhalten-Aussagen wissenschaftlich auswertbar (R/pandas/Excel)."""
+ from .auswertungen import export_stimmverhalten_csv
+ csv_text = export_stimmverhalten_csv(
+ filter_bl=bundesland,
+ filter_wp=wahlperiode,
+ exclude_antragsteller=exclude_antragsteller,
+ )
+ return Response(
+ content=csv_text,
+ media_type="text/csv",
+ headers={"Content-Disposition": 'attachment; filename="gwoe-stimmverhalten.csv"'},
+ )
+
+
# ─── 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 6a1dfe2..0da2090 100644
--- a/app/templates/v2/screens/auswertungen.html
+++ b/app/templates/v2/screens/auswertungen.html
@@ -217,12 +217,20 @@ table.modal-table th { background: var(--ecg-bg-subtle); font-weight: 700; }
Antragsteller-Fraktionen quasi immer „ja" stimmen — das würde den Index
verzerren.
-
-
- Eigene Anträge ausschließen
-
+
+
+
+ Eigene Anträge ausschließen
+
+
+ CSV-Export (Long-Format)
+
+
@@ -265,9 +273,23 @@ table.modal-table th { background: var(--ecg-bg-subtle); font-weight: 700; }
-
+
4. Stimm-Index pro Bundesland (Cross-BL)
+ margin:1.5rem 0 0.5rem;">4. Empfehlungs-Konsistenz (gegen GWÖ-Empfehlung)
+
+ Anteil der Anträge mit GWÖ-Empfehlung „Uneingeschränkt unterstützen" oder
+ „Unterstützen mit Änderungen", bei denen die Fraktion trotzdem
+ Nein gestimmt hat. Orthogonal zur Heuchelei-Quote — prüft NICHT
+ gegen Wahlprogramm-Treue, sondern gegen die GWÖ-Empfehlung des Systems.
+
+
+
+
+
+
+
+ 5. 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
@@ -311,7 +333,7 @@ table.modal-table th { background: var(--ecg-bg-subtle); font-weight: 700; }
{% block body_scripts %}