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>
242 lines
8.6 KiB
Python
242 lines
8.6 KiB
Python
"""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"},
|
||
}
|