839 lines
30 KiB
Python
839 lines
30 KiB
Python
"""Intake der Quellen `kv` (WKO.at-KV-Seiten) und `ris` (RIS-Gesetze) in die
|
||
Layer-2-Wissensbasis.
|
||
|
||
Abweichend von den lizenzierten Quellen (Lexis/WIKU: Eigene-Worte-Kuratierung)
|
||
werden kv/ris-Einträge **quellentreu generiert** (Entscheidung 2026-09-15):
|
||
Gesetze sind amtliche Werke, KV-Lohntabellen müssen zahlenexakt bleiben.
|
||
Die generierten Dateien (`kv_*.md`, `ris_*.md`) sind damit Tool-Output —
|
||
keine manuelle Kuratierung, bei Änderungen neu generieren.
|
||
|
||
IDs werden aus den Katalogen `tools/catalogs/<source>_catalog.json` vergeben
|
||
und danach eingefroren (Wissensbasis-Konvention: nie wiederverwenden).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import re
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
from bs4 import BeautifulSoup, Comment, Tag
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
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,
|
||
stand_from_date,
|
||
)
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
KV_DIR = ROOT / ".firecrawl" / "kv-portal" / "wko-kv" / "docs"
|
||
RIS_DIR = ROOT / ".firecrawl" / "ris" / "gesetze"
|
||
RJ_DIR = ROOT / ".rechtsprechung"
|
||
DOCS = ROOT / "wissensbasis" / "dokumente"
|
||
CATALOGS = ROOT / "tools" / "catalogs"
|
||
CRAWL_STAND = "2026-09" # Beschaffungsstand der beiden Quellen
|
||
|
||
|
||
TOPIC_PREFIX_FULL: dict[str, str] | None = None
|
||
|
||
|
||
def topic_prefix(topic: str) -> str:
|
||
global TOPIC_PREFIX_FULL
|
||
if TOPIC_PREFIX_FULL is None:
|
||
TOPIC_PREFIX_FULL = load_topic_prefix_map(ROOT / "wissensbasis" / "kb.json")
|
||
return TOPIC_PREFIX_FULL[topic]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Kataloge (eingefrorene IDs)
|
||
# --------------------------------------------------------------------------
|
||
|
||
def load_catalog(source: str) -> dict:
|
||
path = CATALOGS / f"{source}_catalog.json"
|
||
if path.is_file():
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
return {"source": source, "updated": None, "next_seq": 1, "entries": {}}
|
||
|
||
|
||
def save_catalog(catalog: dict) -> None:
|
||
catalog["updated"] = datetime.now(timezone.utc).isoformat()
|
||
CATALOGS.mkdir(parents=True, exist_ok=True)
|
||
path = CATALOGS / f"{catalog['source']}_catalog.json"
|
||
path.write_text(
|
||
json.dumps(catalog, ensure_ascii=False, indent=1, sort_keys=True) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# WKO-KV: HTML -> Markdown
|
||
# --------------------------------------------------------------------------
|
||
|
||
NOISE_TAGS = ("script", "style", "button", "svg", "nav", "iframe", "form", "img")
|
||
NOISE_CLASSES = re.compile(
|
||
r"social|share|related|teaser|breadcrumb|pagination|print|meta-details", re.I
|
||
)
|
||
|
||
|
||
def prep_soup(html: str) -> BeautifulSoup:
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
for tag in soup.find_all(NOISE_TAGS):
|
||
tag.decompose()
|
||
for br in soup.find_all("br"):
|
||
br.replace_with("\n")
|
||
return soup
|
||
|
||
|
||
def clean_text(text: str) -> str:
|
||
"""Zeilenweise strippen, interne Whitespace-Läufe kollabieren."""
|
||
lines = []
|
||
for line in text.split("\n"):
|
||
line = re.sub(r"[ \t\xa0]+", " ", line).strip()
|
||
if line:
|
||
lines.append(line)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def cell_text(cell: Tag) -> str:
|
||
text = cell.get_text(" ")
|
||
return re.sub(r"\s+", " ", text).strip().replace("|", "\\|")
|
||
|
||
|
||
def table_to_md(table: Tag) -> str:
|
||
rows = []
|
||
for tr in table.find_all("tr"):
|
||
if tr.find_parent("table") is not table:
|
||
continue # verschachtelte Tabellen nur einmal ausgeben
|
||
cells = [cell_text(c) for c in tr.find_all(["td", "th"], recursive=False)]
|
||
if cells:
|
||
rows.append(cells)
|
||
if not rows:
|
||
return ""
|
||
width = max(len(r) for r in rows)
|
||
rows = [r + [""] * (width - len(r)) for r in rows]
|
||
out = ["| " + " | ".join(rows[0]) + " |"]
|
||
out.append("| " + " | ".join(["---"] * width) + " |")
|
||
for r in rows[1:]:
|
||
out.append("| " + " | ".join(r) + " |")
|
||
return "\n".join(out)
|
||
|
||
|
||
def list_to_md(el: Tag, depth: int = 0) -> list[str]:
|
||
lines = []
|
||
indent = " " * depth
|
||
ordered = el.name == "ol"
|
||
for i, li in enumerate(el.find_all("li", recursive=False), 1):
|
||
own = []
|
||
for child in li.children:
|
||
if isinstance(child, Tag) and child.name in ("ul", "ol"):
|
||
continue
|
||
own.append(child.get_text(" ") if isinstance(child, Tag) else str(child))
|
||
text = clean_text("".join(own))
|
||
marker = f"{i}. " if ordered else "- "
|
||
if text:
|
||
lines.append(f"{indent}{marker}{text.replace(chr(10), chr(10) + indent + ' ')}")
|
||
for sub in li.find_all(["ul", "ol"], recursive=False):
|
||
lines.extend(list_to_md(sub, depth + 1))
|
||
return lines
|
||
|
||
|
||
def _is_toc_paragraph(p: Tag) -> bool:
|
||
"""Absatz, der ausschließlich einen In-Seiten-Anker-Link enthält."""
|
||
links = p.find_all("a", href=re.compile(r"^#"))
|
||
if not links:
|
||
return False
|
||
rest = clean_text(p.get_text(" "))
|
||
for a in links:
|
||
rest = rest.replace(clean_text(a.get_text(" ")), "", 1)
|
||
return not rest
|
||
|
||
|
||
def blocks_to_md(el: Tag) -> list[str]:
|
||
"""Rekursiver Block-Konverter; liefert Markdown-Zeilen."""
|
||
lines: list[str] = []
|
||
skip_toc = False # TOC-Absätze nach „Inhalt“-Überschrift auslassen
|
||
for child in el.children:
|
||
if isinstance(child, Comment):
|
||
continue
|
||
if isinstance(child, str):
|
||
text = clean_text(child)
|
||
if text:
|
||
lines.extend([text, ""])
|
||
continue
|
||
if not isinstance(child, Tag):
|
||
continue
|
||
classes = " ".join(child.get("class", []))
|
||
if child.name in NOISE_TAGS or NOISE_CLASSES.search(classes):
|
||
continue
|
||
if child.name in ("h1", "h2", "h3", "h4", "h5", "h6"):
|
||
title = clean_text(child.get_text(" "))
|
||
if not title:
|
||
continue
|
||
if title.casefold() == "inhalt":
|
||
skip_toc = True
|
||
continue
|
||
prefix = "###" if child.name in ("h5", "h6") else "##"
|
||
lines.extend([f"{prefix} {title}", ""])
|
||
skip_toc = False
|
||
continue
|
||
if child.name == "p" and skip_toc and _is_toc_paragraph(child):
|
||
continue # TOC-Einträge (reine Anker-Links) auslassen
|
||
skip_toc = False
|
||
if child.name in ("ul", "ol"):
|
||
lines.extend(list_to_md(child))
|
||
lines.append("")
|
||
continue
|
||
if child.name == "table":
|
||
md = table_to_md(child)
|
||
if md:
|
||
lines.extend([md, ""])
|
||
continue
|
||
if child.name == "p":
|
||
text = clean_text(child.get_text())
|
||
if text:
|
||
lines.extend([text, ""])
|
||
continue
|
||
if child.name == "dl":
|
||
for dt in child.find_all("dt"):
|
||
dd = dt.find_next_sibling("dd")
|
||
if dd is not None:
|
||
lines.append(f"- **{clean_text(dt.get_text(' '))}:** "
|
||
f"{clean_text(dd.get_text(' '))}")
|
||
lines.append("")
|
||
continue
|
||
if child.name in ("div", "section", "article", "main", "header", "footer"):
|
||
lines.extend(blocks_to_md(child))
|
||
continue
|
||
# blockquote, pre, hr, sonstiges: Text durchreichen
|
||
text = clean_text(child.get_text())
|
||
if text:
|
||
lines.extend([text, ""])
|
||
return lines
|
||
|
||
|
||
def parse_kv_meta(main: Tag) -> list[tuple[str, str]]:
|
||
"""dt/dd-Paare aus dem meta-details-Block."""
|
||
pairs = []
|
||
box = main.find("div", class_="meta-details")
|
||
if box is None:
|
||
return pairs
|
||
for dt in box.find_all("dt"):
|
||
dd = dt.find_next_sibling("dd")
|
||
label = clean_text(dt.get_text(" ")).rstrip(":")
|
||
value = clean_text(dd.get_text(" ")) if dd is not None else ""
|
||
if label:
|
||
pairs.append((label, value))
|
||
return pairs
|
||
|
||
|
||
DOCTYPE_RULES: list[tuple[str, str]] = [
|
||
("zusatzkollektivvertrag", "Zusatz-KV"),
|
||
("zusatz-kv", "Zusatz-KV"),
|
||
("zusatzvereinbarung", "Zusatz-KV"),
|
||
("rahmenkollektivvertrag", "Rahmen-KV"),
|
||
("rahmen-kv", "Rahmen-KV"),
|
||
("generalkollektivvertrag", "Generalkollektivvertrag"),
|
||
("general-kv", "Generalkollektivvertrag"),
|
||
("mantelvertrag", "Mantelvertrag"),
|
||
("berechnungsbeispiele", "Berechnungsbeispiele"),
|
||
("fragen", "Fragen & Antworten"),
|
||
("kommentar", "Kommentar"),
|
||
("interpretation", "Interpretation"),
|
||
("erläuterung", "Erläuterungen"),
|
||
("handlungsempfehlung", "Handlungsempfehlung"),
|
||
("empfehlung", "Empfehlung"),
|
||
("information", "Information"),
|
||
("erklärung", "Erklärung"),
|
||
("klarstellung", "Klarstellung"),
|
||
("änderung", "Änderung"),
|
||
("anhang", "Anhang/Anlage"),
|
||
("anlage", "Anhang/Anlage"),
|
||
("sozialpartnervereinbarung", "Vereinbarung"),
|
||
("vereinbarung", "Vereinbarung"),
|
||
("punktation", "Vereinbarung"),
|
||
("überleitungsschema", "Tabelle/Schema"),
|
||
("umstiegsdienstzettel", "Tabelle/Schema"),
|
||
("rückzahlungstabelle", "Tabelle/Schema"),
|
||
("dienst- und besoldungsordnung", "Dienst- und Besoldungsordnung"),
|
||
("gehaltsordnung", "Gehaltsordnung"),
|
||
("gehaltsordnungen", "Gehaltsordnung"),
|
||
("gehaltstabelle", "Gehaltsordnung"),
|
||
("gehaltstafel", "Gehaltsordnung"),
|
||
("lohnordnung", "Lohnordnung"),
|
||
("lohntafeln", "Lohnordnung"),
|
||
("lohntafel", "Lohnordnung"),
|
||
("lohn- gehaltsordnung", "Lohn-/Gehaltsordnung"),
|
||
("lohn-gehaltsordnung", "Lohn-/Gehaltsordnung"),
|
||
("lohn- und gehaltsordnung", "Lohn-/Gehaltsordnung"),
|
||
("heimarbeitstarif", "Heimarbeitstarif"),
|
||
("kollektivvertragsabschluss", "KV-Abschluss"),
|
||
("kollektivvertrag", "Kollektivvertrag"),
|
||
]
|
||
|
||
|
||
def classify_doctype(title: str) -> str:
|
||
low = title.casefold()
|
||
for needle, doctype in DOCTYPE_RULES:
|
||
if low.startswith(needle):
|
||
return doctype
|
||
for needle, doctype in DOCTYPE_RULES:
|
||
if needle in low:
|
||
return doctype
|
||
return "Sonstiges KV-Dokument"
|
||
|
||
|
||
def derive_stand(title: str, meta: list[tuple[str, str]], filename: str) -> tuple[str, str]:
|
||
"""(stand, tags) — explizites Datum > Titel-Datum > Titel-Jahr >
|
||
Dateiname-Jahr > Beschaffungsstand."""
|
||
for label, value in meta:
|
||
if "geltungsdauer" in label.casefold():
|
||
m = DATE_RE.search(value)
|
||
if m:
|
||
return stand_from_date(m.groups()), []
|
||
m = YEAR_RE.search(value)
|
||
if m:
|
||
return f"{m.group(1)}-01", ["stand-jahr"]
|
||
m = DATE_RE.search(title)
|
||
if m:
|
||
return stand_from_date(m.groups()), []
|
||
m = YEAR_RE.search(title)
|
||
if m:
|
||
return f"{m.group(1)}-01", ["stand-jahr"]
|
||
m = YEAR_RE.search(filename)
|
||
if m:
|
||
return f"{m.group(1)}-01", ["stand-jahr"]
|
||
return CRAWL_STAND, ["stand-geschaetzt"]
|
||
|
||
|
||
def kv_entry(path: Path, catalog: dict, stats: dict) -> tuple[str, str] | None:
|
||
html = path.read_text(encoding="utf-8")
|
||
soup = prep_soup(html)
|
||
og_title = soup.find("meta", property="og:title")
|
||
og_url = soup.find("meta", property="og:url")
|
||
if og_title is None or og_url is None:
|
||
stats["no_og"].append(path.name)
|
||
return None
|
||
title = clean_text(og_title["content"])
|
||
url = og_url["content"]
|
||
main = soup.find("main", class_="col-lg-8")
|
||
if main is None:
|
||
stats["no_main"].append(path.name)
|
||
return None
|
||
meta = parse_kv_meta(main)
|
||
body_blocks = blocks_to_md(main)
|
||
|
||
key = path.name
|
||
entry = catalog["entries"].get(key)
|
||
if entry is None:
|
||
seq = catalog["next_seq"]
|
||
catalog["next_seq"] = seq + 1
|
||
slug = "kv_" + path.stem.lstrip("-")
|
||
entry = {"id": f"kv-kvt-{seq:03d}", "slug": slug}
|
||
catalog["entries"][key] = entry
|
||
kid, slug = entry["id"], entry["slug"]
|
||
|
||
stand, stand_tags = derive_stand(title, meta, path.name)
|
||
doctype = classify_doctype(title)
|
||
year = YEAR_RE.search(path.stem)
|
||
tags = ["kollektivvertrag", ascii_slug(doctype)]
|
||
if year:
|
||
tags.append(f"jahr-{year.group(1)}")
|
||
tags.extend(stand_tags)
|
||
|
||
meta_rows = "\n".join(f"| {label} | {value} |" for label, value in meta)
|
||
meta_section = ""
|
||
if meta_rows:
|
||
meta_section = (
|
||
"## Geltungsbereich (WKO-Angaben)\n\n"
|
||
"| Merkmal | Angabe |\n|---|---|\n" + meta_rows + "\n\n"
|
||
)
|
||
body = "\n".join(body_blocks).strip()
|
||
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_KV, ensure_ascii=False)}\n"
|
||
f"chapter: {json.dumps(doctype, ensure_ascii=False)}\n"
|
||
f"topic: kollektivvertraege\n"
|
||
f"author: \"WKO\"\n"
|
||
f"stand: {stand}\n"
|
||
f"source:\n"
|
||
f" html: \".firecrawl/kv-portal/wko-kv/docs/{path.name}\"\n"
|
||
f"legal_bases: []\n"
|
||
f"tags: {json.dumps(tags, ensure_ascii=False)}\n"
|
||
f"cross_refs: []\n"
|
||
f"---\n\n"
|
||
f"# {title}\n\n"
|
||
f"*{WORK_KV}, WKO, Stand {stand} ({kid}). Quelle: {url}*\n\n"
|
||
f"{meta_section}"
|
||
f"{body}\n"
|
||
)
|
||
entry.update({"title": title, "stand": stand, "doctype": doctype, "url": url})
|
||
stats["kv_written"] += 1
|
||
return slug, md
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# RIS: Markdown -> Layer-2
|
||
# --------------------------------------------------------------------------
|
||
|
||
GNR_RE = re.compile(r"Gesetzesnummer[:\s]+(\d+)")
|
||
ISO_DATE_RE = re.compile(r"(\d{4}-\d{2}-\d{2})")
|
||
|
||
|
||
def ris_entry(path: Path, catalog: dict, stats: dict) -> tuple[str, str] | None:
|
||
key = path.name
|
||
base = path.stem
|
||
law = LAW_MAP.get(base)
|
||
if law is None:
|
||
stats["ris_unmapped"].append(base)
|
||
return None
|
||
topic, chapter, nn = law
|
||
|
||
raw = path.read_text(encoding="utf-8").replace("\xa0", " ")
|
||
lines = raw.split("\n")
|
||
title = lines[0].lstrip("# ").strip()
|
||
preamble, sections = raw, ""
|
||
m = re.search(r"^## .*$", raw, flags=re.M)
|
||
if m:
|
||
preamble, sections = raw[: m.start()], raw[m.start():]
|
||
preamble_lines = [ln for ln in preamble.split("\n")[1:] if ln.strip()]
|
||
|
||
gnr = GNR_RE.search(preamble)
|
||
stand_match = ISO_DATE_RE.search(preamble)
|
||
stand = stand_match.group(1)[:7] if stand_match else CRAWL_STAND
|
||
|
||
entry = catalog["entries"].get(key)
|
||
if entry is None:
|
||
prefix = topic_prefix(topic)
|
||
kid = f"ris-{prefix}-{nn:02d}"
|
||
entry = {"id": kid}
|
||
catalog["entries"][key] = entry
|
||
kid = entry["id"]
|
||
|
||
abbr = extract_law_abbreviation(title, base)
|
||
tags = ["gesetz", ascii_slug(abbr), ascii_slug(topic)]
|
||
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_RIS, ensure_ascii=False)}\n"
|
||
f"chapter: {json.dumps(chapter, ensure_ascii=False)}\n"
|
||
f"topic: {topic}\n"
|
||
f"author: \"RIS (Bundeskanzleramt)\"\n"
|
||
f"stand: {stand}\n"
|
||
f"source:\n"
|
||
f" text: \".firecrawl/ris/gesetze/{path.name}\"\n"
|
||
f"legal_bases: {json.dumps([abbr], ensure_ascii=False)}\n"
|
||
f"tags: {json.dumps(tags, ensure_ascii=False)}\n"
|
||
f"cross_refs: []\n"
|
||
f"---\n\n"
|
||
f"# {title}\n\n"
|
||
+ "\n".join(preamble_lines).strip() + "\n\n"
|
||
f"*Wissensbasis: {WORK_RIS}, Stand {stand} ({kid}).*\n\n"
|
||
f"{sections.strip()}\n"
|
||
)
|
||
entry.update({"title": title, "stand": stand, "gnr": gnr.group(1) if gnr else None})
|
||
stats["ris_written"] += 1
|
||
return f"ris_{ascii_slug(base)}", md
|
||
|
||
|
||
def extract_law_abbreviation(title: str, base: str) -> str:
|
||
"""Abkürzung aus dem Titel: 'X (ABK)' am Ende oder 'ABK – Name' am Anfang."""
|
||
parens = re.findall(r"\(([^()]+)\)\s*$", title.strip())
|
||
if parens:
|
||
return parens[-1].strip()
|
||
m = re.match(r"^([A-Za-zÄÖÜäöüß\-]+)\s*[–-]\s", title.strip())
|
||
if m:
|
||
return m.group(1).strip()
|
||
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
|
||
# --------------------------------------------------------------------------
|
||
|
||
def write_entries(results: list[tuple[str, str]], dry: bool) -> int:
|
||
n = 0
|
||
for slug, md in results:
|
||
target = DOCS / f"{slug}.md"
|
||
if not dry:
|
||
if target.is_file() and target.read_text(encoding="utf-8") == md:
|
||
continue
|
||
target.write_text(md, encoding="utf-8")
|
||
n += 1
|
||
return n
|
||
|
||
|
||
def run(source: str, dry: bool, limit: int | None, single: str | None) -> int:
|
||
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"):
|
||
catalog = load_catalog("kv")
|
||
files = sorted(KV_DIR.glob("*.html"))
|
||
if single:
|
||
files = [f for f in files if single in f.name]
|
||
if limit:
|
||
files = files[:limit]
|
||
results = []
|
||
for f in files:
|
||
res = kv_entry(f, catalog, stats)
|
||
if res:
|
||
results.append(res)
|
||
written += write_entries(results, dry)
|
||
if not dry:
|
||
save_catalog(catalog)
|
||
|
||
if source in ("ris", "all"):
|
||
catalog = load_catalog("ris")
|
||
files = sorted(RIS_DIR.glob("*.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 = ris_entry(f, catalog, stats)
|
||
if res:
|
||
results.append(res)
|
||
written += write_entries(results, dry)
|
||
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']} "
|
||
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
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description=__doc__)
|
||
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")
|
||
args = ap.parse_args()
|
||
return run(args.source, args.dry_run, args.limit, args.file)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main()) |