gwoe-antragspruefer/app/models.py
Dotty Dotter 8f0f6d6e32 refactor(#136): DDD-Lightweight Tag 1-4 (Ports, Adapter, Repositories, Domain-Verhalten)
ADR 0008: Lightweight-Migration ohne Package-Split

- ports/llm_bewerter.py: Protocol + LlmRequest-Dataclass
- adapters/qwen_bewerter.py: Qwen/DashScope-Adapter mit Retry-Loop
- repositories/{antrag,bewertung,abonnement}_repository.py: Protocol + Sqlite-Impl + InMemory-Fake
- analyzer.py refactored: nimmt Optional[LlmBewerter], AsyncOpenAI-Import raus
- models.py: 5 Domain-Methoden auf Bewertung/MatrixEntry
  (ist_ablehnung, hat_fundamental_kritisches_feld, verletzt_score_cap, ...)
- analyzer loggt WARNING wenn LLM Score-Cap-Invariante verletzt

Folge-PR: Callsite-Migration in main.py (~21 direkte database.*-Aufrufe)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-25 20:55:16 +02:00

242 lines
8.6 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Python types ported from TypeScript types.ts — GWÖ-Matrix 2.0 für Gemeinden."""
from __future__ import annotations
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field
# --- Enums ---
class Empfehlung(str, Enum):
ABLEHNEN = "Ablehnen"
UEBERARBEITEN = "Überarbeiten"
UNTERSTUETZEN_MIT = "Unterstützen mit Änderungen"
UNEINGESCHRAENKT = "Uneingeschränkt unterstützen"
class EmpfehlungSymbol(str, Enum):
X = "[X]"
BANG = "[!]"
PLUS = "[+]"
DPLUS = "[++]"
class Verbesserungspotenzial(str, Enum):
GERING = "gering"
MITTEL = "mittel"
HOCH = "hoch"
FUNDAMENTAL = "fundamental"
# --- Sub-models ---
class MatrixEntry(BaseModel):
field: str = Field(..., pattern=r"^[A-E][1-5]$")
label: str
aspect: str
rating: int = Field(..., ge=-5, le=5) # Neue Skala: -5 bis +5
symbol: Optional[str] = None
# ─── Domain-Verhalten (ADR 0008) ──────────────────────────────────────
def ist_fundamental_kritisch(self) -> bool:
"""True, wenn das Feld einen fundamentalen Widerspruch zu
GWÖ-Werten beschreibt (rating ≤ -4).
Diese Regel triggert den Score-Cap: ein einziges fundamental-
kritisches Feld deckelt den Gesamt-Score auf 3/10 (siehe
``Assessment.verletzt_score_cap``).
"""
return self.rating <= -4
def to_symbol(self) -> str:
"""Berechnet das Matrix-Symbol aus dem Rating.
Quelle: analyzer.py System-Prompt „Matrix-Feldwertung (Skala -5 bis +5)".
Der LLM liefert das Symbol heute selbst; diese Methode erlaubt
server-seitige Konsistenz-Prüfung und ist die Basis, um das
Symbol-Feld perspektivisch ganz aus dem LLM-Output zu entfernen.
"""
r = self.rating
if r >= 4:
return "++"
if r >= 1:
return "+"
if r == 0:
return ""
if r >= -3:
return ""
return ""
class Zitat(BaseModel):
text: str
quelle: str
url: Optional[str] = None
verified: Optional[bool] = None # True=wörtlich im Chunk, False=paraphrasiert, None=pre-#97
class ProgrammScore(BaseModel):
score: float = Field(..., ge=0, le=10)
begruendung: str = Field(..., alias="begründung")
zitate: list[Zitat] = Field(default_factory=list)
model_config = {"populate_by_name": True}
class FraktionScores(BaseModel):
fraktion: str
ist_antragsteller: Optional[bool] = Field(None, alias="istAntragsteller")
ist_regierung: Optional[bool] = Field(None, alias="istRegierung")
wahlprogramm: ProgrammScore
parteiprogramm: ProgrammScore
model_config = {"populate_by_name": True}
class Verbesserung(BaseModel):
original: str
vorschlag: str
begruendung: str
# --- Main Assessment ---
class Assessment(BaseModel):
drucksache: str
title: str
fraktionen: list[str]
datum: str
link: Optional[str] = None
gwoe_score: float = Field(..., ge=0, le=10, alias="gwoeScore")
gwoe_begruendung: str = Field(..., alias="gwoeBegründung")
gwoe_matrix: list[MatrixEntry] = Field(..., alias="gwoeMatrix")
gwoe_schwerpunkt: list[str] = Field(..., alias="gwoeSchwerpunkt")
wahlprogramm_scores: list[FraktionScores] = Field(..., alias="wahlprogrammScores")
verbesserungen: list[Verbesserung] = []
staerken: list[str] = Field(default_factory=list, alias="stärken")
schwaechen: list[str] = Field(default_factory=list, alias="schwächen")
empfehlung: Empfehlung
empfehlung_symbol: Optional[str] = Field(None, alias="empfehlungSymbol")
verbesserungspotenzial: Verbesserungspotenzial
themen: list[str] = []
antrag_zusammenfassung: Optional[str] = Field(None, alias="antragZusammenfassung")
antrag_kernpunkte: Optional[list[str]] = Field(None, alias="antragKernpunkte")
konfidenz: Optional[str] = Field(None, description="LLM-Selbsteinschätzung: hoch/mittel/niedrig")
share_threads: Optional[str] = Field(None, alias="shareThreads", description="Social-Post für Threads (max 500 Zeichen)")
share_twitter: Optional[str] = Field(None, alias="shareTwitter", description="Social-Post für X/Twitter (max 280 Zeichen)")
share_mastodon: Optional[str] = Field(None, alias="shareMastodon", description="Social-Post für Mastodon (max 500 Zeichen)")
# #128: Fraktionen ohne hinterlegtes Wahlprogramm — wird server-seitig
# nach dem LLM-Call befüllt, nicht vom LLM selbst.
fehlende_programme: Optional[list[str]] = Field(
default_factory=list,
alias="fehlendeProgramme",
description="Fraktionen ohne hinterlegtes Wahlprogramm für dieses Bundesland",
)
model_config = {"populate_by_name": True}
# ─── Domain-Verhalten (ADR 0008) ──────────────────────────────────────
def ist_ablehnung(self) -> bool:
"""True, wenn die Empfehlung „Ablehnen" lautet."""
return self.empfehlung == Empfehlung.ABLEHNEN
def ist_uneingeschraenkt_unterstuetzend(self) -> bool:
"""True, wenn die Empfehlung „Uneingeschränkt unterstützen" lautet."""
return self.empfehlung == Empfehlung.UNEINGESCHRAENKT
def hat_fundamental_kritisches_feld(self) -> bool:
"""True, wenn mindestens ein Matrix-Feld rating ≤ -4 hat.
Basis für ``verletzt_score_cap``. Nutzt die VO-Methode
``MatrixEntry.ist_fundamental_kritisch``.
"""
return any(m.ist_fundamental_kritisch() for m in self.gwoe_matrix)
def verletzt_score_cap(self) -> bool:
"""Prüft die Regel aus dem System-Prompt:
Wenn ein Matrix-Feld rating ≤ -4 hat, ist Gesamt-Score max. 3/10.
Der LLM-Prompt formuliert diese Regel als Soll-Anweisung; sie kann
trotzdem verletzt werden. Diese Methode macht die Regel server-
seitig prüfbar und ist der Anker für die Warning-Logik in
``analyzer.py`` (Tag-4-Schritt der DDD-Lightweight-Migration).
"""
return self.hat_fundamental_kritisches_feld() and self.gwoe_score > 3.0
# --- Matrix constants ---
MATRIX_LABELS: dict[str, str] = {
"A1": "Grundrechtsschutz und Menschenwürde in der Lieferkette",
"A2": "Nutzen für die Gemeinde",
"A3": "Ökologische Verantwortung für die Lieferkette",
"A4": "Soziale Verantwortung für die Lieferkette",
"A5": "Öffentliche Rechenschaft und Mitsprache",
"B1": "Ethisches Finanzgebaren / Geld und Mensch",
"B2": "Gemeinnutz im Finanzgebaren",
"B3": "Ökologische Verantwortung der Finanzpolitik",
"B4": "Soziale Verantwortung der Finanzpolitik",
"B5": "Rechenschaft und Partizipation in der Finanzpolitik",
"C1": "Individuelle Rechts- und Gleichstellung",
"C2": "Gemeinsame Zielvereinbarung für das Gemeinwohl",
"C3": "Förderung ökologischen Verhaltens",
"C4": "Gerechte Verteilung von Arbeit",
"C5": "Transparente Kommunikation und demokratische Prozesse",
"D1": "Schutz des Individuums, Rechtsgleichheit",
"D2": "Gesamtwohl in der Gemeinde",
"D3": "Ökologische Gestaltung der öffentlichen Leistung",
"D4": "Soziale Gestaltung der öffentlichen Leistung",
"D5": "Transparente Kommunikation und demokratische Einbindung",
"E1": "Gestaltung der Bedingungen für ein menschenwürdiges Leben zukünftige Generationen",
"E2": "Beitrag zum Gesamtwohl",
"E3": "Verantwortung für ökologische Auswirkungen",
"E4": "Beitrag zum sozialen Ausgleich",
"E5": "Transparente und demokratische Mitbestimmung",
}
ROW_LABELS: dict[str, str] = {
"A": "Ausgelagerte Betriebe, Lieferant:innen, Dienstleister:innen",
"B": "Finanzpartner:innen, Geldgeber:innen, Steuerzahler:innen",
"C": "Politische Führung, Verwaltung, Ehrenamtliche",
"D": "Bürger:innen und Wirtschaft",
"E": "Staat, Gesellschaft und Natur",
}
COL_LABELS = [
"Menschenwürde",
"Solidarität",
"Ökologische Nachhaltigkeit",
"Soziale Gerechtigkeit",
"Transparenz & Demokratie",
]
COL_STAATSPRINZIPIEN = [
"Rechtsstaatsprinzip",
"Gemeinnutz",
"Umwelt-Verantwortung",
"Sozialstaatsprinzip",
"Demokratie",
]
MATRIX_VERSION = "2.0"
MATRIX_TITLE = "Matrix 2.0 für Gemeinden"
EMPFEHLUNG_CONFIG: dict[str, dict] = {
"Ablehnen": {"symbol": "[X]", "color": "#d00000", "css_class": "empf-ablehnen"},
"Überarbeiten": {"symbol": "[!]", "color": "#F7941D", "css_class": "empf-ueberarbeiten"},
"Unterstützen mit Änderungen": {"symbol": "[+]", "color": "#009da5", "css_class": "empf-unterstuetzen"},
"Uneingeschränkt unterstützen": {"symbol": "[++]", "color": "#889e33", "css_class": "empf-voll"},
}