Michael Hermann
2bb70ae0ed
Historie 1949–2022 bleibt in lawgit, bis der GPU-Lauf endet; Schnitt ohne Lücke. Co-authored-by: Cursor <cursoragent@cursor.com>
447 lines
15 KiB
Python
447 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Strukturiert BGBl-Texte zu Ereignissen der Normzeitachse.
|
|
|
|
Standard: Heuristik (kein GPU, keine Kosten).
|
|
Optional: --llm (LLM_BASE_URL in .env) — nicht starten, solange lawgit den
|
|
Historien-Parse auf RunPod fährt.
|
|
|
|
python3 parse_bgbl.py --year 2025
|
|
python3 parse_bgbl.py --year 2025 --llm # nur bewusst, kostet GPU-Zeit
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
from bgbl_util import META_DIR, PARSED_DIR, TEXT_DIR, fundstelle
|
|
|
|
TYP_ERLASS = "Erlass"
|
|
TYP_AENDERUNG = "Änderung"
|
|
TYP_NEUBEKANNT = "Neubekanntmachung"
|
|
TYP_AUFHEBUNG = "Aufhebung"
|
|
TYP_VERORDNUNG = "Verordnung"
|
|
TYP_BEKANNT = "Bekanntmachung"
|
|
TYP_SONSTIGE = "Sonstige"
|
|
|
|
ALLE_TYPEN = {
|
|
TYP_ERLASS, TYP_AENDERUNG, TYP_NEUBEKANNT, TYP_AUFHEBUNG,
|
|
TYP_VERORDNUNG, TYP_BEKANNT, TYP_SONSTIGE,
|
|
}
|
|
|
|
STAMMGESETZE = [
|
|
(r"Grundgesetz", "GG", "gg"),
|
|
(r"Bürgerlichen Gesetzbuchs|Bürgerliche(?:s)? Gesetzbuch", "BGB", "bgb"),
|
|
(r"Strafgesetzbuch", "StGB", "stgb"),
|
|
(r"Zivilprozessordnung", "ZPO", "zpo"),
|
|
(r"Strafprozessordnung", "StPO", "stpo"),
|
|
(r"Handelsgesetzbuch", "HGB", "hgb"),
|
|
(r"Aktiengesetz", "AktG", "aktg"),
|
|
(r"GmbH-Gesetz|Gesetz betreffend die Gesellschaften mit beschränkter Haftung", "GmbHG", "gmbhg"),
|
|
(r"Verwaltungsverfahrensgesetz", "VwVfG", "vwvfg"),
|
|
(r"Verwaltungsgerichtsordnung", "VwGO", "vwgo"),
|
|
(r"Abgabenordnung", "AO", "ao"),
|
|
(r"Einkommensteuergesetz", "EStG", "estg"),
|
|
(r"Umsatzsteuergesetz", "UStG", "ustg"),
|
|
(r"Sozialgesetzbuch", "SGB", None),
|
|
(r"Aufenthaltsgesetz", "AufenthG", "aufenthg"),
|
|
(r"Asylgesetz", "AsylG", "asylvfg_1992"),
|
|
(r"Bundes-Immissionsschutzgesetz", "BImSchG", "bimschg"),
|
|
(r"Straßenverkehrsgesetz", "StVG", "stvg"),
|
|
(r"Straßenverkehrs-Ordnung", "StVO", "stvo_2013"),
|
|
]
|
|
|
|
INKRAFT_RE = re.compile(
|
|
r"(?:tritt|treten)\s+"
|
|
r"(?:am\s+)?"
|
|
r"("
|
|
r"mit\s+Ablauf\s+des\s+Tages\s+der\s+Verkündung"
|
|
r"|Tage?s?\s+der\s+Verkündung"
|
|
r"|Tage?s?\s+nach\s+(?:der\s+)?Verkündung"
|
|
r"|\d{1,2}\.\s*(?:Januar|Februar|März|April|Mai|Juni|Juli|August|September|Oktober|November|Dezember)\s+\d{4}"
|
|
r"|\d{1,2}\.\s*\d{1,2}\.\s*\d{4}"
|
|
r")"
|
|
r"[^.]*?in\s+Kraft",
|
|
re.IGNORECASE | re.DOTALL,
|
|
)
|
|
|
|
DATE_DE = re.compile(
|
|
r"(\d{1,2})\.\s*(Januar|Februar|März|April|Mai|Juni|Juli|August|September|Oktober|November|Dezember)\s+(\d{4})"
|
|
)
|
|
DATE_NUM = re.compile(r"(\d{1,2})\.\s*(\d{1,2})\.\s*(\d{4})")
|
|
|
|
MONTHS = {
|
|
"januar": 1, "februar": 2, "märz": 3, "april": 4, "mai": 5, "juni": 6,
|
|
"juli": 7, "august": 8, "september": 9, "oktober": 10, "november": 11, "dezember": 12,
|
|
}
|
|
|
|
SYSTEM_PROMPT = """Du bist ein Parser für das deutsche Bundesgesetzblatt.
|
|
Aus dem Text einer Verkündung extrahierst du strukturierte JSON-Daten für eine Normzeitachse.
|
|
Antworte NUR mit einem JSON-Objekt.
|
|
|
|
Schema:
|
|
{
|
|
"titel": "amtlicher Titel",
|
|
"typ": "Erlass|Änderung|Neubekanntmachung|Aufhebung|Verordnung|Bekanntmachung|Sonstige",
|
|
"ausfertigung": "YYYY-MM-DD oder null",
|
|
"inkrafttreten": [
|
|
{"datum": "YYYY-MM-DD oder 'Tag nach der Verkündung'", "geltung": "ganzes Gesetz oder z.B. Art. 2"}
|
|
],
|
|
"betroffene_normen": [
|
|
{
|
|
"name": "Bürgerliches Gesetzbuch",
|
|
"kurz": "BGB",
|
|
"operation": "erlass|aenderung|aufhebung|neubekanntmachung",
|
|
"aenderungen": [
|
|
{"stelle": "§ 433", "aktion": "neufassung|einfuegung|streichung|aenderung|wortlaut", "kurzbeschreibung": "ein Satz"}
|
|
]
|
|
}
|
|
],
|
|
"zusammenfassung": "1-2 Sätze, verständlich für Juristinnen ohne IT-Hintergrund"
|
|
}
|
|
|
|
Regeln:
|
|
- Änderungsgesetze ändern oft MEHRERE Stammgesetze (Artikelgesetz). Jedes Stammgesetz einzeln auflisten.
|
|
- Stelle immer mit § oder Art. angeben.
|
|
- Keine Git-Begriffe. Keine Spekulation: unsichere Felder null / leere Liste.
|
|
"""
|
|
|
|
|
|
def normalize_ocr(text: str) -> str:
|
|
t = text.replace("\u00ad", "")
|
|
t = re.sub(r"-\s*\n\s*", "", t)
|
|
t = re.sub(r"(?<=[A-Za-zÄÖÜäöüß])\n(?=[a-zäöüß])", "", t)
|
|
t = t.replace("~", "m")
|
|
t = re.sub(r"\s+", " ", t)
|
|
return t
|
|
|
|
|
|
def parse_date_de(text: str) -> str | None:
|
|
m = DATE_DE.search(text)
|
|
if m:
|
|
d, mon, y = int(m.group(1)), MONTHS[m.group(2).lower()], int(m.group(3))
|
|
return f"{y:04d}-{mon:02d}-{d:02d}"
|
|
m = DATE_NUM.search(text)
|
|
if m:
|
|
d, mo, y = int(m.group(1)), int(m.group(2)), int(m.group(3))
|
|
if 1 <= mo <= 12 and 1 <= d <= 31:
|
|
return f"{y:04d}-{mo:02d}-{d:02d}"
|
|
return None
|
|
|
|
|
|
def classify_typ(title: str) -> str:
|
|
t = title.lower()
|
|
if "bekanntmachung der neufassung" in t or "neufassung" in t:
|
|
return TYP_NEUBEKANNT
|
|
if "aufhebung" in t:
|
|
return TYP_AUFHEBUNG
|
|
if "änderung" in t or "aenderung" in t:
|
|
return TYP_AENDERUNG
|
|
if t.startswith("verordnung") or " verordnung " in f" {t} ":
|
|
return TYP_VERORDNUNG
|
|
if t.startswith("bekanntmachung") or "bekanntmachung" in t:
|
|
return TYP_BEKANNT
|
|
if t.startswith("gesetz") or "gesetz" in t:
|
|
return TYP_ERLASS
|
|
return TYP_SONSTIGE
|
|
|
|
|
|
def _norm_hit(kurz: str, gii: str | None, operation: str) -> dict:
|
|
return {"name": kurz, "kurz": kurz, "gii": gii, "operation": operation, "aenderungen": []}
|
|
|
|
|
|
def find_betroffene(title: str, text: str, typ: str) -> list[dict]:
|
|
found: list[dict] = []
|
|
seen: set[str] = set()
|
|
op_title = "aenderung" if typ == TYP_AENDERUNG else "erlass"
|
|
for pattern, kurz, gii in STAMMGESETZE:
|
|
if re.search(pattern, title, re.IGNORECASE) and kurz not in seen:
|
|
seen.add(kurz)
|
|
found.append(_norm_hit(kurz, gii, op_title))
|
|
|
|
head = text[:8000]
|
|
for pattern, kurz, gii in STAMMGESETZE:
|
|
if kurz in seen:
|
|
continue
|
|
aend = re.search(
|
|
rf"(?:Änderung|Anderung|Abänderung)\s+(?:des|der|von)\s+(?:[^\n.]{{0,40}})?{pattern}",
|
|
head,
|
|
re.IGNORECASE,
|
|
)
|
|
geaendert = re.search(
|
|
rf"{pattern}[^.{{]{{0,120}}wird wie folgt geändert",
|
|
head,
|
|
re.IGNORECASE,
|
|
)
|
|
if aend or geaendert:
|
|
seen.add(kurz)
|
|
found.append(_norm_hit(kurz, gii, "aenderung"))
|
|
return found
|
|
|
|
|
|
def find_inkrafttreten(text: str, verkundung: str | None) -> list[dict]:
|
|
hits = []
|
|
seen = set()
|
|
blob = normalize_ocr(text)
|
|
for m in INKRAFT_RE.finditer(blob):
|
|
raw = re.sub(r"\s+", " ", m.group(1)).strip()
|
|
low = raw.lower()
|
|
hinweis = None
|
|
datum: str | None = None
|
|
if "verkündung" in low:
|
|
if verkundung:
|
|
try:
|
|
d = datetime.fromisoformat(verkundung[:10])
|
|
if "ablauf" in low:
|
|
datum = (d + timedelta(days=1)).strftime("%Y-%m-%d")
|
|
hinweis = "mit Ablauf des Verkündungstages, gilt ab dem Folgetag"
|
|
elif "nach" in low:
|
|
datum = (d + timedelta(days=1)).strftime("%Y-%m-%d")
|
|
hinweis = "Tag nach der Verkündung"
|
|
else:
|
|
datum = d.strftime("%Y-%m-%d")
|
|
hinweis = "Tag der Verkündung"
|
|
except ValueError:
|
|
datum = raw
|
|
else:
|
|
datum = raw
|
|
else:
|
|
datum = parse_date_de(raw) or raw
|
|
key = (datum, hinweis)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
rec = {"datum": datum, "geltung": "ganzes Gesetz"}
|
|
if hinweis:
|
|
rec["hinweis"] = hinweis
|
|
hits.append(rec)
|
|
return hits
|
|
|
|
|
|
def load_meta() -> list[dict]:
|
|
items = []
|
|
if not META_DIR.exists():
|
|
return items
|
|
seen: set[str] = set()
|
|
for path in sorted(META_DIR.glob("*.json")):
|
|
if path.name in {"offenegesetze-quelle.json"}:
|
|
continue
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(data, list):
|
|
continue
|
|
for item in data:
|
|
key = item.get("id")
|
|
if not key or key in seen:
|
|
continue
|
|
seen.add(key)
|
|
items.append(item)
|
|
return items
|
|
|
|
|
|
def _load_extracted(item: dict) -> dict | None:
|
|
kind = item.get("kind") or "bgbl1"
|
|
year = item.get("year")
|
|
number = item.get("number_raw") or item.get("number")
|
|
candidates = [
|
|
TEXT_DIR / kind / str(year) / f"{kind}_{year}_{number}.json",
|
|
]
|
|
doc = (item.get("document_url") or "").split("#")[0]
|
|
if doc:
|
|
name = Path(doc).name
|
|
candidates.append(TEXT_DIR / kind / str(year) / name.replace(".pdf", ".json"))
|
|
candidates.append(TEXT_DIR / kind / str(year) / Path(name).with_suffix(".json").name)
|
|
pdf = item.get("pdf")
|
|
if pdf:
|
|
candidates.append(TEXT_DIR / kind / str(year) / Path(pdf).with_suffix(".json").name)
|
|
for c in candidates:
|
|
if c.exists():
|
|
return json.loads(c.read_text(encoding="utf-8"))
|
|
return None
|
|
|
|
|
|
def text_for_item(item: dict) -> dict | None:
|
|
extracted = _load_extracted(item)
|
|
if not extracted:
|
|
return None
|
|
page_texts = extracted.get("page_texts") or []
|
|
if not page_texts:
|
|
return extracted if extracted.get("text") else None
|
|
start = item.get("pdf_page") or 1
|
|
n = item.get("num_pages") or 0
|
|
seiten = [p for p in page_texts if start <= p["page"] < start + n] if n else page_texts
|
|
return {
|
|
**extracted,
|
|
"text": "\n\n".join(f"--- Seite {p['page']} ---\n{p['text']}" for p in seiten),
|
|
"pages": len(seiten),
|
|
}
|
|
|
|
|
|
def heuristic_parse(item: dict, extracted: dict | None) -> dict:
|
|
title = item.get("title") or item.get("titel") or ""
|
|
text = (extracted or {}).get("text") or ""
|
|
date = (item.get("date") or item.get("verkundung") or "")[:10] or None
|
|
kind = item.get("kind") or "bgbl1"
|
|
year = item.get("year")
|
|
number = item.get("number_raw") or item.get("number")
|
|
page = item.get("page")
|
|
fs = item.get("fundstelle") or fundstelle(kind, year or 0, number or 0, page)
|
|
|
|
typ = classify_typ(title)
|
|
body = normalize_ocr(text)
|
|
if typ == TYP_ERLASS and re.search(r"wird wie folgt geändert", body, re.IGNORECASE):
|
|
typ = TYP_AENDERUNG
|
|
return {
|
|
"id": item.get("id"),
|
|
"fundstelle": fs,
|
|
"verkundung": date,
|
|
"titel": title,
|
|
"typ": typ,
|
|
"ausfertigung": item.get("ausfertigung") or parse_date_de(title) or date,
|
|
"inkrafttreten": find_inkrafttreten(text, date),
|
|
"betroffene_normen": find_betroffene(title, text, typ),
|
|
"zusammenfassung": title,
|
|
"pdf": (extracted or {}).get("pdf") or item.get("pdf"),
|
|
"text_quality": (extracted or {}).get("quality"),
|
|
"quelle": item.get("quelle") or ("offenegesetze" if year and year < 2023 else "recht.bund.de"),
|
|
"parser": "heuristik",
|
|
"kind": kind,
|
|
"year": year,
|
|
"number": number,
|
|
"page": page,
|
|
"url": item.get("url"),
|
|
"document_url": item.get("document_url"),
|
|
"sha256": item.get("sha256"),
|
|
}
|
|
|
|
|
|
def llm_parse(item: dict, extracted: dict, base: dict) -> dict:
|
|
import llm_client
|
|
|
|
meta = json.dumps(
|
|
{k: item.get(k) for k in ("id", "title", "titel", "date", "year", "number", "kind")},
|
|
ensure_ascii=False,
|
|
)
|
|
text = (extracted.get("text") or "")[:18000]
|
|
daten = llm_client.chat_json(
|
|
[
|
|
{"role": "system", "content": SYSTEM_PROMPT},
|
|
{"role": "user", "content": f"Metadaten: {meta}\n\nText der Verkündung:\n{text}"},
|
|
],
|
|
max_tokens=4000,
|
|
)
|
|
event = dict(base)
|
|
for feld in ("titel", "typ", "ausfertigung", "inkrafttreten", "zusammenfassung", "betroffene_normen"):
|
|
if daten.get(feld):
|
|
if feld == "typ" and daten[feld] not in ALLE_TYPEN:
|
|
continue
|
|
event[feld] = daten[feld]
|
|
event["parser"] = "llm"
|
|
return event
|
|
|
|
|
|
def parse_item(item: dict, use_llm: bool) -> dict:
|
|
extracted = text_for_item(item)
|
|
event = heuristic_parse(item, extracted)
|
|
if use_llm and extracted and extracted.get("text"):
|
|
try:
|
|
return llm_parse(item, extracted, event)
|
|
except Exception as exc:
|
|
event["parser"] = "heuristik"
|
|
event["parser_fehler"] = [f"{type(exc).__name__}: {exc}"]
|
|
return event
|
|
|
|
|
|
def _schreibe(event: dict) -> Path:
|
|
dest = PARSED_DIR / f"{event['id']}.json"
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_text(json.dumps(event, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
return dest
|
|
|
|
|
|
def baue_index() -> Path:
|
|
alle = []
|
|
for f in sorted(PARSED_DIR.glob("bgbl*.json")):
|
|
try:
|
|
alle.append(json.loads(f.read_text(encoding="utf-8")))
|
|
except (json.JSONDecodeError, OSError) as exc:
|
|
print(f" übersprungen: {f.name} ({exc})")
|
|
alle.sort(key=lambda e: (e.get("verkundung") or "", e.get("id") or ""))
|
|
index = PARSED_DIR / "index.json"
|
|
index.write_text(json.dumps(alle, ensure_ascii=False), encoding="utf-8")
|
|
return index
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="BGBl-Texte zu Zeitachsen-Ereignissen parsen")
|
|
parser.add_argument("--year", type=int, action="append")
|
|
parser.add_argument("--von", type=int)
|
|
parser.add_argument("--bis", type=int)
|
|
parser.add_argument(
|
|
"--llm",
|
|
action="store_true",
|
|
help="RunPod/vLLM verwenden (LLM_BASE_URL). Standard ist Heuristik — --llm nicht "
|
|
"starten, solange der lawgit-Historien-Parse auf der GPU läuft.",
|
|
)
|
|
parser.add_argument("--limit", type=int, default=0)
|
|
parser.add_argument("--force", action="store_true")
|
|
parser.add_argument("--kein-index", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
items = load_meta()
|
|
if args.year:
|
|
years = set(args.year)
|
|
items = [i for i in items if i.get("year") in years]
|
|
if args.von:
|
|
items = [i for i in items if (i.get("year") or 0) >= args.von]
|
|
if args.bis:
|
|
items = [i for i in items if (i.get("year") or 9999) <= args.bis]
|
|
items.sort(key=lambda i: (i.get("year") or 0, i.get("number") or 0, i.get("id") or ""))
|
|
if args.limit:
|
|
items = items[: args.limit]
|
|
|
|
if not items:
|
|
print("Keine Metadaten. Zuerst: python3 download_rechtbund.py --live-test")
|
|
return
|
|
|
|
use_llm = args.llm
|
|
if use_llm:
|
|
import llm_client
|
|
|
|
if not llm_client.llm_configured():
|
|
print("LLM_BASE_URL ist nicht gesetzt — falle auf Heuristik zurück.")
|
|
use_llm = False
|
|
|
|
PARSED_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
if not args.force:
|
|
offen = []
|
|
for item in items:
|
|
dest = PARSED_DIR / f"{item.get('id')}.json"
|
|
if dest.exists():
|
|
continue
|
|
offen.append(item)
|
|
if len(offen) < len(items):
|
|
print(f"{len(items) - len(offen)} bereits geparst, übersprungen")
|
|
items = offen
|
|
if not items:
|
|
print("Nichts zu tun.")
|
|
return
|
|
|
|
for i, item in enumerate(items, 1):
|
|
print(f"[{i}/{len(items)}] {item.get('id')} {(item.get('title') or item.get('titel') or '')[:70]}")
|
|
_schreibe(parse_item(item, use_llm))
|
|
|
|
if args.kein_index:
|
|
print(f"{len(items)} Ereignisse geschrieben, Index nicht neu gebaut")
|
|
return
|
|
index = baue_index()
|
|
print(f"{len(items)} in diesem Lauf → {index}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|