diff --git a/app/parlamente.py b/app/parlamente.py index fe8a7b5..f087a19 100644 --- a/app/parlamente.py +++ b/app/parlamente.py @@ -1446,22 +1446,238 @@ class StarFinderCGIAdapter(ParlamentAdapter): class BayernAdapter(ParlamentAdapter): - """Adapter for Bayerischer Landtag.""" + """Adapter for Bayerischer Landtag (#23) — TYPO3-Solr HTML scraping. + + Backend ist eine TYPO3-Site mit ext-solr-Suche unter + ``/parlament/dokumente/drucksachen``. Server-side rendering, keine + SPA, keine API. Reverse-engineering ist trivial — die Drucksachen- + Liste hat ein stabiles HTML-Pattern und der Server akzeptiert die + Filter direkt als URL-Query-Parameter. + + Search-URL: + + GET /parlament/dokumente/drucksachen?dokumentenart=Drucksache + &wahlperiodeid[]=19 + &q= + &sort=date + &anzahl_treffer=100 + &page= + + Response-Pattern (HTML): + +
+
+

+ + Drucksache Nr. 19/11407 vom 08.04.2026 + +

+

Antrag AfD

+
Kostenloses Parken für E-Fahrzeuge…
+
+
+ + Felder pro Eintrag: + * ``Drucksache Nr. 19/ vom DD.MM.YYYY`` → drucksache + datum + * ```` → PDF-Link (Anträge) + oder ``…Schriftliche Anfragen/…pdf`` für Anfragen — Anträge + werden client-seitig über ``

Antrag …`` gefiltert + * ``

Antrag [, ]

`` → typ + Fraktionen + * ``
TITLE
`` → title + + Drucksachen-Lookup nutzt denselben Endpoint mit ``q=``; + die Solr-Suche matcht die Nummer im Volltext und liefert sie als + einzigen oder ersten Treffer. + + Pagination: 100 pro Page (Maximum), max 17.598 Drucksachen in WP19 + Stand 2026-04-10. Wir holen client-side max ``limit*5`` Anträge nach + Filterung. + """ bundesland = "BY" name = "Bayerischer Landtag" base_url = "https://www.bayern.landtag.de" + _RE_RESULT_BLOCK = re.compile( + r'
(.*?)
\s*', re.DOTALL, + ) + _RE_DRUCKSACHE_HEADER = re.compile( + r'Drucksache\s+Nr\.\s*(\d+/\d+)\s*vom\s*(\d{2}\.\d{2}\.\d{4})', + re.IGNORECASE, + ) + _RE_PDF_HREF = re.compile(r'href="([^"]+\.pdf)"') + _RE_TYP_FRAKTION = re.compile(r'

\s*([^<]+?)\s*

') + _RE_TITLE = re.compile(r'
\s*([^<]+)\s*
') + + def __init__(self, *, wahlperiode: int = 19): + self.wahlperiode = wahlperiode + + @staticmethod + def _datum_de_to_iso(datum_de: str) -> str: + if not datum_de: + return "" + try: + d, m, y = datum_de.split(".") + return f"{y}-{m.zfill(2)}-{d.zfill(2)}" + except ValueError: + return "" + + def _parse_results(self, html: str) -> list[Drucksache]: + """Extrahiere alle Drucksachen-Einträge aus einer Result-Page. + + Filtert client-seitig auf ``

Antrag …

