Spaces:
Sleeping
Sleeping
File size: 4,787 Bytes
5887b57 d491d8b 81e8d50 d491d8b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | 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)
|