odoo-at-payroll/personalverrechnung/tools/extract_gembg_katalog.py

#!/usr/bin/env python3
"""Extract Bgld. GemBG 2014 pay scale tables (Entlohnungsschemata) into
module CSV seed data for l10n_at_gemeinde_payroll.

Source of truth: consolidated GemBG text (RIS LrBgld GesNr 20001002,
Fassung 09.09.2026, incl. Bezügeanpassung 2026 / RV 0715/XXIII. GP).
Run from the repository root:

    python3 personalverrechnung/tools/extract_gembg_katalog.py

Outputs:
    addons/l10n_at_gemeinde_payroll/data/gemeinde.payroll.entlohnungsgruppe.csv
    addons/l10n_at_gemeinde_payroll/data/gemeinde.payroll.entlohnungsstufe.csv
plus a protocol printed to stdout. Effective dates: tables amended by
Artikel 4 RV 0715 carry the "mit 1. Juli 2026" entry into force; all other
tables are dated 2025-07-01 as a documented assumption to be verified
against the LGBl in AP0 (see personalverrechnung/IMPLEMENTIERUNGSPLAN-Bgld.md).
"""
from __future__ import annotations

import csv
import re
import sys
from pathlib import Path

REPO = Path(__file__).resolve().parents[2]
SOURCE = REPO / ".firecrawl" / "ris-gbg-konsolidiert.md"
TARGET_DIR = REPO / "addons" / "l10n_at_gemeinde_payroll" / "data"
GRUPPE_CSV = TARGET_DIR / "gemeinde.payroll.entlohnungsgruppe.csv"
STUFE_CSV = TARGET_DIR / "gemeinde.payroll.entlohnungsstufe.csv"

# Tables amended by Artikel 4 RV 0715 with effect from 1 July 2026.
AMENDED_2026 = {"57", "58", "62", "88", "133g", "133i", "133j", "133u",
                "150c", "151", "151c", "151e", "157m", "158a"}
DEFAULT_GUELTIG_AB = "2025-07-01"
AMENDED_GUELTIG_AB = "2026-07-01"

# § 66 Abs 2 GemBG: default 4 years; § 150c Abs 2 and § 150d (Betreuungs-
# personen gb1/gb2): 2 years for these groups. To verify in AP0.
TWO_YEAR_GROUPS = {"gb1", "gb2", "l2b1", "l3", "gb1a", "gb3"}

PARAGRAF_RE = re.compile(r"^#### § (\d+[a-z]?)")
SECTION_RE = re.compile(r"^#### (Monatsentgelt|Einstufung)(?!.*Ausbildungsphase)")
SCHEMA_RE = re.compile(r"Entlohnungsschemas (.+?)(?:Monatsentgelt|$)")
GROUP_CELL_RE = re.compile(r"^[a-z][a-z0-9]*$")
GLUED_GROUP_RE = re.compile(r"Entlohnungsgruppe\s+([a-z][a-z0-9]*)")


def parse_german_number(raw: str) -> float | None:
    raw = raw.strip().replace("Euro", "").strip()
    if not re.match(r"^\d{1,3}(\.\d{3})*,\d{2}$|^\d+,\d{2}$", raw):
        return None
    return float(raw.replace(".", "").replace(",", "."))


def cells(row: str) -> list[str]:
    return [c.strip() for c in row.strip().strip("|").split("|")]


def sanitize(code: str) -> str:
    return re.sub(r"[^a-z0-9_]", "_", code.lower())


