"""Erhebung 2026, Schritt 1: SHAB-Meldungen abrufen und als Rohdaten speichern. Setzt die Methodik-Seite (dentkompass.com/erhebung-2026/methodik) um: - Rubriken HR und KK getrennt abrufen - neun Suchwoerter ueber den Parameter `keyword` (Volltext) - Zeitfenster nach Publikationsdatum: 1.9.2025 bis 31.8.2026 - Deutschschweizer Kantone - jede Meldung als XML nachladen Ergebnis: erhebung-2026/rohdaten/meldungen.csv plus ein XML pro Meldung. Die Klassifikation (Praxis, Labor, ansprechbar) ist ein eigener Schritt. Aufruf aus dem Ordner backend: python scripts/erhebung_shab_abruf.py """ from __future__ import annotations import csv import json import re import time import xml.etree.ElementTree as ET from datetime import datetime, timezone from pathlib import Path import httpx SHAB_URL = "https://www.shab.ch/api/v1/publications" FENSTER_START = "2025-09-01" FENSTER_ENDE = "2026-08-31" SUCHWOERTER = ( "Zahnarztpraxis", "Zahnmedizin", "zahnärztliche", "Kieferorthopädie", "Dentalhygiene", "Zahntechnik", "Dentallabor", "Zahnklinik", "Zahnärzte", ) RUBRIKEN = ("HR", "KK") DEUTSCHSCHWEIZ = { "ZH", "BE", "LU", "UR", "SZ", "OW", "NW", "GL", "ZG", "SO", "BS", "BL", "SH", "AR", "AI", "SG", "GR", "AG", "TG", } SEITENGROESSE = 100 WURZEL = Path(__file__).resolve().parents[2] / "erhebung-2026" / "rohdaten" XML_ORDNER = WURZEL / "xml" CSV_DATEI = WURZEL / "meldungen.csv" PROTOKOLL = WURZEL / "abruf-protokoll.json" SPALTEN = [ "id", "publikationsnummer", "rubrik", "unterrubrik", "publikationsdatum", "erstellungsdatum", "kantone", "titel", "firma", "uid", "sitz", "rechtsform", "zweck", "publikationstext", "suchwoerter", "url", ] def mit_wiederholung(client: httpx.Client, url: str, **kwargs) -> httpx.Response: for versuch in range(4): try: antwort = client.get(url, **kwargs) antwort.raise_for_status() return antwort except httpx.HTTPError: if versuch == 3: raise time.sleep(2 * (versuch + 1)) raise RuntimeError("unerreichbar") def suche(client: httpx.Client, rubrik: str, wort: str) -> list[dict]: """Alle Treffer eines Suchworts in einer Rubrik, ueber alle Seiten.""" treffer: list[dict] = [] seite = 0 while True: antwort = mit_wiederholung( client, SHAB_URL, params={ "keyword": wort, "rubrics": rubrik, "publicationStates": "PUBLISHED", "publicationDate.start": FENSTER_START, "publicationDate.end": FENSTER_ENDE, "pageRequest.size": SEITENGROESSE, "pageRequest.page": seite, }, ) daten = antwort.json() inhalt = daten.get("content") or [] treffer.extend(inhalt) if len(treffer) >= int(daten.get("total") or 0) or not inhalt: break seite += 1 return treffer def text(element: ET.Element | None, pfad: str) -> str: if element is None: return "" gefunden = element.find(pfad) if gefunden is None or not gefunden.text: return "" return gefunden.text.strip() def felder_aus_xml(xml_text: str) -> dict[str, str]: """Firma, Sitz, Zweck usw. aus dem XML. Namespaces werden entfernt, damit der Pfad fuer HR01, HR02, HR03 und KK gleich bleibt.""" bereinigt = re.sub(r'\sxmlns(:\w+)?="[^"]+"', "", xml_text) bereinigt = re.sub(r'\s\w+:\w+="[^"]*"', "", bereinigt) bereinigt = re.sub(r"<(/?)\w+:", r"<\1", bereinigt) wurzel = ET.fromstring(bereinigt) inhalt = wurzel.find("content") firma = None if inhalt is not None: firma = inhalt.find("commonsNew/company") if firma is None: firma = inhalt.find("commonsActual/company") zweck = text(inhalt, "commonsNew/purpose") or text(inhalt, "commonsActual/purpose") publikationstext = text(inhalt, "publicationText") if not publikationstext and inhalt is not None: # KK-Meldungen haben keinen publicationText, dafuer Schuldnerangaben publikationstext = " ".join(t.strip() for t in inhalt.itertext() if t and t.strip())[:3000] return { "firma": text(firma, "name"), "uid": text(firma, "uid"), "sitz": text(firma, "seat"), "rechtsform": text(firma, "legalForm"), "zweck": zweck, "publikationstext": publikationstext, } def main() -> None: XML_ORDNER.mkdir(parents=True, exist_ok=True) start = datetime.now(timezone.utc) meldungen: dict[str, dict] = {} zaehlung: dict[str, int] = {} with httpx.Client(timeout=30.0) as client: for rubrik in RUBRIKEN: for wort in SUCHWOERTER: treffer = suche(client, rubrik, wort) zaehlung[f"{rubrik}:{wort}"] = len(treffer) print(f"{rubrik} {wort:<20} {len(treffer):>5} Treffer") for eintrag in treffer: meta = eintrag.get("meta") or {} pid = str(meta.get("id") or "") if not pid: continue kantone = [str(k).upper() for k in (meta.get("cantons") or [])] if not DEUTSCHSCHWEIZ.intersection(kantone): continue if pid in meldungen: meldungen[pid]["suchwoerter"].add(wort) continue meldungen[pid] = { "id": pid, "publikationsnummer": str(meta.get("publicationNumber") or ""), "rubrik": str(meta.get("rubric") or ""), "unterrubrik": str(meta.get("subRubric") or ""), "publikationsdatum": str(meta.get("publicationDate") or "")[:10], "erstellungsdatum": str(meta.get("creationDate") or "")[:10], "kantone": "|".join(kantone), "titel": str(((meta.get("title") or {}).get("de")) or ""), "suchwoerter": {wort}, "url": f"https://www.shab.ch/#!/search/publications/detail/{pid}", } time.sleep(0.3) print(f"\n{len(meldungen)} Meldungen aus der Deutschschweiz, lade XML ...") for n, (pid, meldung) in enumerate(meldungen.items(), start=1): ziel = XML_ORDNER / f"{pid}.xml" if ziel.is_file(): xml_text = ziel.read_text(encoding="utf-8") else: xml_text = mit_wiederholung(client, f"{SHAB_URL}/{pid}/xml").text ziel.write_text(xml_text, encoding="utf-8") time.sleep(0.2) meldung.update(felder_aus_xml(xml_text)) if n % 50 == 0: print(f" {n}/{len(meldungen)}") zeilen = sorted(meldungen.values(), key=lambda m: (m["publikationsdatum"], m["id"])) with CSV_DATEI.open("w", newline="", encoding="utf-8") as f: schreiber = csv.DictWriter(f, fieldnames=SPALTEN) schreiber.writeheader() for zeile in zeilen: zeile["suchwoerter"] = "|".join(sorted(zeile["suchwoerter"])) schreiber.writerow({k: zeile.get(k, "") for k in SPALTEN}) nach_rubrik: dict[str, int] = {} for zeile in zeilen: schluessel = f"{zeile['rubrik']}/{zeile['unterrubrik']}" nach_rubrik[schluessel] = nach_rubrik.get(schluessel, 0) + 1 PROTOKOLL.write_text( json.dumps( { "abruf_begonnen": start.isoformat(timespec="seconds"), "abruf_beendet": datetime.now(timezone.utc).isoformat(timespec="seconds"), "fenster": [FENSTER_START, FENSTER_ENDE], "suchwoerter": list(SUCHWOERTER), "kantone": sorted(DEUTSCHSCHWEIZ), "treffer_je_suchwort_vor_kantonsfilter": zaehlung, "meldungen_deutschschweiz": len(zeilen), "nach_rubrik": nach_rubrik, }, ensure_ascii=False, indent=2, ), encoding="utf-8", ) print(f"\nFertig: {len(zeilen)} Meldungen in {CSV_DATEI}") for k, v in sorted(nach_rubrik.items()): print(f" {k:<10} {v}") if __name__ == "__main__": main()