"""Wahlprogramm-Suche (Keyword-Fallback, wenn Embeddings nicht verfügbar). Bis #222: hier lebte zusätzlich die ``WAHLPROGRAMME``-Datentabelle als zweite Source-of-Truth. Sie wurde nach ``programme.PROGRAMME`` migriert (siehe ADR 0013). Dieses Modul wurde dadurch reduziert auf: - ``get_wahlprogramm(bl, partei)`` — Compat-Adapter auf ``programme.aktuelles_wahlprogramm`` (für Aufrufer, die das Programm via (Bundesland, Partei) suchen — Hauptverwender ist die Keyword-Suche hier in diesem Modul plus admin-Tools). - ``load_wahlprogramm_text`` / ``search_wahlprogramm`` / ``find_relevant_quotes`` / ``format_quote_for_prompt`` — Keyword- Fallback für die ``analyzer.py``-Pipeline, wenn die Embeddings-DB nicht initialisiert ist. Die semantische Suche lebt in ``embeddings.py``, die Programm-Stamm- daten in ``programme.py``. """ import re from pathlib import Path from typing import Optional from .bundeslaender import BUNDESLAENDER from .programme import ( aktuelles_wahlprogramm, parteien_mit_wahlprogramm, ) # 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" KONTEXT_PATH = Path(__file__).parent / "kontext" def get_wahlprogramm(bundesland: str, partei: str) -> Optional[dict]: """Liefert das aktuell gültige Wahlprogramm dieser Partei in dem Bundesland — als ``programme.Programm``-Dict (oder ``None``). Compat-Adapter auf ``programme.aktuelles_wahlprogramm``. Aufrufer, die früher ``info["file"]`` gelesen haben, müssen ``info["pdf"]`` lesen; ``info["partei"]`` (Langform) gibt es nicht mehr — Kurzform + ``info["bundesland"]`` reichen für die Anzeige. """ return aktuelles_wahlprogramm(bundesland, partei) def load_wahlprogramm_text(bundesland: str, partei: str) -> dict[int, str]: """Lädt Wahlprogramm-Text mit Seitenzuordnung. Returns: Dict mit Seitennummer -> Text. Leer, wenn kein Wahlprogramm hinterlegt oder die paged-Textdatei fehlt. """ info = get_wahlprogramm(bundesland, partei) if not info: return {} pdf_name = info.get("pdf", "") if not pdf_name: return {} paged_file = KONTEXT_PATH / pdf_name.replace('.pdf', '-paged.txt') if not paged_file.exists(): # Fallback: Normale Textdatei txt_file = KONTEXT_PATH / pdf_name.replace('.pdf', '.txt') if txt_file.exists(): return {1: txt_file.read_text()} return {} text = paged_file.read_text() pages = {} current_page = 1 current_text = [] for line in text.split('\n'): if line.startswith('--- PAGE '): if current_text: pages[current_page] = '\n'.join(current_text) match = re.search(r'PAGE (\d+)', line) if match: current_page = int(match.group(1)) current_text = [] else: current_text.append(line) if current_text: pages[current_page] = '\n'.join(current_text) return pages def search_wahlprogramm( bundesland: str, partei: str, keywords: list[str], max_results: int = 3, ) -> list[dict]: """Sucht relevante Passagen in einem Wahlprogramm (Keyword-basiert, Fallback wenn die Embeddings-DB nicht da ist). Returns: Liste von {bundesland, partei, seite, text, score, url, quelle}. """ info = get_wahlprogramm(bundesland, partei) if not info: return [] pages = load_wahlprogramm_text(bundesland, partei) if not pages: return [] pdf_name = info.get("pdf", "") name = info.get("name") or partei jahr = info.get("gueltig_ab", "")[:4] or "?" results = [] keywords_lower = [k.lower() for k in keywords] for page_num, text in pages.items(): text_lower = text.lower() score = sum(1 for kw in keywords_lower if kw in text_lower) if score > 0: paragraphs = text.split('\n\n') relevant_paragraphs = [] for para in paragraphs: para_clean = para.strip() if len(para_clean) < 50: continue para_lower = para_clean.lower() if any(kw in para_lower for kw in keywords_lower): relevant_paragraphs.append(para_clean) if relevant_paragraphs: best_para = max( relevant_paragraphs, key=lambda p: sum(1 for kw in keywords_lower if kw in p.lower()), ) if len(best_para) > 300: best_para = best_para[:297] + "..." results.append({ "partei": partei, "bundesland": bundesland, "seite": page_num, "text": best_para, "score": score, "url": f"/static/referenzen/{pdf_name}#page={page_num}", "quelle": f"{name}, S. {page_num}", }) results.sort(key=lambda x: x['score'], reverse=True) return results[:max_results] 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. Args: antrag_text: Volltext des Antrags fraktionen: Liste der einreichenden Fraktionen bundesland: Bundesland-Code (Pflichtparameter; bestimmt, welche Wahlprogramme durchsucht werden und welche Regierungsfraktionen zusätzlich einbezogen werden). Returns: Dict mit Partei -> Liste von Zitaten """ if bundesland not in BUNDESLAENDER: raise ValueError(f"Unbekanntes Bundesland: {bundesland}") # Extrahiere Keywords aus Antrag (einfache Heuristik) stopwords = { 'der', 'die', 'das', 'und', 'oder', 'für', 'mit', 'von', 'zu', 'auf', 'ist', 'sind', 'wird', 'werden', 'hat', 'haben', 'ein', 'eine', 'einer', 'den', 'dem', 'des', 'im', 'in', 'an', 'bei', 'nach', 'über', 'unter', 'durch', 'als', 'auch', 'nur', 'noch', 'aber', 'wenn', 'dass', 'sich', 'nicht', 'wie', 'so', 'aus', 'zum', 'zur', 'vom', 'beim', 'seit', 'bis', } words = re.findall(r'\b[A-Za-zäöüÄÖÜß]{4,}\b', antrag_text) keywords = [w for w in words if w.lower() not in stopwords] word_freq: dict[str, int] = {} for w in keywords: w_lower = w.lower() word_freq[w_lower] = word_freq.get(w_lower, 0) + 1 top_keywords = sorted(word_freq.keys(), key=lambda x: word_freq[x], reverse=True)[:15] # Antragsteller + Regierungsfraktionen des Bundeslands regierungsfraktionen = BUNDESLAENDER[bundesland].regierungsfraktionen parteien_to_search = set(fraktionen) | set(regierungsfraktionen) quotes: dict[str, list[dict]] = {} for partei in parteien_to_search: if get_wahlprogramm(bundesland, partei): found = search_wahlprogramm(bundesland, partei, top_keywords, max_results=2) if found: quotes[partei] = found return quotes def format_quote_for_prompt(quotes: dict[str, list[dict]]) -> str: """Formatiert Zitate für den LLM-Prompt.""" if not quotes: return "" lines = ["\n## Relevante Passagen aus Wahlprogrammen\n"] for partei, partei_quotes in quotes.items(): if partei_quotes: lines.append(f"\n### {partei}") for q in partei_quotes: lines.append(f"- S. {q['seite']}: \"{q['text']}\"") return "\n".join(lines)