def main() -> int:
    text = [line.replace("\xa0", " ")
             for line in SOURCE.read_text(encoding="utf-8").splitlines()]
    gruppen: dict[tuple[str, str], dict] = {}
    stufen: list[dict] = []
    protocol: list[str] = []
    paragraf = None
    schema = None
    in_monatsentgelt = False
    current_groups: list[str] = []

    for line in text:
        m = PARAGRAF_RE.match(line)
        if m:
            paragraf = m.group(1)
            in_monatsentgelt = False
            current_groups = []
            continue
        if SECTION_RE.match(line):
            in_monatsentgelt = True
            ms = SCHEMA_RE.search(line)
            schema = ms.group(1) if ms else (f"§{paragraf}" if paragraf else "unbekannt")
            current_groups = []
            continue
        if not in_monatsentgelt or not line.startswith("|"):
            continue

        row = cells(line)
        if row and all(c and GROUP_CELL_RE.match(c) for c in row):
            current_groups = row  # codes on their own row
            continue
        if "Entlohnungsgruppe" in line:
            glued = [m.group(1) for m in
                     (GLUED_GROUP_RE.search(c) for c in row) if m]
            if glued:
                current_groups = glued  # codes glued into the header row
            continue
        if not current_groups:
            continue
        if len(row) >= 2 and row[0].isdigit():
            stufe_nr = int(row[0])
            values = row[1:]
            if len(values) < len(current_groups):
                continue
            for group, raw in zip(current_groups, values):
                amount = parse_german_number(raw)
                if amount is None:
                    protocol.append(
                        f"WARN §{paragraf} {group} Stufe {stufe_nr}: "
                        f"unverständlicher Wert {raw!r}")
                    continue
                key = (schema, group)
                if key not in gruppen:
                    gueltig_ab = (AMENDED_GUELTIG_AB if paragraf in AMENDED_2026
                                  else DEFAULT_GUELTIG_AB)
                    gruppen[key] = {
                        "code": group,
                        "schema": schema,
                        "paragraph": paragraf or "",
                        "bundesland": "bgld",
                        "vorrueckungszeitraum_jahre": 2 if group in TWO_YEAR_GROUPS else 4,
                        "gueltig_ab": gueltig_ab,
                        "max_stufe": 0,
                    }
                else:
                    gueltig_ab = gruppen[key]["gueltig_ab"]
                gruppen[key]["max_stufe"] = max(gruppen[key]["max_stufe"], stufe_nr)
                stufen.append({
                    "group": group,
                    "schema": schema,
                    "stufe": stufe_nr,
                    "monatsentgelt": f"{amount:.2f}",
                    "gueltig_ab": gueltig_ab,
                })

    TARGET_DIR.mkdir(parents=True, exist_ok=True)
    with GRUPPE_CSV.open("w", newline="", encoding="utf-8") as fh:
        writer = csv.writer(fh)
        writer.writerow(["id", "code", "schema", "paragraph", "bundesland",
                         "vorrueckungszeitraum_jahre"])
        for (schema_id, group), data in sorted(gruppen.items()):
            writer.writerow([f"bgld_grp_{sanitize(schema_id)}_{sanitize(group)}",
                             group, schema_id, data["paragraph"],
                             data["bundesland"], data["vorrueckungszeitraum_jahre"]])
    with STUFE_CSV.open("w", newline="", encoding="utf-8") as fh:
        writer = csv.writer(fh)
        writer.writerow(["id", "entlohnungsgruppe_id:id", "stufe",
                         "monatsentgelt", "gueltig_ab"])
        for s in sorted(stufen, key=lambda r: (r["schema"], r["group"], r["stufe"])):
            writer.writerow([
                f"bgld_st_{sanitize(s['schema'])}_{sanitize(s['group'])}_{s['stufe']}",
                f"l10n_at_gemeinde_payroll.bgld_grp_{sanitize(s['schema'])}_{sanitize(s['group'])}",
                s["stufe"], s["monatsentgelt"], s["gueltig_ab"],
            ])

    for (schema_id, group), data in sorted(gruppen.items()):
        protocol.append(
            f"§{data['paragraph']:>6} | Schema {schema_id:>7} | {group:<6} "
            f"| {data['max_stufe']:>2} Stufen | ab {data['gueltig_ab']} "
            f"| Vorrückung {data['vorrueckungszeitraum_jahre']} J.")
    protocol.append(f"\n{len(gruppen)} Entlohnungsgruppen, {len(stufen)} Stufenwerte.")
    protocol.append("Annahme: nicht 2026 angepasste Tabellen = Wert 1.7.2025 "
                    "(AP0 gegen LGBl verifizieren).")
    print("\n".join(protocol))
    return 0


if __name__ == "__main__":
    sys.exit(main())