Files
pv-agent/tools/kb_common.py
T

151 lines
7.0 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Gemeinsame Konstanten für die Wissensbasis-Tools (Intake + Registry).
Quellen-ID-Räume: `lb` (Lexis Briefings), `wk` (WIKU Personal), `kv`
(WKO.at Kollektivvertrag-Seiten), `ris` (RIS-Gesetze). Die bestehende
Cluster-Map (topic -> ID-Präfix) der lb/wk-Einträge wird von
`build_registry.py` aus den Layer-2-Frontmatterdaten abgeleitet; nur die
Cluster der neuen Quellen sind hier deklariert.
LAW_MAP begründet je Gesetz topic/chapter aus dem tatsächlichen Inhalt
der .ris-Dateien (Stand 2026-09-15, vgl. §-Inventar der Quelldateien):
Gesetze ohne natürlichen Cluster bekommen einen eigenen Rechtsgebiet-
Cluster oder landen bewusst im Sammel-Cluster `normen-sonstige`.
"""
from __future__ import annotations
import re
from pathlib import Path
# Alle ID-Räume des Korpus (Schema-Beschreibung für kb.json).
ID_SPACES = {
"lb": "Lexis Briefings Personalrecht (Lexis 360-Export)",
"wk": "WIKU Personal (Fachbroschüren, Arbeitsunterlagen, Casebook, Personal aktuell)",
"kv": "WKO.at — Kollektivvertrag-Dokumente (Lohn-/Gehaltsordnungen, KV-Texte)",
"ris": "RIS — Rechtsinformationssystem des Bundes (Gesetzes-§-Auschnitte)",
"rj": "RIS/OGD — Rechtsprechung (OGH, VwGH, VfGH, EuGH) und zitierte Normen",
}
# Neue Cluster der Quellen kv/ris/rj: slug -> (ID-Präfix, Anzeigename).
NEW_CLUSTERS = {
"kollektivvertraege": ("kvt", "Kollektivverträge (WKO.at)"),
"zivilrecht-normen": ("zvr", "Zivilrechtliche Normen (ABGB, ZPO & Co.)"),
"arbeitsvertragsrecht": ("avr", "Arbeitsvertragsrecht (AVRAG)"),
"arbeitsgerichtsbarkeit": ("agg", "Arbeits- und Sozialgerichtsbarkeit (ASGG)"),
"lohnsteuer": ("lst", "Lohnsteuer & Einkommensteuer (EStG)"),
"abgabenverfahren": ("abo", "Abgabenverfahren (BAO)"),
"normen-sonstige": ("nso", "Weitere Gesetze und Verordnungen"),
"rechtsprechung": (
"rjs",
"Rechtsprechung (OGH, VwGH, VfGH, EuGH) und zitierte Normen",
),
}
# Werk-Namen der neuen Quellen (Frontmatter `work`).
WORK_KV = "WKO.at Kollektivvertrag"
WORK_RIS = "RIS Rechtsinformationssystem des Bundes"
WORK_RJ = "RIS Rechtsprechung (OGD-Justiz/VwGH/VfGH)"
WORK_EUGH = "EuGH Rechtsprechung (Textwiedergabe lexetius, nicht amtlich)"
WORK_NORM = "RIS Originalwortlaut zitierte Norm"
# .ris/<Datei ohne .md> -> (topic, chapter, lfd. Nummer im Cluster).
# Nummern sind fix (dann eingefroren); neue Gesetze hängen in ihrem Cluster
# an der höchsten Nummer an.
LAW_MAP: dict[str, tuple[str, str, int]] = {
"ABGB": ("zivilrecht-normen", "Zivilrecht", 1),
"IPRG": ("zivilrecht-normen", "Zivilrecht", 2),
"KSchG": ("zivilrecht-normen", "Verbraucherschutz", 3),
"ZPO": ("zivilrecht-normen", "Zivilrecht", 4),
"APG": ("pension", "Pensionsrecht", 1),
"APSG": ("arbeitszeitmodelle", "Arbeitsrecht (Kurzarbeit)", 1),
"APflG": ("lehrlinge", "Arbeitsrecht (Lehrverhältnis)", 1),
"BAG": ("lehrlinge", "Arbeitsrecht (Lehrverhältnis)", 2),
"ARG": ("ruhezeiten", "Arbeitsrecht", 1),
"ASGG": ("arbeitsgerichtsbarkeit", "Arbeits- und Sozialgerichtsbarkeit", 1),
"ASVG": ("beitragsrecht-asvg", "Sozialversicherungsrecht", 1),
"ASchG": ("arbeitnehmerschutz", "Arbeitnehmerschutz", 1),
"BS-V": ("arbeitnehmerschutz", "Arbeitnehmerschutz", 2),
"BauV": ("arbeitnehmerschutz", "Arbeitnehmerschutz", 3),
"AVRAG": ("arbeitsvertragsrecht", "Arbeitsrecht", 1),
"AZG": ("arbeitszeitgrenzen", "Arbeitsrecht", 1),
"AktG": ("vorstand", "Gesellschaftsrecht", 1),
"AlVG": ("altersteilzeit", "Sozialversicherungsrecht", 1),
"AngG": ("beendigungsarten", "Arbeitsrecht", 1),
"LAG": ("beendigungsarten", "Arbeitsrecht", 2),
"ArbIG": ("normen-sonstige", "Arbeitsrecht", 1),
"RStDG": ("normen-sonstige", "Dienstrecht", 2),
"StGB": ("normen-sonstige", "Strafrecht", 3),
"StPO": ("normen-sonstige", "Strafrecht", 4),
"UGB": ("normen-sonstige", "Gesellschaftsrecht", 5),
"ArbVG": ("betriebsrat", "Arbeitsverfassungsrecht", 1),
"AÜG": ("arbeitskrafteuberlassung", "Arbeitsrecht", 1),
"BAO": ("abgabenverfahren", "Steuerrecht", 1),
"BBG": ("behinderte", "Sozialrecht", 1),
"BEinstG": ("behinderte", "Sozialrecht", 2),
"BMSVG": ("vorsorgeleistungen", "Betriebliche Vorsorge", 1),
"PKG": ("vorsorgeleistungen", "Betriebliche Vorsorge", 2),
"FlexKapGG": ("vorsorgeleistungen", "Gesellschaftsrecht", 3),
"BSVG": ("bauerliche-sozialversicherung", "Sozialversicherungsrecht", 1),
"BUAG": ("endabrechnung", "Arbeitsrecht", 1),
"DHG": ("dienstnehmerhaftung", "Arbeitsrecht", 1),
"EFZG": ("krankenstand", "Arbeitsrecht", 1),
"EO": ("lohnpfandung", "Exekutionsrecht", 1),
"EStG": ("lohnsteuer", "Steuerrecht", 1),
"FLAG": ("kinderbetreuungsgeld", "Familienleistungen", 1),
"KBGG": ("kinderbetreuungsgeld", "Familienleistungen", 2),
"FamZeitbG": ("familienzeit", "Familienleistungen", 1),
"GSVG": ("gsvg-fsvg", "Sozialversicherungsrecht", 1),
"GewO": ("gewerbe", "Gewerberecht", 1),
"GlBG": ("gleichbehandlung", "Arbeitsrecht", 1),
"GmbHG": ("geschaftsfuhrer", "Gesellschaftsrecht", 1),
"IESG": ("insolvenz-betriebsubergang", "Insolvenzrecht", 1),
"KJBG": ("jugendliche", "Jugendschutz", 1),
"LSD-BG": ("lohndumping", "Arbeitsrecht", 1),
"MSchG": ("schwangerschaft", "Mutterschutz", 1),
"MSchV": ("schwangerschaft", "Mutterschutz", 2),
"NSchG": ("nachtschwerarbeit", "Arbeitsrecht", 1),
"Sachbezugswerteverordnung": ("sachbezuge", "Steuerrecht", 1),
"Schwerarbeitsverordnung": ("schwerarbeit", "Arbeitsrecht", 1),
"UrlG": ("urlaub", "Arbeitsrecht", 1),
"TAG": ("urlaub", "Arbeitsrecht", 2),
"VBG": ("urlaub", "Dienstrecht", 3),
"VKG": ("karenz", "Familienrecht", 1),
"ZDG": ("prasenzdienst", "Zivildienst", 1),
}
DATE_RE = re.compile(r"(\d{1,2})\.\s*(\d{1,2})\.\s*(\d{4})")
YEAR_RE = re.compile(r"(20\d\d)")
def load_topic_prefix_map(kb_json_path) -> dict[str, str]:
"""topic -> ID-Präfix über ALLE Cluster: aus den bestehenden Layer-2-
Einträgen (kb.json) abgeleitet, ergänzt um die neuen Cluster."""
import json
kb = json.loads(Path(kb_json_path).read_text(encoding="utf-8"))
mapping: dict[str, str] = {}
for e in kb["entries"]:
prefix = e["id"].split("-")[1]
known = mapping.get(e["topic"])
if known is not None and known != prefix:
raise SystemExit(
f"inkonsistente Cluster-Map: topic '{e['topic']}' hat Präfixe "
f"{known} und {prefix}"
)
mapping[e["topic"]] = prefix
for slug, (prefix, _name) in NEW_CLUSTERS.items():
mapping[slug] = prefix
return mapping
def ascii_slug(text: str) -> str:
"""Wissensbasis-Konvention: Umlaute auf Basisbuchstabe (ü→u, nicht ue),
ß→ss, Rest klein; alles außer a-z0-9 und Bindestrich fällt weg."""
repl = {"ä": "a", "ö": "o", "ü": "u", "Ä": "a", "Ö": "o", "Ü": "u", "ß": "ss"}
for k, v in repl.items():
text = text.replace(k, v)
return re.sub(r"[^a-z0-9-]+", "-", text.lower()).strip("-")
def stand_from_date(date_match: tuple[str, str, str]) -> str:
d, m, y = date_match
return f"{y}-{int(m):02d}"