Phase G: BundestagAdapter via DIP-API (#56)
Schließt #56 (Bundespolitik überprüfbar machen). Neuer ``BundestagAdapter`` in ``app/parlamente.py``, neuer ``BUND``-Eintrag in ``app/bundeslaender.py`` als 17. Parlament-Slot. API: - DIP-Search-API auf ``search.dip.bundestag.de/api/v1/drucksache`` - API-Key aus ``dip-config.js`` gescraped (öffentlich, klartext) - Auth via URL-Param ``?apikey=...`` plus ``Origin: https://dip.bundestag.de``- Header (Origin-Locking, server-to-server-tauglich) - Pagination via ``cursor``-Parameter, 100 Hits pro Page - ``f.drucksachetyp=Antrag`` und ``f.wahlperiode=21`` als Server-Filter Mapping: - ``dokumentnummer`` → ``Drucksache.drucksache`` - ``titel`` → ``title`` - ``urheber[*].titel`` → durch ``parteien.extract_fraktionen`` zu ``["AfD"]``/``["GRÜNE"]``/etc. — die ``"Fraktion der AfD"``- Schreibweise wird vom zentralen Mapper aus #55 bereits korrekt geparst, kein Adapter-spezifisches Pattern nötig - ``fundstelle.pdf_url`` → ``link`` - ``datum`` → bereits ISO ``YYYY-MM-DD`` ``get_document(drucksache)`` nutzt ``f.dokumentnummer`` als direkter Server-Filter, kein linearer Pagination-Scan. BUND-Eintrag in ``bundeslaender.py``: - ``code="BUND"``, ``parlament_name="Deutscher Bundestag"``, ``wahlperiode=21``, ``wahlperiode_start="2025-03-25"`` (Konstituierung 21. WP nach BTW 2025), ``regierungsfraktionen=["CDU", "CSU", "SPD"]`` (Kabinett Merz) - ``aktiv=True`` — taucht automatisch in ``alle_bundeslaender()`` und ``aktive_bundeslaender()`` auf, damit die UI- und Auswertungs-Pipelines BUND ohne zusätzliche Sonderpfade kennen - 17 Einträge in ``BUNDESLAENDER`` statt 16 — Tests entsprechend aktualisiert (``test_sixteen_bundeslaender_plus_bund``, ``test_alle_bundeslaender_returns_all``, ``test_all_wahlperioden_lists_each_bl_twice``) Live-Probe direkt im Repo: ``` adapter: Deutscher Bundestag (DIP), wahlperiode=21 search returned 5 docs 21/5136 2026-03-31 | ['AfD'] | Transparenz, Wirtschaftlichkeit ... 21/5064 2026-03-27 | ['GRÜNE'] | Ausverkauf der Energieinfrastruktur ... 21/5059 2026-03-27 | ['AfD'] | Berufsfreiheit für Selbstständige ... get_document('21/5136') -> drucksache=21/5136 ``` 176 Unit-Tests grün, Live-Verifikation Sub-A im Container nach Deploy. Refs: #56, #59 (Phase G) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dotty Dotter
parent
15b9af8795
commit
0f7d35f20e
@ -66,7 +66,31 @@ class Bundesland:
|
||||
|
||||
# Hauptregister: code -> Bundesland-Instanz.
|
||||
# Reihenfolge alphabetisch nach offiziellem Namen für stabile UI-Sortierung.
|
||||
# Sonder-Eintrag "BUND" für den Deutschen Bundestag (technisch kein BL,
|
||||
# aber teilt die gesamte Adapter/Analyzer-Pipeline mit den 16 BL).
|
||||
BUNDESLAENDER: dict[str, Bundesland] = {
|
||||
"BUND": Bundesland(
|
||||
code="BUND",
|
||||
name="Deutscher Bundestag",
|
||||
parlament_name="Deutscher Bundestag",
|
||||
wahlperiode=21,
|
||||
wahlperiode_start="2025-03-25", # Konstituierung 21. WP nach BTW 2025
|
||||
naechste_wahl="2029-09-30", # geschätzt
|
||||
regierungsfraktionen=["CDU", "CSU", "SPD"], # Kabinett Merz, schwarz-rot
|
||||
landtagsfraktionen=["CDU", "CSU", "AfD", "SPD", "GRÜNE", "LINKE", "BSW", "FDP"],
|
||||
doku_system="DIP",
|
||||
doku_base_url="https://search.dip.bundestag.de",
|
||||
drucksache_format="21/12345",
|
||||
dokukratie_scraper=None,
|
||||
aktiv=True,
|
||||
anmerkung=(
|
||||
"DIP-API auf search.dip.bundestag.de mit öffentlichem "
|
||||
"API-Key aus dip-config.js und Origin-Header-Locking auf "
|
||||
"https://dip.bundestag.de. ~600 Anträge pro Wahlperiode. "
|
||||
"Kabinett Merz seit Mai 2025 (CDU/CSU+SPD nach BSW-Aus). "
|
||||
"BundestagAdapter implementiert in #56."
|
||||
),
|
||||
),
|
||||
"BW": Bundesland(
|
||||
code="BW",
|
||||
name="Baden-Württemberg",
|
||||
|
||||
@ -1845,8 +1845,221 @@ class PARLISAdapter(ParlamentAdapter):
|
||||
return None
|
||||
|
||||
|
||||
class BundestagAdapter(ParlamentAdapter):
|
||||
"""Adapter für den Deutschen Bundestag via DIP-API.
|
||||
|
||||
Quelle: ``search.dip.bundestag.de/api/v1`` — die offizielle REST-API
|
||||
des Dokumentations- und Informationssystems (DIP). Schema dokumentiert
|
||||
unter https://dip.bundestag.de/über-dip/hilfe/api (SPA, Inhalt im
|
||||
Bundle ``main.*.chunk.js``). Auth via URL-Parameter ``apikey=...``
|
||||
PLUS einem ``Origin: https://dip.bundestag.de``-Header — der Server
|
||||
macht Origin-Locking auf seine eigene Single-Page-App.
|
||||
|
||||
Der API-Key liegt offen in ``dip-config.js`` und wird vom DIP-Frontend
|
||||
bei jedem Request als URL-Parameter mitgeschickt. Solange wir den
|
||||
Origin-Header setzen, akzeptiert die API das auch von server-to-
|
||||
server-Calls.
|
||||
|
||||
Doc-Mapping (``/api/v1/drucksache``):
|
||||
|
||||
- ``dokumentnummer`` → ``drucksache`` (z.B. ``"21/5136"``)
|
||||
- ``titel`` → ``title``
|
||||
- ``urheber[*].bezeichnung``/``titel`` → ``fraktionen`` (durch
|
||||
``parteien.extract_fraktionen`` normalisiert, deckt
|
||||
``"Fraktion der AfD"`` → ``"AfD"`` ab)
|
||||
- ``datum`` → ``datum`` (bereits ISO YYYY-MM-DD)
|
||||
- ``fundstelle.pdf_url`` → ``link``
|
||||
- ``drucksachetyp`` → ``typ`` (Filter auf ``"Antrag"``)
|
||||
|
||||
Pagination via ``cursor``-Parameter — der Server gibt nach jedem
|
||||
Result einen neuen Cursor zurück, den wir als nächsten Request
|
||||
mitschicken. 100 Hits pro Page, pro Wahlperiode ~600 Anträge.
|
||||
"""
|
||||
|
||||
bundesland = "BUND"
|
||||
name = "Deutscher Bundestag (DIP)"
|
||||
base_url = "https://search.dip.bundestag.de/api/v1"
|
||||
|
||||
# Aus dip-config.js gescraped (öffentlich, klartext, von der DIP-SPA
|
||||
# bei jedem Request mitgesendet). Origin-Locking macht den Key
|
||||
# nicht-trivial weiterzugeben, aber für server-to-server-Calls mit
|
||||
# gesetztem Origin-Header voll funktional.
|
||||
DEFAULT_APIKEY = "SbGXhWA.3cpnNdb8rkht7iWpvSgTP8XIG88LoCrGd4"
|
||||
ORIGIN = "https://dip.bundestag.de"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
apikey: Optional[str] = None,
|
||||
wahlperiode: int = 21,
|
||||
document_typ: str = "Antrag",
|
||||
):
|
||||
self.apikey = apikey or self.DEFAULT_APIKEY
|
||||
self.wahlperiode = wahlperiode
|
||||
self.document_typ = document_typ
|
||||
|
||||
def _make_client(self) -> httpx.AsyncClient:
|
||||
return httpx.AsyncClient(
|
||||
timeout=30,
|
||||
follow_redirects=True,
|
||||
headers={
|
||||
"Origin": self.ORIGIN,
|
||||
"Referer": f"{self.ORIGIN}/",
|
||||
"User-Agent": "Mozilla/5.0 GWOE-Antragspruefer",
|
||||
"Accept": "application/json",
|
||||
},
|
||||
)
|
||||
|
||||
def _doc_to_drucksache(self, doc: dict) -> Optional[Drucksache]:
|
||||
"""Map ein DIP-/drucksache-JSON auf unser ``Drucksache``-dataclass.
|
||||
``None`` wenn essentielle Felder fehlen."""
|
||||
from .parteien import extract_fraktionen
|
||||
|
||||
nummer = doc.get("dokumentnummer")
|
||||
if not nummer:
|
||||
return None
|
||||
|
||||
# PDF-URL aus fundstelle ziehen — ist die zuverlässige Adresse
|
||||
fundstelle = doc.get("fundstelle") or {}
|
||||
pdf_url = fundstelle.get("pdf_url") or ""
|
||||
if not pdf_url:
|
||||
return None
|
||||
|
||||
# Fraktionen aus urheber-Liste extrahieren. DIP listet sie als
|
||||
# "Fraktion der AfD" o.ä. — extract_fraktionen kennt das Pattern
|
||||
# bereits aus den Landtags-Adaptern.
|
||||
urheber_strs: list[str] = []
|
||||
for u in (doc.get("urheber") or []):
|
||||
if isinstance(u, dict):
|
||||
urheber_strs.append(u.get("titel") or u.get("bezeichnung") or "")
|
||||
urheber_combined = ", ".join(filter(None, urheber_strs))
|
||||
fraktionen = extract_fraktionen(urheber_combined, bundesland=self.bundesland)
|
||||
|
||||
return Drucksache(
|
||||
drucksache=nummer,
|
||||
title=doc.get("titel", ""),
|
||||
fraktionen=fraktionen,
|
||||
datum=doc.get("datum", ""),
|
||||
link=pdf_url,
|
||||
bundesland=self.bundesland,
|
||||
typ=doc.get("drucksachetyp", "Antrag"),
|
||||
)
|
||||
|
||||
async def _fetch_page(
|
||||
self, client: httpx.AsyncClient, *, cursor: Optional[str] = None,
|
||||
) -> tuple[list[dict], Optional[str]]:
|
||||
"""Lade eine Page vom /drucksache-Endpoint. Returns (docs, next_cursor)."""
|
||||
params = {
|
||||
"apikey": self.apikey,
|
||||
"f.drucksachetyp": self.document_typ,
|
||||
"f.wahlperiode": str(self.wahlperiode),
|
||||
}
|
||||
if cursor:
|
||||
params["cursor"] = cursor
|
||||
try:
|
||||
resp = await client.get(f"{self.base_url}/drucksache", params=params)
|
||||
if resp.status_code != 200:
|
||||
logger.error("BUND DIP HTTP %s: %s", resp.status_code, resp.text[:200])
|
||||
return [], None
|
||||
data = resp.json()
|
||||
return data.get("documents", []), data.get("cursor")
|
||||
except Exception:
|
||||
logger.exception("BUND DIP request error")
|
||||
return [], None
|
||||
|
||||
async def search(self, query: str, limit: int = 20) -> list[Drucksache]:
|
||||
"""Liste die neuesten Anträge der konfigurierten Wahlperiode.
|
||||
|
||||
Server liefert Antrags-gefiltert + nach Aktualität sortiert; wir
|
||||
paginieren über cursor bis ``limit`` (oder das Ende der Periode)
|
||||
erreicht ist. Query wird client-side als Title-Substring-Filter
|
||||
angewandt — die DIP-API hat einen ``f.titel``-Filter, aber für
|
||||
Konsistenz mit den Landtags-Adaptern (alle nutzen client-side
|
||||
Filter wegen Schema-Drift) machen wir es hier auch so.
|
||||
"""
|
||||
results: list[Drucksache] = []
|
||||
seen: set[str] = set()
|
||||
query_terms = [t.lower() for t in query.split() if t] if query else []
|
||||
|
||||
async with self._make_client() as client:
|
||||
cursor: Optional[str] = None
|
||||
for _ in range(20): # max 20 pages = 2000 docs als Hard-Cap
|
||||
docs, next_cursor = await self._fetch_page(client, cursor=cursor)
|
||||
if not docs:
|
||||
break
|
||||
for raw in docs:
|
||||
doc = self._doc_to_drucksache(raw)
|
||||
if not doc:
|
||||
continue
|
||||
if doc.drucksache in seen:
|
||||
continue
|
||||
seen.add(doc.drucksache)
|
||||
if query_terms:
|
||||
hay = doc.title.lower()
|
||||
if not all(t in hay for t in query_terms):
|
||||
continue
|
||||
results.append(doc)
|
||||
if len(results) >= limit:
|
||||
return results
|
||||
# Cursor unverändert → letzte Page erreicht
|
||||
if not next_cursor or next_cursor == cursor:
|
||||
break
|
||||
cursor = next_cursor
|
||||
|
||||
return results
|
||||
|
||||
async def get_document(self, drucksache: str) -> Optional[Drucksache]:
|
||||
"""Look up a single Drucksache by ID. Nutzt den f.dokumentnummer-
|
||||
Filter — direkter Treffer ohne Pagination."""
|
||||
async with self._make_client() as client:
|
||||
try:
|
||||
resp = await client.get(
|
||||
f"{self.base_url}/drucksache",
|
||||
params={
|
||||
"apikey": self.apikey,
|
||||
"f.dokumentnummer": drucksache,
|
||||
"f.wahlperiode": str(self.wahlperiode),
|
||||
},
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
return None
|
||||
docs = resp.json().get("documents", [])
|
||||
for raw in docs:
|
||||
if raw.get("dokumentnummer") == drucksache:
|
||||
return self._doc_to_drucksache(raw)
|
||||
except Exception:
|
||||
logger.exception("BUND get_document error for %s", drucksache)
|
||||
return None
|
||||
|
||||
async def download_text(self, drucksache: str) -> Optional[str]:
|
||||
"""Download das Drucksachen-PDF und extrahiere Volltext."""
|
||||
import fitz
|
||||
|
||||
doc = await self.get_document(drucksache)
|
||||
if not doc 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:
|
||||
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("BUND download error for %s", drucksache)
|
||||
return None
|
||||
|
||||
|
||||
# Registry of adapters
|
||||
ADAPTERS = {
|
||||
"BUND": BundestagAdapter(),
|
||||
"NRW": NRWAdapter(),
|
||||
"LSA": PortalaAdapter(
|
||||
bundesland="LSA",
|
||||
|
||||
@ -43,11 +43,13 @@ class TestWahlperiodeFor:
|
||||
|
||||
def test_all_wahlperioden_lists_each_bl_twice(self):
|
||||
out = all_wahlperioden()
|
||||
# 16 Bundesländer × 2 WPs = 32 Einträge
|
||||
assert len(out) == 32
|
||||
# 16 Bundesländer + BUND × 2 WPs = 34 Einträge (#56 fügt BUND hinzu)
|
||||
assert len(out) == 34
|
||||
# Aktuelle und vorherige WP für NRW
|
||||
assert "NRW-WP18" in out
|
||||
assert "NRW-WP17" in out
|
||||
# BUND ist auch dabei
|
||||
assert "BUND-WP21" in out
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
@ -1,4 +1,9 @@
|
||||
"""Tests for bundeslaender.py — sanity over 16-state registry.
|
||||
"""Tests for bundeslaender.py — sanity over the parliament registry.
|
||||
|
||||
Stand: 16 Bundesländer + Sondereintrag ``BUND`` für den Deutschen
|
||||
Bundestag (siehe #56). Der ``Bundesland``-Dataclass-Name ist historisch
|
||||
und wird inzwischen als generischer Parlament-Slot verwendet — eine
|
||||
Umbenennung ist als separates Refactoring-Item geplant.
|
||||
|
||||
Includes the #48 classification regression: TH must be ParlDok, HB must
|
||||
be StarWeb, SN must be Eigensystem (not ParlDok).
|
||||
@ -7,8 +12,10 @@ from app.bundeslaender import BUNDESLAENDER, get, aktive_bundeslaender, alle_bun
|
||||
|
||||
|
||||
class TestRegistryStructure:
|
||||
def test_sixteen_bundeslaender(self):
|
||||
assert len(BUNDESLAENDER) == 16
|
||||
def test_sixteen_bundeslaender_plus_bund(self):
|
||||
# 16 echte Bundesländer + BUND-Sondereintrag (#56)
|
||||
assert len(BUNDESLAENDER) == 17
|
||||
assert "BUND" in BUNDESLAENDER
|
||||
|
||||
def test_codes_are_uppercase(self):
|
||||
for code in BUNDESLAENDER:
|
||||
@ -33,8 +40,9 @@ class TestActiveBundeslaender:
|
||||
original = {"NRW", "LSA", "MV", "BE"}
|
||||
assert original <= active_codes
|
||||
|
||||
def test_alle_bundeslaender_returns_all_sixteen(self):
|
||||
assert len(alle_bundeslaender()) == 16
|
||||
def test_alle_bundeslaender_returns_all(self):
|
||||
# 16 BL + BUND
|
||||
assert len(alle_bundeslaender()) == 17
|
||||
|
||||
def test_alle_bundeslaender_active_first(self):
|
||||
out = alle_bundeslaender()
|
||||
|
||||
Loading…
Reference in New Issue
Block a user