fegger/pv-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(kb): ingest rechtsprechung corpus

Browse Source
This commit is contained in:
Florian Egger
2026-09-15 15:27:28 +02:00
parent 1594aa9cbd
commit 5209539ba3
365 changed files with 33112 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/ingest_sources.py
+309 -5
View File
@@ -27,8 +27,11 @@ from pathlib import Path
from tools.kb_common import ( # noqa: E402
DATE_RE,
LAW_MAP,
WORK_EUGH,
WORK_KV,
WORK_NORM,
WORK_RIS,
WORK_RJ,
YEAR_RE,
ascii_slug,
load_topic_prefix_map,
@@ -38,6 +41,7 @@ from tools.kb_common import ( # noqa: E402
ROOT = Path(__file__).resolve().parents[1]
KV_DIR = ROOT / ".firecrawl" / "wko-kv" / "docs"
RIS_DIR = ROOT / ".ris"
RJ_DIR = ROOT / ".rechtsprechung"
DOCS = ROOT / "wissensbasis" / "dokumente"
CATALOGS = ROOT / "tools" / "catalogs"
CRAWL_STAND = "2026-09" # Beschaffungsstand der beiden Quellen
@@ -460,6 +464,283 @@ def extract_law_abbreviation(title: str, base: str) -> str:
return base
# --------------------------------------------------------------------------
# Rechtsprechung (RJ): Metadaten-MD -> Layer 2
# --------------------------------------------------------------------------
RJ_FIELD_RE = re.compile(r"^\*\*(.+?):\*\*\s*(.*)$", re.M)
RJ_H2_SPLIT_RE = re.compile(r"(?m)^## ")
RJ_NOISE_LINE_RE = re.compile(r"//-->|^\s*[\[\]]\s*$|^[\s\[]*(Kanzlei|Gesetze|Entscheidungen)\]?\s*$")
def rj_parse_fields(raw: str) -> tuple[dict[str, str], str]:
"""Extrahiert die '**Feld:** Wert'-Metadatenzeilen und den Body nach dem
ersten '---'-Trenner (bei Rechtssätzen leer)."""
fields: dict[str, str] = {}
for m in RJ_FIELD_RE.finditer(raw):
fields[m.group(1).strip()] = m.group(2).strip()
parts = re.split(r"\n\s*---\s*\n", raw, maxsplit=1)
body = parts[1] if len(parts) > 1 else ""
return fields, body
def rj_clean_body(body: str) -> str:
"""Rauscht Zeilen (lexetius-Kommentarreste, Bracket-Fragmente) und
führende Strukturetiketten heraus; kollabiert Leerzeilen."""
lines = []
for line in body.split("\n"):
if RJ_NOISE_LINE_RE.search(line):
continue
lines.append(line.rstrip())
text = "\n".join(lines)
text = re.sub(
r"\A\s*(?:Urteil|Beschluss|Erkenntnis|Entscheidung|Begr\u00fcndung:?|"
r"Entscheidungsgr\u00fcnde:?|Spruch:?)\s*\n+",
"",
text,
)
text = re.sub(r"\n{3,}", "\n\n", text)
return text.strip()
def rj_split_text(text: str, limit: int = 3500) -> list[tuple[str, str]]:
"""Lange Begründungen in ~limit-Zeichen-Stücke an Absatzgrenzen
splitten -> eigene H2-Chunks (Retrieval-Granularität, D10/D11-Budget)."""
if not text:
return []
if len(text) <= limit:
return [("Begründung", text)]
parts: list[str] = []
current: list[str] = []
size = 0
for para in re.split(r"\n\n+", text):
if size and size + len(para) > limit:
parts.append("\n\n".join(current))
current, size = [], 0
current.append(para)
size += len(para) + 2
if current:
parts.append("\n\n".join(current))
if len(parts) == 1:
return [("Begründung", parts[0])]
return [(f"Begründung ({i})", p) for i, p in enumerate(parts, 1)]
def _rj_short(text: str, limit: int = 100) -> str:
if len(text) <= limit:
return text
cut = text[:limit + 1].rsplit(" ", 1)[0]
return cut.rstrip(".,;") + "…"
def rj_build_sections(fields: dict[str, str], body: str, kind: str) -> list[tuple[str, str]]:
"""H2-Sektionen je Entscheidungstyp (Rechtssatz/Volltext/EuGH/Norm)."""
sections: list[tuple[str, str]] = []
if kind == "norm":
h2 = _h2_sections(body)
if not h2:
# 17 Norm-Dateien haben den Wortlaut als Fließtext nach dem
# '---'-Trenner (ohne H2-Struktur) — als eigene Sektion übernehmen.
flat = rj_clean_body(body)
if flat:
sections.append(("Normtext", flat))
return sections
for title, text in h2:
t = title.strip().rstrip(".")
if t.lower().startswith("zuletzt aktualisiert"):
continue
if t.lower() in ("gesetzesnummer", "dokumentnummer", "alte dokumentnummer"):
continue
if t.startswith("\u00a7") or t.startswith("Art"):
sections.append((f"{t} \u2013 Wortlaut", text))
else:
sections.append((f"{t} (Index)", text))
return sections
if kind == "rechtssatz":
if fields.get("Rechtssatz"):
sections.append(("Rechtssatz", fields["Rechtssatz"]))
if fields.get("Entscheidungstexte"):
sections.append(("Entscheidungstexte (Fundstellen)", fields["Entscheidungstexte"]))
return sections
# Volltext (JJT/JWT/JFT) und EuGH
if fields.get("Betreff"):
sections.append(("Sachbetreff", fields["Betreff"]))
if fields.get("Leitsatz"):
sections.append(("Leitsatz", fields["Leitsatz"]))
if fields.get("Spruch"):
sections.append(("Spruch", fields["Spruch"]))
if fields.get("Rechtliche Beurteilung"):
sections.append(("Rechtliche Beurteilung", fields["Rechtliche Beurteilung"]))
text = rj_clean_body(body)
sections.extend(rj_split_text(text))
if fields.get("Beachte"):
sections.append(("Beachte", fields["Beachte"]))
return sections
def _h2_sections(body: str) -> list[tuple[str, str]]:
"""Bestehende H2-Struktur (gesetze/*.md) in (Titel, Text) zerlegen."""
out: list[tuple[str, str]] = []
parts = RJ_H2_SPLIT_RE.split(body)
for part in parts[1:]:
lines = part.split("\n", 1)
title = lines[0].strip()
text = lines[1].strip() if len(lines) > 1 else ""
if title and text:
out.append((title, text))
return out
def rj_entry(path: Path, catalog: dict, stats: dict) -> tuple[str, str] | None:
rel = path.relative_to(RJ_DIR).as_posix()
raw = path.read_text(encoding="utf-8")
fields, body = rj_parse_fields(raw)
if path.parent.name == "gesetze":
kind = "norm"
elif rel.startswith("originale/EUGH"):
kind = "eugh"
elif not fields.get("Gericht"):
stats["rj_unparsed"].append(rel)
return None
elif "Rechtssatz" in fields:
kind = "rechtssatz"
else:
kind = "volltext"
court = fields.get("Gericht", "EuGH").strip()
court_tag = ascii_slug(court)
gnr = None
if kind == "norm":
kurztitel = fields.get("Kurztitel", "Norm")
abk = fields.get("Abk\u00fcrzung", "")
para = fields.get("\u00a7/Artikel/Anlage", "")
stand = None
for title, text in _h2_sections(body):
if title.strip().lower().startswith("zuletzt aktualisiert"):
m = DATE_RE.search(text)
if m:
stand = stand_from_date(m.groups())
if title.strip() == "Gesetzesnummer":
gnr = text.strip()
if not stand:
m = DATE_RE.search(body or "")
stand = stand_from_date(m.groups()) if m else CRAWL_STAND
stand_fallback = True
else:
stand_fallback = False
title = f"{kurztitel} ({abk}) \u2013 {para}" if abk else f"{kurztitel} \u2013 {para}"
work = WORK_NORM
chapter = "Zitierte Norm (Originalwortlaut)"
legal_bases = [f"{abk} {para}".strip()] if abk and para else ([abk] if abk else [])
tags = ["rechtsprechung", "norm", ascii_slug(abk) or "ris"]
if stand_fallback:
tags.append("stand-abruf")
quelle = (
f"*{work}, Stand {stand}. {kurztitel}"
+ (f" ({abk})" if abk else "")
+ (f", Gesetzesnummer {gnr}" if gnr else "")
+ f" \u2013 Originalwortlaut aus dem RIS, zitiert in der Rechtsprechung dieses Korpus.*"
)
else:
m = DATE_RE.search(fields.get("Entscheidungsdatum", ""))
stand = stand_from_date(m.groups()) if m else CRAWL_STAND
datum_de = fields.get("Entscheidungsdatum", "")
gz = fields.get("Gesch\u00e4ftszahl", "")
ecli = fields.get("European Case Law Identifier", "")
if kind == "rechtssatz":
title = _rj_short(fields.get("Rechtssatz", f"{court} {gz}"))
chapter = f"{court}-Rechtssatz"
tags = ["rechtsprechung", court_tag, "rechtssatz"]
elif kind == "eugh":
h1 = re.match(r"\s*#\s*(.+)", raw)
rs = h1.group(1).strip() if h1 else path.stem.replace("_", " ")
dm = re.search(r"Urteil vom (\d{1,2}\.\s*\d{1,2}\.\s*\d{4})", raw)
datum_de = dm.group(1).replace(" ", " ").strip() if dm else datum_de
dm2 = DATE_RE.search(datum_de)
if dm2:
stand = stand_from_date(dm2.groups())
title = (
f"{rs}, Urteil vom {datum_de}"
if rs.lower().startswith("eugh")
else f"EuGH {rs}, Urteil vom {datum_de}"
)
court_tag = "eugh"
chapter = "EuGH-Urteil"
tags = ["rechtsprechung", "eugh", "urteil"]
else:
title = f"{court} {gz} vom {datum_de}"
chapter = f"{court}-Erkenntnis (Volltext)"
tags = ["rechtsprechung", court_tag, "volltext"]
year = YEAR_RE.search(stand)
if year:
tags.append(f"jahr-{year.group(1)}")
work = WORK_EUGH if kind == "eugh" else WORK_RJ
legal_bases = [
n.strip().rstrip(";")
for n in re.split(r";\s*", fields.get("Norm", ""))
if n.strip()
]
quelle = (
f"*{work}, {court}, Entscheidung vom {datum_de}"
+ (f", GZ {gz}" if gz else "")
+ (f", {ecli}" if ecli else "")
)
if kind == "eugh":
quelle += (
f". Textwiedergabe (nicht amtlich): {fields.get('URL', '')}"
+ (
f"; amtliche Fassung: {fields.get('Offizielle Fassung', '')}"
if fields.get("Offizielle Fassung")
else ""
)
+ ".*"
)
else:
quelle += ".*"
key = rel
entry = catalog["entries"].get(key)
if entry is None:
seq = catalog["next_seq"]
catalog["next_seq"] = seq + 1
slug = "rj_" + ascii_slug(path.stem)
entry = {"id": f"rj-rjs-{seq:03d}", "slug": slug}
catalog["entries"][key] = entry
kid, slug = entry["id"], entry["slug"]
sections = rj_build_sections(fields, body, kind)
body_md = "\n\n".join(f"## {t}\n\n{text}" for t, text in sections if text)
if not body_md.strip():
stats["rj_empty"].append(rel)
return None
md = (
f"---\n"
f"id: {kid}\n"
f"batch: 1\n"
f"title: {json.dumps(title, ensure_ascii=False)}\n"
f"work: {json.dumps(work, ensure_ascii=False)}\n"
f"chapter: {json.dumps(chapter, ensure_ascii=False)}\n"
f"topic: rechtsprechung\n"
f"author: {json.dumps(court if kind != 'norm' else (fields.get('Abk\u00fcrzung') or court), ensure_ascii=False)}\n"
f"stand: {stand}\n"
f"source:\n"
f" text: \".rechtsprechung/{rel}\"\n"
f"legal_bases: {json.dumps(legal_bases, ensure_ascii=False)}\n"
f"tags: {json.dumps(tags, ensure_ascii=False)}\n"
f"cross_refs: []\n"
f"---\n\n"
f"# {title}\n\n"
f"{quelle}\n\n"
f"*Wissensbasis-ID: {kid}*\n\n"
f"{body_md}\n"
)
entry.update({"title": title, "stand": stand, "kind": kind})
stats["rj_written"] += 1
return slug, md
# --------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------
@@ -477,8 +758,9 @@ def write_entries(results: list[tuple[str, str]], dry: bool) -> int:
def run(source: str, dry: bool, limit: int | None, single: str | None) -> int:
stats: dict = {"kv_written": 0, "ris_written": 0, "no_og": [], "no_main": [],
"ris_unmapped": []}
stats: dict = {"kv_written": 0, "ris_written": 0, "rj_written": 0,
"no_og": [], "no_main": [], "ris_unmapped": [],
"rj_unparsed": [], "rj_empty": []}
written = 0
if source in ("kv", "all"):
@@ -513,9 +795,31 @@ def run(source: str, dry: bool, limit: int | None, single: str | None) -> int:
if not dry:
save_catalog(catalog)
if source in ("rj", "all"):
catalog = load_catalog("rj")
files = sorted(RJ_DIR.glob("originale/*.md")) + sorted(
RJ_DIR.glob("gesetze/*.md")
)
files = [f for f in files if f.name != "INDEX.md"]
if single:
files = [f for f in files if single in f.name]
if limit:
files = files[:limit]
results = []
for f in files:
res = rj_entry(f, catalog, stats)
if res:
results.append(res)
written += write_entries(results, dry)
if not dry:
save_catalog(catalog)
print(f"source={source} dry={dry} written={written}")
print(f" kv: {stats['kv_written']} ris: {stats['ris_written']}")
for key in ("no_og", "no_main", "ris_unmapped"):
print(
f" kv: {stats['kv_written']} ris: {stats['ris_written']} "
f"rj: {stats['rj_written']}"
)
for key in ("no_og", "no_main", "ris_unmapped", "rj_unparsed", "rj_empty"):
if stats[key]:
print(f" {key}: {stats[key]}")
return 0
@@ -523,7 +827,7 @@ def run(source: str, dry: bool, limit: int | None, single: str | None) -> int:
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--source", choices=("kv", "ris", "all"), default="all")
ap.add_argument("--source", choices=("kv", "ris", "rj", "all"), default="all")
ap.add_argument("--dry-run", action="store_true")
ap.add_argument("--limit", type=int, default=None)
ap.add_argument("--file", default=None, help="Teilstring des Quelldateinamens")