tobias/gwoe-antragspruefer
1
0
Fork 0
Code Issues 16 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Refactor wahlprogramme/embeddings/analyzer for multi-state (#5)

Browse Source 
Atomic refactor of the three modules that previously hardcoded NRW
behaviour. After this commit, every analysis path consults the central
BUNDESLAENDER registry for governing fractions, parliament name, and
state metadata.

wahlprogramme.py
- WAHLPROGRAMME is now nested {bundesland: {partei: meta}}; NRW data
  hoisted unchanged under the "NRW" key.
- New WAHLPROGRAMM_KONTEXT_FILES dict maps a state to its overview
  markdown file (currently only NRW).
- find_relevant_quotes(text, fraktionen, bundesland) — bundesland is
  now a required positional. Governing fractions for the requested
  state are merged with the submitting fractions before lookup.
- Helpers get_wahlprogramm() and parteien_mit_wahlprogramm() expose
  the new shape to other modules.
- ValueError on unknown bundesland (no silent fallback).

embeddings.py
- Schema migration in init_embeddings_db: adds a `bundesland` column
  to the chunks table when missing, plus an index, and backfills
  existing rows from the PROGRAMME registry. Grundsatzprogramme
  (federal level) keep bundesland NULL by design.
- find_relevant_chunks accepts a bundesland filter that matches state
  rows OR NULL — so federal Grundsatzprogramme remain visible to every
  analysis.
- get_relevant_quotes_for_antrag(text, fraktionen, bundesland, …) —
  bundesland required, governing fractions read from BUNDESLAENDER
  instead of hardcoded ["CDU","GRÜNE"]. Order-preserving dedup
  replaces the previous set-based merge.
- index_programm now writes the bundesland column on insert.
- Dropped the hardcoded "Wahlprogramm NRW 2022" label in
  format_quotes_for_prompt — bundesland context is implicit in the
  surrounding prompt block.

analyzer.py
- get_bundesland_context reads parlament_name, regierungsfraktionen,
  landtagsfraktionen and the optional WAHLPROGRAMM_KONTEXT_FILES entry
  from the central registry. Throws ValueError on unknown OR inactive
  bundesland — kills the silent NRW fallback that previously masked
  configuration gaps.
- The Antragsteller-detection heuristic now iterates
  BUNDESLAENDER[bundesland].landtagsfraktionen instead of
  WAHLPROGRAMME.keys(), so we recognise parties for which we don't
  yet have a Wahlprogramm PDF.
- Both quote lookups (semantic + keyword fallback) now receive the
  bundesland.

Resolves issue #5. Foundation for #2 (LSA), #3 (Berlin), #4 (MV).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Dotty Dotter 2026-04-07 18:48:11 +02:00
parent ac18743ff2
commit ee0218b5af
3 changed files with 289 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

95
app/analyzer.py
View File

@ -8,7 +8,12 @@ from openai import AsyncOpenAI
from .config import settings from .config import settings
from .models import Assessment from .models import Assessment
from .wahlprogramme import find_relevant_quotes, format_quote_for_prompt, WAHLPROGRAMME from .bundeslaender import BUNDESLAENDER
from .wahlprogramme import (
find_relevant_quotes,
format_quote_for_prompt,
WAHLPROGRAMM_KONTEXT_FILES,
)
from .embeddings import get_relevant_quotes_for_antrag, format_quotes_for_prompt, EMBEDDINGS_DB from .embeddings import get_relevant_quotes_for_antrag, format_quotes_for_prompt, EMBEDDINGS_DB
# Load context files # Load context files
@ -144,32 +149,52 @@ Antworte NUR mit einem JSON-Objekt im folgenden Format (keine Markdown-Codeblöc
def get_bundesland_context(bundesland: str) -> str: def get_bundesland_context(bundesland: str) -> str:
"""Get context for a specific state.""" """Build the LLM context block for a specific state.
contexts = {
"NRW": {
"wahlprogramme": "wahlprogramme-nrw-2022.md",
"parteiprogramme": "parteiprogramme.md",
"regierungsfraktionen": ["CDU", "GRÜNE"],
}
}
ctx = contexts.get(bundesland, contexts["NRW"])
wahlprogramme = load_context_file(ctx["wahlprogramme"])
parteiprogramme = load_context_file(ctx["parteiprogramme"])
return f"""
## Wahlprogramme {bundesland} 2022
{wahlprogramme} Liest Regierungsfraktionen und Parlamentsname aus ``BUNDESLAENDER`` und
die optionale Wahlprogramm-Übersichtsdatei aus ``WAHLPROGRAMM_KONTEXT_FILES``.
Federal-level Grundsatzprogramme (parteiprogramme.md) sind bundesländer-
übergreifend.
Raises:
ValueError: bei unbekanntem oder inaktivem Bundesland. Pre-#5
existierte hier ein silent fallback auf NRW — bewusst entfernt,
damit Konfigurationslücken früh sichtbar werden.
"""
bl = BUNDESLAENDER.get(bundesland)
if bl is None:
raise ValueError(f"Unbekanntes Bundesland: {bundesland}")
if not bl.aktiv:
raise ValueError(
f"Bundesland {bundesland} ist nicht aktiv (siehe bundeslaender.py)"
)
wahlprogramm_kontext_file = WAHLPROGRAMM_KONTEXT_FILES.get(bundesland)
wahlprogramme_text = (
load_context_file(wahlprogramm_kontext_file) if wahlprogramm_kontext_file else ""
)
parteiprogramme_text = load_context_file("parteiprogramme.md")
return f"""
## Parlament
{bl.parlament_name} (Wahlperiode {bl.wahlperiode}, seit {bl.wahlperiode_start})
## Wahlprogramme {bl.name}
{wahlprogramme_text or '(keine Übersichtsdatei hinterlegt)'}
## Grundsatzprogramme der Parteien ## Grundsatzprogramme der Parteien
{parteiprogramme} {parteiprogramme_text}
## Regierungsfraktionen in {bundesland} ## Regierungsfraktionen in {bl.name}
{', '.join(ctx['regierungsfraktionen'])} {', '.join(bl.regierungsfraktionen)}
## Im Landtag vertretene Fraktionen
{', '.join(bl.landtagsfraktionen)}
Bei Oppositionsanträgen: Bewerte zusätzlich, ob die Regierungsfraktionen zustimmen würden. Bei Oppositionsanträgen: Bewerte zusätzlich, ob die Regierungsfraktionen zustimmen würden.
""" """
@ -185,26 +210,34 @@ async def analyze_antrag(text: str, bundesland: str = "NRW", model: str = "qwen-
system_prompt = get_system_prompt() system_prompt = get_system_prompt()
bundesland_context = get_bundesland_context(bundesland) bundesland_context = get_bundesland_context(bundesland)
# Extrahiere Fraktionen aus Text (einfache Heuristik) # Extrahiere Fraktionen aus Text (einfache Heuristik): Welche der im
fraktionen = [] # Landtag vertretenen Parteien werden im Antrag genannt? Quelle ist
for partei in WAHLPROGRAMME.keys(): # BUNDESLAENDER.landtagsfraktionen — nicht WAHLPROGRAMME, weil wir
if partei in text or partei.lower() in text.lower(): # auch Fraktionen erkennen wollen, für die wir (noch) kein Wahlprogramm
fraktionen.append(partei) # hinterlegt haben.
landtagsfraktionen = BUNDESLAENDER[bundesland].landtagsfraktionen
text_lower = text.lower()
fraktionen = [
partei for partei in landtagsfraktionen
if partei in text or partei.lower() in text_lower
]
# Suche relevante Zitate via semantische Suche (Embeddings) # Suche relevante Zitate via semantische Suche (Embeddings)
quotes_context = "" quotes_context = ""
if EMBEDDINGS_DB.exists(): if EMBEDDINGS_DB.exists():
try: try:
semantic_quotes = get_relevant_quotes_for_antrag(text, fraktionen, top_k_per_partei=2) semantic_quotes = get_relevant_quotes_for_antrag(
text, fraktionen, bundesland=bundesland, top_k_per_partei=2,
)
quotes_context = format_quotes_for_prompt(semantic_quotes) quotes_context = format_quotes_for_prompt(semantic_quotes)
except Exception as e: except Exception as e:
print(f"Semantic search failed: {e}, falling back to keyword search") print(f"Semantic search failed: {e}, falling back to keyword search")
quotes = find_relevant_quotes(text, fraktionen) quotes = find_relevant_quotes(text, fraktionen, bundesland=bundesland)
quotes_context = format_quote_for_prompt(quotes) quotes_context = format_quote_for_prompt(quotes)
else: else:
# Fallback to keyword search # Fallback to keyword search
quotes = find_relevant_quotes(text, fraktionen) quotes = find_relevant_quotes(text, fraktionen, bundesland=bundesland)
quotes_context = format_quote_for_prompt(quotes) quotes_context = format_quote_for_prompt(quotes)
user_prompt = f"""Analysiere den folgenden Antrag: user_prompt = f"""Analysiere den folgenden Antrag:

97
app/embeddings.py
View File

@ -84,7 +84,14 @@ PROGRAMME = {
def init_embeddings_db(): def init_embeddings_db():
"""Initialize the embeddings database.""" """Initialize the embeddings database.
Includes a forward-only migration step (Issue #5): adds the
``bundesland`` column if missing and backfills existing rows from the
``PROGRAMME`` registry. Grundsatzprogramme (federal level) keep
``bundesland = NULL``; the ``find_relevant_chunks`` query treats NULL
as "matches any state".
"""
conn = sqlite3.connect(EMBEDDINGS_DB) conn = sqlite3.connect(EMBEDDINGS_DB)
conn.execute(""" conn.execute("""
CREATE TABLE IF NOT EXISTS chunks ( CREATE TABLE IF NOT EXISTS chunks (
@ -100,6 +107,23 @@ def init_embeddings_db():
""") """)
conn.execute("CREATE INDEX IF NOT EXISTS idx_chunks_partei ON chunks(partei)") conn.execute("CREATE INDEX IF NOT EXISTS idx_chunks_partei ON chunks(partei)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_chunks_typ ON chunks(typ)") conn.execute("CREATE INDEX IF NOT EXISTS idx_chunks_typ ON chunks(typ)")
# Migration: bundesland-Spalte ergänzen, falls Tabelle aus Pre-#5-Zeit
cols = {row[1] for row in conn.execute("PRAGMA table_info(chunks)").fetchall()}
if "bundesland" not in cols:
conn.execute("ALTER TABLE chunks ADD COLUMN bundesland TEXT")
conn.execute("CREATE INDEX IF NOT EXISTS idx_chunks_bundesland ON chunks(bundesland)")
# Backfill: Bundesland aus PROGRAMME-Registry für bestehende Zeilen
# nachtragen. Grundsatzprogramme bleiben NULL.
for prog_id, info in PROGRAMME.items():
bl = info.get("bundesland")
if bl is not None:
conn.execute(
"UPDATE chunks SET bundesland = ? WHERE programm_id = ? AND bundesland IS NULL",
(bl, prog_id),
)
conn.commit() conn.commit()
conn.close() conn.close()
@ -187,8 +211,8 @@ def index_programm(programm_id: str, pdf_dir: Path) -> int:
embedding_blob = json.dumps(embedding).encode() embedding_blob = json.dumps(embedding).encode()
conn.execute(""" conn.execute("""
INSERT INTO chunks (programm_id, partei, typ, seite, text, embedding) INSERT INTO chunks (programm_id, partei, typ, seite, text, embedding, bundesland)
VALUES (?, ?, ?, ?, ?, ?) VALUES (?, ?, ?, ?, ?, ?, ?)
""", ( """, (
programm_id, programm_id,
info["partei"], info["partei"],
@ -196,6 +220,7 @@ def index_programm(programm_id: str, pdf_dir: Path) -> int:
page_num, page_num,
chunk_text_content, chunk_text_content,
embedding_blob, embedding_blob,
info.get("bundesland"), # NULL für Grundsatzprogramme
)) ))
total_chunks += 1 total_chunks += 1
except Exception as e: except Exception as e:
@ -223,29 +248,41 @@ def find_relevant_chunks(
query: str, query: str,
parteien: list[str] = None, parteien: list[str] = None,
typ: str = None, typ: str = None,
bundesland: str = None,
top_k: int = 3, top_k: int = 3,
min_similarity: float = 0.5, min_similarity: float = 0.5,
) -> list[dict]: ) -> list[dict]:
"""Find most relevant chunks for a query.""" """Find most relevant chunks for a query.
Args:
bundesland: Wenn gesetzt, werden nur Chunks dieses Bundeslands ODER
globale Chunks (bundesland IS NULL, z.B. Grundsatzprogramme)
berücksichtigt. Wenn None, kein Filter.
"""
query_embedding = create_embedding(query) query_embedding = create_embedding(query)
conn = sqlite3.connect(EMBEDDINGS_DB) conn = sqlite3.connect(EMBEDDINGS_DB)
conn.row_factory = sqlite3.Row conn.row_factory = sqlite3.Row
# Build query # Build query
sql = "SELECT * FROM chunks WHERE 1=1" sql = "SELECT * FROM chunks WHERE 1=1"
params = [] params = []
if parteien: if parteien:
placeholders = ",".join("?" * len(parteien)) placeholders = ",".join("?" * len(parteien))
sql += f" AND partei IN ({placeholders})" sql += f" AND partei IN ({placeholders})"
params.extend(parteien) params.extend(parteien)
if typ: if typ:
sql += " AND typ = ?" sql += " AND typ = ?"
params.append(typ) params.append(typ)
if bundesland:
# Bundesland-spezifische ODER globale Chunks (Grundsatzprogramme).
sql += " AND (bundesland = ? OR bundesland IS NULL)"
params.append(bundesland)
rows = conn.execute(sql, params).fetchall() rows = conn.execute(sql, params).fetchall()
conn.close() conn.close()
@ -273,39 +310,57 @@ def find_relevant_chunks(
def get_relevant_quotes_for_antrag( def get_relevant_quotes_for_antrag(
antrag_text: str, antrag_text: str,
fraktionen: list[str], fraktionen: list[str],
bundesland: str,
top_k_per_partei: int = 2, top_k_per_partei: int = 2,
) -> dict[str, list[dict]]: ) -> dict[str, list[dict]]:
"""Get relevant quotes from Wahl- and Parteiprogramme for an Antrag.""" """Get relevant quotes from Wahl- and Parteiprogramme for an Antrag.
Args:
bundesland: Pflicht. Bestimmt, welche Wahlprogramme durchsucht werden
und welche Regierungsfraktionen zusätzlich zu den Antragstellern
einbezogen werden.
"""
# Lokaler Import vermeidet Zirkularität: bundeslaender.py importiert nichts
# aus diesem Modul, aber der saubere Trennstrich bleibt erhalten.
from .bundeslaender import BUNDESLAENDER
if bundesland not in BUNDESLAENDER:
raise ValueError(f"Unbekanntes Bundesland: {bundesland}")
regierungsfraktionen = BUNDESLAENDER[bundesland].regierungsfraktionen
parteien_to_search = list(dict.fromkeys(fraktionen + regierungsfraktionen)) # dedupe, Reihenfolge stabil
results = {} results = {}
for partei in fraktionen + ["CDU", "GRÜNE"]: # Include Regierungsfraktionen for partei in parteien_to_search:
partei_upper = partei.upper() if partei != "GRÜNE" else "GRÜNE" partei_upper = partei.upper() if partei != "GRÜNE" else "GRÜNE"
# Wahlprogramm # Wahlprogramm — bundesland-gefiltert
wahl_chunks = find_relevant_chunks( wahl_chunks = find_relevant_chunks(
antrag_text, antrag_text,
parteien=[partei_upper], parteien=[partei_upper],
typ="wahlprogramm", typ="wahlprogramm",
bundesland=bundesland,
top_k=top_k_per_partei, top_k=top_k_per_partei,
min_similarity=0.45, min_similarity=0.45,
) )
# Parteiprogramm # Parteiprogramm (Grundsatz, federal — bundesland=NULL matched implizit)
partei_chunks = find_relevant_chunks( partei_chunks = find_relevant_chunks(
antrag_text, antrag_text,
parteien=[partei_upper], parteien=[partei_upper],
typ="parteiprogramm", typ="parteiprogramm",
bundesland=bundesland,
top_k=top_k_per_partei, top_k=top_k_per_partei,
min_similarity=0.45, min_similarity=0.45,
) )
if wahl_chunks or partei_chunks: if wahl_chunks or partei_chunks:
results[partei_upper] = { results[partei_upper] = {
"wahlprogramm": wahl_chunks, "wahlprogramm": wahl_chunks,
"parteiprogramm": partei_chunks, "parteiprogramm": partei_chunks,
} }
return results return results
@ -320,7 +375,7 @@ def format_quotes_for_prompt(quotes: dict) -> str:
lines.append(f"\n### {partei}\n") lines.append(f"\n### {partei}\n")
if data.get("wahlprogramm"): if data.get("wahlprogramm"):
lines.append("**Wahlprogramm NRW 2022:**") lines.append("**Wahlprogramm:**")
for chunk in data["wahlprogramm"]: for chunk in data["wahlprogramm"]:
text = chunk["text"][:500] + "..." if len(chunk["text"]) > 500 else chunk["text"] text = chunk["text"][:500] + "..." if len(chunk["text"]) > 500 else chunk["text"]
lines.append(f'- S. {chunk["seite"]}: "{text}"') lines.append(f'- S. {chunk["seite"]}: "{text}"')

255
app/wahlprogramme.py
View File

@ -1,126 +1,162 @@
"""Wahlprogramm-Referenzsystem mit Zitaten und Seitenreferenzen.""" """Wahlprogramm-Referenzsystem mit Zitaten und Seitenreferenzen.
Bundesland-bewusst seit Issue #5: ``WAHLPROGRAMME[bundesland][partei]`` statt
flach. Konsumiert ``BUNDESLAENDER`` aus ``bundeslaender.py`` für die
Regierungsfraktionen-Lookup und für Plausibilitätsprüfungen.
Verantwortlich für die schlüsselwortbasierte Fallback-Suche in den
paged-Textversionen der Wahlprogramme. Die semantische Suche lebt in
``embeddings.py``.
"""
import json
import re import re
from pathlib import Path from pathlib import Path
from typing import Optional from typing import Optional
# Wahlprogramm-Metadaten from .bundeslaender import BUNDESLAENDER
WAHLPROGRAMME = {
"CDU": {
"file": "cdu-nrw-2022.pdf", # WAHLPROGRAMME[bundesland][partei] -> Metadaten
"titel": "Machen, worauf es ankommt", # Beim Hinzufügen eines neuen Bundeslands: Eintrag hier UND parallel
"partei": "CDU NRW", # in WAHLPROGRAMM_KONTEXT_FILES.
"jahr": 2022, WAHLPROGRAMME: dict[str, dict[str, dict]] = {
"seiten": 109, "NRW": {
}, "CDU": {
"SPD": { "file": "cdu-nrw-2022.pdf",
"file": "spd-nrw-2022.pdf", "titel": "Machen, worauf es ankommt",
"titel": "Unser Land von morgen", "partei": "CDU NRW",
"partei": "SPD NRW", "jahr": 2022,
"jahr": 2022, "seiten": 109,
"seiten": 116, },
}, "SPD": {
"GRÜNE": { "file": "spd-nrw-2022.pdf",
"file": "gruene-nrw-2022.pdf", "titel": "Unser Land von morgen",
"titel": "Von hier an Zukunft", "partei": "SPD NRW",
"partei": "BÜNDNIS 90/DIE GRÜNEN NRW", "jahr": 2022,
"jahr": 2022, "seiten": 116,
"seiten": 100, },
}, "GRÜNE": {
"FDP": { "file": "gruene-nrw-2022.pdf",
"file": "fdp-nrw-2022.pdf", "titel": "Von hier an Zukunft",
"titel": "Nie gab es mehr zu tun", "partei": "BÜNDNIS 90/DIE GRÜNEN NRW",
"partei": "FDP NRW", "jahr": 2022,
"jahr": 2022, "seiten": 100,
"seiten": 96, },
}, "FDP": {
"AfD": { "file": "fdp-nrw-2022.pdf",
"file": "afd-nrw-2022.pdf", "titel": "Nie gab es mehr zu tun",
"titel": "Wer sonst.", "partei": "FDP NRW",
"partei": "AfD NRW", "jahr": 2022,
"jahr": 2022, "seiten": 96,
"seiten": 68, },
"AfD": {
"file": "afd-nrw-2022.pdf",
"titel": "Wer sonst.",
"partei": "AfD NRW",
"jahr": 2022,
"seiten": 68,
},
}, },
} }
# Basis-Pfad für Referenzdokumente # Pro Bundesland: Markdown-Übersichtsdatei mit Wahlprogramm-Zusammenfassungen,
# wird als Kontext in den LLM-Prompt geladen (nicht für die Suche).
WAHLPROGRAMM_KONTEXT_FILES: dict[str, str] = {
"NRW": "wahlprogramme-nrw-2022.md",
}
REFERENZEN_PATH = Path(__file__).parent / "static" / "referenzen" REFERENZEN_PATH = Path(__file__).parent / "static" / "referenzen"
KONTEXT_PATH = Path(__file__).parent / "kontext" KONTEXT_PATH = Path(__file__).parent / "kontext"
def load_wahlprogramm_text(partei: str) -> dict[int, str]: def get_wahlprogramm(bundesland: str, partei: str) -> Optional[dict]:
"""Liefert die Wahlprogramm-Metadaten oder None, wenn keins vorliegt."""
return WAHLPROGRAMME.get(bundesland, {}).get(partei)
def parteien_mit_wahlprogramm(bundesland: str) -> list[str]:
"""Liste der Parteien, für die im gegebenen Bundesland ein Wahlprogramm vorliegt."""
return list(WAHLPROGRAMME.get(bundesland, {}).keys())
def load_wahlprogramm_text(bundesland: str, partei: str) -> dict[int, str]:
"""Lädt Wahlprogramm-Text mit Seitenzuordnung. """Lädt Wahlprogramm-Text mit Seitenzuordnung.
Returns: Returns:
Dict mit Seitennummer -> Text Dict mit Seitennummer -> Text. Leer, wenn kein Wahlprogramm hinterlegt
oder die paged-Textdatei fehlt.
""" """
if partei not in WAHLPROGRAMME: info = get_wahlprogramm(bundesland, partei)
if not info:
return {} return {}
# Versuche paged-Textdatei zu laden # Versuche paged-Textdatei zu laden
paged_file = KONTEXT_PATH / f"{WAHLPROGRAMME[partei]['file'].replace('.pdf', '-paged.txt')}" paged_file = KONTEXT_PATH / info['file'].replace('.pdf', '-paged.txt')
if not paged_file.exists(): if not paged_file.exists():
# Fallback: Normale Textdatei # Fallback: Normale Textdatei
txt_file = KONTEXT_PATH / f"{WAHLPROGRAMME[partei]['file'].replace('.pdf', '.txt')}" txt_file = KONTEXT_PATH / info['file'].replace('.pdf', '.txt')
if txt_file.exists(): if txt_file.exists():
return {1: txt_file.read_text()} return {1: txt_file.read_text()}
return {} return {}
text = paged_file.read_text() text = paged_file.read_text()
pages = {} pages = {}
current_page = 1 current_page = 1
current_text = [] current_text = []
for line in text.split('\n'): for line in text.split('\n'):
if line.startswith('--- PAGE '): if line.startswith('--- PAGE '):
# Speichere vorherige Seite
if current_text: if current_text:
pages[current_page] = '\n'.join(current_text) pages[current_page] = '\n'.join(current_text)
# Extrahiere neue Seitenzahl
match = re.search(r'PAGE (\d+)', line) match = re.search(r'PAGE (\d+)', line)
if match: if match:
current_page = int(match.group(1)) current_page = int(match.group(1))
current_text = [] current_text = []
else: else:
current_text.append(line) current_text.append(line)
# Letzte Seite speichern
if current_text: if current_text:
pages[current_page] = '\n'.join(current_text) pages[current_page] = '\n'.join(current_text)
return pages return pages
def search_wahlprogramm(partei: str, keywords: list[str], max_results: int = 3) -> list[dict]: def search_wahlprogramm(
bundesland: str,
partei: str,
keywords: list[str],
max_results: int = 3,
) -> list[dict]:
"""Sucht relevante Passagen in einem Wahlprogramm. """Sucht relevante Passagen in einem Wahlprogramm.
Args: Args:
partei: Partei-Kürzel (CDU, SPD, GRÜNE, FDP, AfD) bundesland: Bundesland-Code (NRW, LSA, …)
partei: Partei-Kürzel (CDU, SPD, GRÜNE, FDP, AfD, …)
keywords: Suchbegriffe keywords: Suchbegriffe
max_results: Maximale Anzahl Ergebnisse max_results: Maximale Anzahl Ergebnisse
Returns: Returns:
Liste von {seite, text, score, url} Liste von {bundesland, partei, seite, text, score, url, quelle}
""" """
pages = load_wahlprogramm_text(partei) info = get_wahlprogramm(bundesland, partei)
if not info:
return []
pages = load_wahlprogramm_text(bundesland, partei)
if not pages: if not pages:
return [] return []
results = [] results = []
keywords_lower = [k.lower() for k in keywords] keywords_lower = [k.lower() for k in keywords]
for page_num, text in pages.items(): for page_num, text in pages.items():
text_lower = text.lower() text_lower = text.lower()
# Zähle Keyword-Treffer
score = sum(1 for kw in keywords_lower if kw in text_lower) score = sum(1 for kw in keywords_lower if kw in text_lower)
if score > 0: if score > 0:
# Finde relevante Absätze (mit Keyword)
paragraphs = text.split('\n\n') paragraphs = text.split('\n\n')
relevant_paragraphs = [] relevant_paragraphs = []
for para in paragraphs: for para in paragraphs:
para_clean = para.strip() para_clean = para.strip()
if len(para_clean) < 50: if len(para_clean) < 50:
@ -128,72 +164,79 @@ def search_wahlprogramm(partei: str, keywords: list[str], max_results: int = 3)
para_lower = para_clean.lower() para_lower = para_clean.lower()
if any(kw in para_lower for kw in keywords_lower): if any(kw in para_lower for kw in keywords_lower):
relevant_paragraphs.append(para_clean) relevant_paragraphs.append(para_clean)
if relevant_paragraphs: if relevant_paragraphs:
# Nimm den relevantesten Absatz (mit meisten Keywords) best_para = max(
best_para = max(relevant_paragraphs, relevant_paragraphs,
key=lambda p: sum(1 for kw in keywords_lower if kw in p.lower())) key=lambda p: sum(1 for kw in keywords_lower if kw in p.lower()),
)
# Kürze auf ~300 Zeichen
if len(best_para) > 300: if len(best_para) > 300:
best_para = best_para[:297] + "..." best_para = best_para[:297] + "..."
results.append({ results.append({
"partei": partei, "partei": partei,
"bundesland": bundesland,
"seite": page_num, "seite": page_num,
"text": best_para, "text": best_para,
"score": score, "score": score,
"url": f"/static/referenzen/{WAHLPROGRAMME[partei]['file']}#page={page_num}", "url": f"/static/referenzen/{info['file']}#page={page_num}",
"quelle": f"{WAHLPROGRAMME[partei]['partei']} Wahlprogramm {WAHLPROGRAMME[partei]['jahr']}, S. {page_num}" "quelle": f"{info['partei']} Wahlprogramm {info['jahr']}, S. {page_num}",
}) })
# Sortiere nach Score, nimm Top-Ergebnisse
results.sort(key=lambda x: x['score'], reverse=True) results.sort(key=lambda x: x['score'], reverse=True)
return results[:max_results] return results[:max_results]
def find_relevant_quotes(antrag_text: str, fraktionen: list[str]) -> dict[str, list[dict]]: def find_relevant_quotes(
antrag_text: str,
fraktionen: list[str],
bundesland: str,
) -> dict[str, list[dict]]:
"""Findet relevante Zitate aus Wahlprogrammen für einen Antrag. """Findet relevante Zitate aus Wahlprogrammen für einen Antrag.
Args: Args:
antrag_text: Volltext des Antrags antrag_text: Volltext des Antrags
fraktionen: Liste der Fraktionen (Antragsteller + Regierung) fraktionen: Liste der einreichenden Fraktionen
bundesland: Bundesland-Code (Pflichtparameter; bestimmt, welche
Wahlprogramme durchsucht werden und welche Regierungsfraktionen
zusätzlich einbezogen werden).
Returns: Returns:
Dict mit Partei -> Liste von Zitaten Dict mit Partei -> Liste von Zitaten
""" """
if bundesland not in BUNDESLAENDER:
raise ValueError(f"Unbekanntes Bundesland: {bundesland}")
# Extrahiere Keywords aus Antrag (einfache Heuristik) # Extrahiere Keywords aus Antrag (einfache Heuristik)
# Entferne Stoppwörter und kurze Wörter stopwords = {
stopwords = {'der', 'die', 'das', 'und', 'oder', 'für', 'mit', 'von', 'zu', 'auf', 'der', 'die', 'das', 'und', 'oder', 'für', 'mit', 'von', 'zu', 'auf',
'ist', 'sind', 'wird', 'werden', 'hat', 'haben', 'ein', 'eine', 'einer', 'ist', 'sind', 'wird', 'werden', 'hat', 'haben', 'ein', 'eine', 'einer',
'den', 'dem', 'des', 'im', 'in', 'an', 'bei', 'nach', 'über', 'unter', 'den', 'dem', 'des', 'im', 'in', 'an', 'bei', 'nach', 'über', 'unter',
'durch', 'als', 'auch', 'nur', 'noch', 'aber', 'wenn', 'dass', 'sich', 'durch', 'als', 'auch', 'nur', 'noch', 'aber', 'wenn', 'dass', 'sich',
'nicht', 'wie', 'so', 'aus', 'zum', 'zur', 'vom', 'beim', 'seit', 'bis'} 'nicht', 'wie', 'so', 'aus', 'zum', 'zur', 'vom', 'beim', 'seit', 'bis',
}
words = re.findall(r'\b[A-Za-zäöüÄÖÜß]{4,}\b', antrag_text) words = re.findall(r'\b[A-Za-zäöüÄÖÜß]{4,}\b', antrag_text)
keywords = [w for w in words if w.lower() not in stopwords] keywords = [w for w in words if w.lower() not in stopwords]
# Zähle Worthäufigkeit word_freq: dict[str, int] = {}
word_freq = {}
for w in keywords: for w in keywords:
w_lower = w.lower() w_lower = w.lower()
word_freq[w_lower] = word_freq.get(w_lower, 0) + 1 word_freq[w_lower] = word_freq.get(w_lower, 0) + 1
# Top-Keywords (häufigste)
top_keywords = sorted(word_freq.keys(), key=lambda x: word_freq[x], reverse=True)[:15] top_keywords = sorted(word_freq.keys(), key=lambda x: word_freq[x], reverse=True)[:15]
# Suche in relevanten Wahlprogrammen # Antragsteller + Regierungsfraktionen des Bundeslands
quotes = {} regierungsfraktionen = BUNDESLAENDER[bundesland].regierungsfraktionen
parteien_to_search = set(fraktionen) | set(regierungsfraktionen)
# Immer Regierungsfraktionen einbeziehen
parteien_to_search = set(fraktionen) | {"CDU", "GRÜNE"} quotes: dict[str, list[dict]] = {}
for partei in parteien_to_search: for partei in parteien_to_search:
if partei in WAHLPROGRAMME: if get_wahlprogramm(bundesland, partei):
found = search_wahlprogramm(partei, top_keywords, max_results=2) found = search_wahlprogramm(bundesland, partei, top_keywords, max_results=2)
if found: if found:
quotes[partei] = found quotes[partei] = found
return quotes return quotes
@ -201,14 +244,14 @@ def format_quote_for_prompt(quotes: dict[str, list[dict]]) -> str:
"""Formatiert Zitate für den LLM-Prompt.""" """Formatiert Zitate für den LLM-Prompt."""
if not quotes: if not quotes:
return "" return ""
lines = ["\n## Relevante Passagen aus Wahlprogrammen\n"] lines = ["\n## Relevante Passagen aus Wahlprogrammen\n"]
lines.append("Nutze diese Originalzitate als Belege in deiner Bewertung:\n") lines.append("Nutze diese Originalzitate als Belege in deiner Bewertung:\n")
for partei, zitate in quotes.items(): for partei, zitate in quotes.items():
for z in zitate: for z in zitate:
lines.append(f"### {z['quelle']}") lines.append(f"### {z['quelle']}")
lines.append(f'> "{z["text"]}"') lines.append(f'> "{z["text"]}"')
lines.append("") lines.append("")
return "\n".join(lines) return "\n".join(lines)