`` — die Page enthält + Anträge, Schriftliche Anfragen, Mündliche Anfragen, Berichte und + Gesetzentwürfe gemischt. + """ + from .parteien import extract_fraktionen + + results: list[Drucksache] = [] + for block in self._RE_RESULT_BLOCK.findall(html): + m_header = self._RE_DRUCKSACHE_HEADER.search(block) + if not m_header: + continue + drucksache = m_header.group(1) + datum_iso = self._datum_de_to_iso(m_header.group(2)) + + m_typ = self._RE_TYP_FRAKTION.search(block) + typ_frak = m_typ.group(1).strip() if m_typ else "" + # Format ist " " — Typ ist das erste Token, + # Rest ist Fraktion(en) komma-separiert. + parts = typ_frak.split(None, 1) + typ = parts[0] if parts else "" + fraktionen_text = parts[1] if len(parts) > 1 else "" + + # Bayern listet auch Schriftliche Anfragen, Berichte etc. in + # derselben Liste — wir wollen nur Anträge. + if typ.lower() != "antrag": + continue + + fraktionen = extract_fraktionen( + fraktionen_text, bundesland="BY", + ) + + m_title = self._RE_TITLE.search(block) + title = m_title.group(1).strip() if m_title else f"Drucksache {drucksache}" + # Kollabieren von Mehrfach-Whitespace innerhalb des Titels + title = re.sub(r"\s+", " ", title) + + m_pdf = self._RE_PDF_HREF.search(block) + pdf_url = m_pdf.group(1) if m_pdf else "" + + results.append(Drucksache( + drucksache=drucksache, + title=title, + fraktionen=fraktionen, + datum=datum_iso, + link=pdf_url, + bundesland="BY", + typ=typ, + )) + return results + + def _build_search_params(self, query: str, page: int = 1) -> dict: + # Bayern nutzt PHP-Style-Array-Suffix ``wahlperiodeid[]`` — + # httpx codiert Listen als wiederholte Keys, wir bauen den + # Param-Namen mit ``[]`` direkt in den dict-Key ein. + return { + "dokumentenart": "Drucksache", + "wahlperiodeid[]": str(self.wahlperiode), + "q": query or "", + "sort": "date", + "anzahl_treffer": "100", + "page": str(page), + } + async def search(self, query: str, limit: int = 20) -> list[Drucksache]: - # TODO: Implement Bayern search - return [] + """Volltext-Suche über die aktuelle Wahlperiode, gefiltert auf Anträge. + + Sortiert newest-first (``sort=date``). Holt 1-3 Pages, je 100 + Hits (Antrags-Anteil ist ~10-15% des Drucksachen-Mix), client- + seitig nach ``Antrag``-Typ gefiltert. + """ + url = f"{self.base_url}/parlament/dokumente/drucksachen" + results: list[Drucksache] = [] + seen: set[str] = set() + + async with httpx.AsyncClient( + timeout=30, + follow_redirects=True, + headers={"User-Agent": "Mozilla/5.0 GWOE-Antragspruefer"}, + ) as client: + for page in range(1, 4): # max 300 raw hits → ~30-50 Anträge + try: + resp = await client.get(url, params=self._build_search_params(query, page=page)) + except Exception: + logger.exception("BY search request error page=%d", page) + break + if resp.status_code != 200: + logger.error("BY search HTTP %s page=%d", resp.status_code, page) + break + + page_results = self._parse_results(resp.text) + if not page_results: + break + + for d in page_results: + if d.drucksache in seen: + continue + seen.add(d.drucksache) + results.append(d) + if len(results) >= limit: + return results + + return results async def get_document(self, drucksache: str) -> Optional[Drucksache]: - # TODO: Implement + """Direktes Lookup via ``q=``. Solr-Volltext matcht + die Drucksachen-Nummer und liefert sie als einzigen Hit zurück.""" + url = f"{self.base_url}/parlament/dokumente/drucksachen" + params = self._build_search_params(drucksache, page=1) + + async with httpx.AsyncClient( + timeout=30, follow_redirects=True, + headers={"User-Agent": "Mozilla/5.0 GWOE-Antragspruefer"}, + ) as client: + try: + resp = await client.get(url, params=params) + except Exception: + logger.exception("BY get_document request error for %s", drucksache) + return None + + if resp.status_code != 200: + return None + + for d in self._parse_results(resp.text): + if d.drucksache == drucksache: + return d return None async def download_text(self, drucksache: str) -> Optional[str]: - return None + """Download das Antrags-PDF und extrahiere Volltext.""" + import fitz + + doc = await self.get_document(drucksache) + if doc is None or not doc.link: + return None + + async with httpx.AsyncClient( + timeout=60, follow_redirects=True, + headers={"User-Agent": "Mozilla/5.0 GWOE-Antragspruefer"}, + ) as client: + try: + resp = await client.get(doc.link) + if resp.status_code != 200: + logger.error("BY PDF HTTP %s for %s", resp.status_code, drucksache) + return None + pdf = fitz.open(stream=resp.content, filetype="pdf") + text = "" + for page in pdf: + text += page.get_text() + pdf.close() + return text + except Exception: + logger.exception("BY download error for %s", drucksache) + return None class PARLISAdapter(ParlamentAdapter):