#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Extract HH 2026 (Regierungsentwurf) from XML into clean JSON. - Parses hierarchy: Einzelplan -> Kapitel -> Titel - Converts values to EUR (XML uses Tsd. EUR) - Optional category tagging by keywords - Outputs: * HH2026_titel_eur.json (full titles) * EP01_24_Summen_Mrd.json (totals for EP 01–24) * EP{EP}_TopN_Titel.json (per-EP top-N list) Usage: python extract_hh2026.py --xml regierungsentwurf_2026.xml --outdir ./out --topN 50 --tag """ import argparse, json, re, os, sys from lxml import etree from collections import defaultdict CAT_RULES = { "Soziales & Arbeit": [ r"Rente", r"Rentenversicherung", r"Bundeszuschuss.*Rente", r"Arbeitslos", r"Bürgergeld", r"Grundsicherung", r"Jobcenter", r"BAföG", r"Arbeitsförderung", r"SGB" ], "Gesundheit & Pflege": [ r"Gesundheit", r"Gesundheitsfonds", r"GKV", r"Krankenkasse", r"Krankenhaus|Klinik", r"Pflege", r"Pflegeversicherung", r"Hebamme" ], "Familie & Bildung": [ r"Familie", r"Kindergeld", r"Elterngeld", r"Kita|Kindergarten", r"Schule|Bildung", r"Hochschule|Universität|Uni", r"Jugend" ], "Innere Sicherheit": [ r"Bundespolizei|Polizei|BKA|Verfassungsschutz|BfV", r"Katastrophenschutz", r"Justizvollzug|Strafvollzug" ], "Verteidigung": [ r"Bundeswehr|Verteidigung|Rüst(ung|ungs)|Marine|Heer|Luftwaffe|NATO|Beschaffung.*milit" ], "Verkehr & Infrastruktur": [ r"Straß|Autobahn|Brücke|Bahn|Schiene|ÖPNV|Verkehr", r"Digital|Breitband|Netz" ], "Subventionen & Wirtschaftsförderung": [ r"Subvention|Förderung|Zuschuss.*Unternehmen|Mittelstand|Industrie", r"Agrar|Landwirtschaft", r"Energiehilf|Wasserstoff|Batterie|Elektromobil|Ladeinfrastruktur|Start-?up|Gründer|Kohle|Bergbau|Schiffbau" ], "Öffentliche Verwaltung & Politik": [ r"Bundestag|Bundesrat|Kanzleramt|Ministerium|Behörde|Amt|Regierungspräsidium|politische.*Stiftung" ], "Zinsen & Finanzverwaltung": [ r"Zins|Bundesschuld|Schuldendienst|Anleihe|Tilgung|Finanzagentur" ], "Internationale Zusammenarbeit & Entwicklung": [ r"Entwicklungszusammenarbeit|BMZ|ODA|Auslandshilfe|UNO|Vereinte Nationen|EU-?Beitrag|Auslandszahlung|Weltbank|IWF" ], "Kultur, Religion & Medien": [ r"Kirche|Religionsgemeinschaft|Staatsleistung", r"Kultur|Theater|Museum|Filmförder", r"Rundfunk|Medien|Presse" ], "Umwelt & Klima": [ r"Umwelt|Klimaschutz|Nachhaltigkeit|Naturschutz|Energiewende|Ökologie|CO2|Emission" ], } def compile_rules(): return {cat: [re.compile(pat, re.I) for pat in pats] for cat, pats in CAT_RULES.items()} def categorize(texts): """Return best-fit category by first match priority.""" if not texts: return None blob = " | ".join([t for t in texts if t]).strip() if not blob: return None for cat, patterns in compile_rules().items(): for rx in patterns: if rx.search(blob): return cat return None def parse_xml(xml_path): parser = etree.XMLParser(recover=True, huge_tree=True, encoding="utf-8") tree = etree.parse(xml_path, parser) root = tree.getroot() records = [] for ep in root.findall(".//einzelplan"): ep_nr = ep.get("nr") ep_name = (ep.findtext("text") or "").strip() for kap in ep.findall(".//kapitel"): kap_nr = kap.get("nr") kap_name = (kap.findtext("text") or "").strip() for titel in kap.findall(".//titel"): titel_nr = titel.get("nr") fkt = titel.get("fkt") seite = titel.get("seite") bez = (titel.findtext("text") or "").strip() soll = titel.find(".//soll") wert = None if soll is not None and "wert" in soll.attrib: try: wert = float(soll.attrib["wert"].replace(",", ".")) except: wert = None if wert is None: continue rec = { "einzelplan_nr": str(ep_nr).zfill(2) if ep_nr else "", "einzelplan_name": ep_name, "kapitel_nr": kap_nr, "kapitel_name": kap_name, "titel_nr": titel_nr, "bezeichnung": bez, "seite": int(seite) if seite and re.match(r"^\d+$", str(seite)) else None, "betrag_eur": wert * 1000.0, # XML uses Tsd. EUR } records.append(rec) return records def add_categories(records, do_tag=False): if not do_tag: return records out = [] for r in records: cat = categorize([r.get("bezeichnung"), r.get("kapitel_name"), r.get("einzelplan_name")]) r2 = dict(r) r2["category"] = cat or "Sonstiges / nicht zuordenbar" out.append(r2) return out def build_ep_totals(records, ep_range=("01","24")): lo, hi = map(int, ep_range) sums = defaultdict(float) names = {} for r in records: ep = r.get("einzelplan_nr","") if not ep: continue try: epi = int(ep) except: continue if epi < int(lo) or epi > int(hi): continue sums[ep] += float(r.get("betrag_eur") or 0.0) names[ep] = r.get("einzelplan_name","") rows = [] for ep, val in sorted(sums.items(), key=lambda x: -x[1]): rows.append({ "einzelplan_nr": ep, "einzelplan_name": names.get(ep, ""), "etat_mrd_eur": val / 1e9 }) return rows def top_n_per_ep(records, N=20, outdir="."): # group by EP, sort by betrag by_ep = defaultdict(list) for r in records: ep = r.get("einzelplan_nr","") by_ep[ep].append(r) index = [] for ep, rows in by_ep.items(): rows_sorted = sorted(rows, key=lambda x: float(x.get("betrag_eur") or 0.0), reverse=True)[:N] # enrich with Mio for readability for rr in rows_sorted: rr["betrag_mio_eur"] = (float(rr["betrag_eur"]) / 1e6) if rr.get("betrag_eur") is not None else None path = os.path.join(outdir, f"EP{ep}_Top{N}_Titel.json") with open(path, "w", encoding="utf-8") as f: json.dump(rows_sorted, f, ensure_ascii=False, indent=2) index.append({"einzelplan_nr": ep, "topN_path": path}) return index def main(): ap = argparse.ArgumentParser(description="Extract Regierungsentwurf 2026 XML to JSON") ap.add_argument("--xml", required=True, help="Path to regierungsentwurf_2026.xml") ap.add_argument("--outdir", default=".", help="Output directory") ap.add_argument("--topN", type=int, default=20, help="Top-N Titel pro EP exportieren") ap.add_argument("--tag", action="store_true", help="Einfache Kategorien per Keyword zuweisen") ap.add_argument("--ep_lo", default="01", help="EP-Start (inklusive) für Totals") ap.add_argument("--ep_hi", default="24", help="EP-Ende (inklusive) für Totals") args = ap.parse_args() os.makedirs(args.outdir, exist_ok=True) # Parse XML fully records = parse_xml(args.xml) # Optional tagging records = add_categories(records, do_tag=args.tag) # Save full titles titles_json = os.path.join(args.outdir, "HH2026_titel_eur.json") with open(titles_json, "w", encoding="utf-8") as f: json.dump(records, f, ensure_ascii=False, indent=2) # Build EP totals totals = build_ep_totals(records, ep_range=(args.ep_lo, args.ep_hi)) totals_json = os.path.join(args.outdir, f"EP{args.ep_lo}_{args.ep_hi}_Summen_Mrd.json") with open(totals_json, "w", encoding="utf-8") as f: json.dump(totals, f, ensure_ascii=False, indent=2) # Build per-EP TopN index = top_n_per_ep(records, N=args.topN, outdir=args.outdir) index_json = os.path.join(args.outdir, "Ministerien_TopN_Index.json") with open(index_json, "w", encoding="utf-8") as f: json.dump(index, f, ensure_ascii=False, indent=2) # Small stdout summary total_sum = sum(float(r.get("betrag_eur") or 0.0) for r in records) print(f"[OK] Parsed titles: {len(records)}") print(f"[OK] Sum of all titles (EUR): {total_sum:,.2f}") print(f"[OK] Wrote: {titles_json}") print(f"[OK] Wrote: {totals_json}") print(f"[OK] Wrote per-EP Top{args.topN} files (index -> {index_json})") if __name__ == "__main__": main()