from __future__ import annotations from typing import Any import xml.etree.ElementTree as ET import pandas as pd NAMESPACES = {"ehd": "urn:ehd/001", "go": "urn:ehd/go/001"} def get_text_content(elem: ET.Element | None) -> str | None: """Extract all text recursively from an XML element.""" if elem is None: return None text = " ".join(t.strip() for t in elem.itertext() if t.strip()) return text if text else None def parse_ebm_xml_to_dataframe(xml_path: str) -> pd.DataFrame: """Parse EBM XML into a pandas DataFrame with one row per GNR.""" tree = ET.parse(xml_path) root = tree.getroot() rows: list[dict[str, Any]] = [] for gnr in root.findall("./ehd:body/go:gnr_liste/go:gnr", namespaces=NAMESPACES): row: dict[str, Any] = { "code": gnr.get("V"), "use": gnr.get("USE"), "valid_from": gnr.get("VT"), } legende = gnr.find("./go:allgemein/go:legende", namespaces=NAMESPACES) if legende is not None: kurztext = legende.find("go:kurztext", namespaces=NAMESPACES) quittungstext = legende.find("go:quittungstext", namespaces=NAMESPACES) langtext = legende.find("go:langtext", namespaces=NAMESPACES) kap_bez = legende.find("go:kap_bez", namespaces=NAMESPACES) row["short_text"] = kurztext.get("V") if kurztext is not None else None row["receipt_text"] = quittungstext.get("V") if quittungstext is not None else None row["long_text"] = get_text_content(langtext) if kap_bez is not None: bereich = kap_bez.find("go:bereich", namespaces=NAMESPACES) kapitel = kap_bez.find("go:kapitel", namespaces=NAMESPACES) abschnitt = kap_bez.find("go:abschnitt", namespaces=NAMESPACES) row["chapter_code"] = kap_bez.get("V") row["chapter_name"] = kap_bez.get("DN") row["bereich"] = bereich.get("DN") if bereich is not None else None row["kapitel"] = kapitel.get("DN") if kapitel is not None else None row["abschnitt"] = abschnitt.get("DN") if abschnitt is not None else None row["service_period"] = ( gnr.find("./go:allgemein/go:gueltigkeit/go:service_tmr", namespaces=NAMESPACES).get("V") if gnr.find("./go:allgemein/go:gueltigkeit/go:service_tmr", namespaces=NAMESPACES) is not None else None ) row["effective_period"] = ( gnr.find("./go:allgemein/go:gueltigkeit/go:effective_tmr", namespaces=NAMESPACES).get("V") if gnr.find("./go:allgemein/go:gueltigkeit/go:effective_tmr", namespaces=NAMESPACES) is not None else None ) notes: list[str] = [] for note in gnr.findall("./go:allgemein/go:anmerkungen_liste/go:anmerkung", namespaces=NAMESPACES): txt = get_text_content(note) if txt: notes.append(txt) row["notes"] = notes bewertung = gnr.find("./go:allgemein/go:bewertung_liste/go:bewertung", namespaces=NAMESPACES) if bewertung is not None: row["points"] = bewertung.get("V") row["unit"] = bewertung.get("U") lt = bewertung.find("go:leistung_typ", namespaces=NAMESPACES) row["leistung_typ"] = lt.get("V") if lt is not None else None fachgruppen: list[str] = [] for fg in gnr.findall(".//go:fachgruppe_liste//go:fachgruppe", namespaces=NAMESPACES): value = fg.get("V") if value: fachgruppen.append(value) row["fachgruppen"] = fachgruppen exclusions: list[dict[str, str | None]] = [] for ex in gnr.findall("./go:regel/go:ausschluss_liste/go:bezugsraum", namespaces=NAMESPACES): for ex_gnr in ex.findall("./go:gnr_liste/go:gnr", namespaces=NAMESPACES): exclusions.append( { "code": ex_gnr.get("V"), "description": ex_gnr.get("DN"), } ) row["exclusions"] = exclusions gkv_types: list[str] = [] for gkv in gnr.findall("./go:vdx/go:gkv_kontenart_liste/go:gkv_kontenart", namespaces=NAMESPACES): value = gkv.get("V") if value: gkv_types.append(value) row["gkv_account_types"] = gkv_types rows.append(row) return pd.DataFrame(rows) def filter_df_by_fachgruppe(df: pd.DataFrame, fachgruppe: str = "001") -> pd.DataFrame: """Filter DataFrame to include only rows where the specified Fachgruppe is present.""" return df[df["fachgruppen"].apply(lambda x: isinstance(x, list) and fachgruppe in x)].reset_index(drop=True)