fegger
5b629a5916
Cross-checked catalogue of all currently valid Austrian collective agreements, per the source policy: main source = the responsible employer-side chamber, kollektivvertrag.at (OEGB-Verlag) as complete cross-check baseline. - oegb/: 593 variants from kollektivvertrag.at with full consolidated texts (JSON primary format, Markdown for review/diff) plus catalog.json/.csv; fetched via the portal's JSON servlets (dashboard/structure/slices/topics), resumable - wko/: 614 current documents from the WKO KV database (1,820 branch overviews across 10 regions) with texts and catalog; match-report against the baseline: 407 pairs, 30 date disagreements flagged - chambers/chambers.json: curated registry of chamber-side sources and publication gaps (GOeD member-only, RA/Notariat not public, OCR-less scan PDFs, missing hospital-doctors KV) - kv-catalog.json/.csv: merged catalogue, 768 entries, chamber classification, main_source, cross_check status, manual_review lists - tools/fetch_kv_portal.py / fetch_wko_kv.py / build_kv_catalog.py: stdlib-only fetchers and merge step, each with a --refresh diff mode - RUNBOOK.md: update cycle for keeping the library current (annual, tied to the Wartung rhythm and KV rounds) Lohntafel PDFs are linked but not downloaded; wage tables contained in the KV texts are converted. No Odoo module changes.
355 lines
14 KiB
Python
355 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Merge the KV sources into the definitive cross-checked catalog.
|
|
|
|
Combines:
|
|
- ÖGB portal baseline (quellen/kv/oegb/catalog.json, texts via fetch_kv_portal.py)
|
|
- WKO Kollektivvertrags-Datenbank (quellen/kv/wko/catalog.json + match-report.json
|
|
via fetch_wko_kv.py) -- main source for WKO-negotiated KVs
|
|
- chamber-source registry (quellen/kv/chambers/chambers.json)
|
|
|
|
Per KV the catalog records the employer-side chamber (explicit map >
|
|
WKO match > preamble heuristic), the main source with its public
|
|
accessibility, and the cross-check result against the other source.
|
|
WKO documents current somewhere but absent from the ÖGB portal are kept
|
|
as wko_only entries (info pages, per-year Lohntafeln, recommendations).
|
|
|
|
The WKO title embeds "gültig ab <date>"; where a WKO<->ÖGB pair is
|
|
matched, that date is compared against the ÖGB current-slice version
|
|
and reported as date_agreement (flagged when it differs).
|
|
|
|
Run from the repository root after both fetchers:
|
|
|
|
python3 personalverrechnung/tools/build_kv_catalog.py
|
|
|
|
Outputs:
|
|
personalverrechnung/quellen/kv/kv-catalog.json
|
|
personalverrechnung/quellen/kv/kv-catalog.csv
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import datetime as dt
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from fetch_wko_kv import _fuzzy_overlap, _tokens # noqa: E402
|
|
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
KVDIR = REPO / "personalverrechnung" / "quellen" / "kv"
|
|
OEGB = KVDIR / "oegb" / "catalog.json"
|
|
WKO = KVDIR / "wko" / "catalog.json"
|
|
MATCH = KVDIR / "wko" / "match-report.json"
|
|
CHAMBERS = KVDIR / "chambers" / "chambers.json"
|
|
OUT_JSON = KVDIR / "kv-catalog.json"
|
|
OUT_CSV = KVDIR / "kv-catalog.csv"
|
|
|
|
WKO_DATE_RE = re.compile(
|
|
r"gültig ab (\d{1,2})\.\s*(\d{1,2})\.\s*(\d{4})", re.I)
|
|
|
|
|
|
def load(path: Path) -> dict:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def wko_valid_date(title: str) -> str | None:
|
|
m = WKO_DATE_RE.search(title or "")
|
|
if not m:
|
|
return None
|
|
d, mth, y = m.groups()
|
|
return f"{y}{int(mth):02d}{int(d):02d}"
|
|
|
|
|
|
def main() -> None:
|
|
oegb = load(OEGB)
|
|
wko = load(WKO)
|
|
match = load(MATCH)
|
|
chambers = load(CHAMBERS)
|
|
|
|
group_chamber: dict[str, str] = {}
|
|
chamber_info: dict[str, dict] = {}
|
|
for chamber, info in chambers["chambers"].items():
|
|
chamber_info[chamber] = info
|
|
for g in info.get("portal_groups", []):
|
|
group_chamber[g] = chamber
|
|
|
|
wko_by_slug = {d["slug"]: d for d in wko["docs"]}
|
|
matches_by_vid: dict[str, list[dict]] = {}
|
|
low_by_slug: dict[str, dict] = {}
|
|
for m in match.get("matches", []):
|
|
matches_by_vid.setdefault(m["oegb_variant_id"], []).append(m)
|
|
for m in match.get("low_confidence", []):
|
|
low_by_slug[m["wko_slug"]] = m
|
|
unmatched_wko_slugs = {u["slug"] for u in
|
|
match.get("unmatched_wko_current", [])}
|
|
|
|
AUX_RE = re.compile(
|
|
r"tabelle|berechnung|beispiel|erl?ae?uterung|antworten|"
|
|
r"empfehlung|brutto-netto|einstufungsbestimmungen|anhang", re.I)
|
|
KV_RE = re.compile(r"kollektivvertrag", re.I)
|
|
|
|
def primary_match(ms: list[dict]) -> dict:
|
|
"""Pick the WKO doc that IS the KV (not an auxiliary table/example
|
|
page) among several docs matched to the same portal variant."""
|
|
def rank(m: dict) -> tuple:
|
|
is_kv = bool(KV_RE.search(m["wko_title"]))
|
|
is_aux = bool(AUX_RE.search(m["wko_title"]))
|
|
return (0 if is_kv and not is_aux else 1 if is_kv else 2,
|
|
-m["score"])
|
|
return sorted(ms, key=rank)[0]
|
|
|
|
rows = []
|
|
for v in oegb["variants"]:
|
|
group = v["variant_group_id"]
|
|
vid = v["variant_id"]
|
|
ms = matches_by_vid.get(vid, [])
|
|
m = primary_match(ms) if ms else None
|
|
# classification: explicit map > WKO match > heuristic
|
|
if group in group_chamber:
|
|
chamber = group_chamber[group]
|
|
method = "chambers.json portal_groups"
|
|
elif m:
|
|
chamber = "WKO"
|
|
method = "wko match"
|
|
else:
|
|
# second chance: fuzzy match the portal title against all
|
|
# current WKO docs (title morphology often defeats the primary
|
|
# matcher, e.g. 'Ang. bei Architekten und Ingenieurkonsulenten')
|
|
tv = _tokens(v["title"]) or _tokens(group.replace("-", " "))
|
|
best_d, best_s, best_i = None, 0.0, 0
|
|
for d2 in wko["docs"]:
|
|
if not d2.get("is_current"):
|
|
continue
|
|
tw2 = _tokens(d2["title"])
|
|
if not tw2:
|
|
continue
|
|
inter = _fuzzy_overlap(tv, tw2)
|
|
score = inter / min(len(tv), len(tw2))
|
|
if score > best_s:
|
|
best_d, best_s, best_i = d2, score, inter
|
|
if (best_s >= 0.75 and best_i >= 2) or best_s >= 0.999:
|
|
chamber = "WKO"
|
|
method = "wko fuzzy"
|
|
fuzzy_m = {"wko_slug": best_d["slug"],
|
|
"wko_title": best_d["title"],
|
|
"score": best_s}
|
|
else:
|
|
chamber = v.get("chamber_heuristic") or "unclassified"
|
|
method = "preamble heuristic"
|
|
fuzzy_m = None
|
|
|
|
main_source = None
|
|
cross_check = {"wko": None}
|
|
if chamber == "WKO":
|
|
info = chamber_info["WKO"]["main_source"]
|
|
main_source = {"chamber": "WKO", "type": "wko_db",
|
|
"public": True, "site": info["site"]}
|
|
if m:
|
|
wko_doc = wko_by_slug.get(m["wko_slug"], {})
|
|
wko_doc = wko_by_slug.get(m["wko_slug"], {})
|
|
main_source.update({
|
|
"url": m["wko_slug"] and
|
|
("https://www.wko.at/kollektivvertrag/" + m["wko_slug"]),
|
|
"wko_title": m["wko_title"],
|
|
"wko_lastmod": wko_doc.get("wko_lastmod"),
|
|
})
|
|
related = [x["wko_slug"] for x in ms
|
|
if x["wko_slug"] != m["wko_slug"]]
|
|
if related:
|
|
main_source["related_wko_docs"] = related
|
|
cross_check["wko"] = {"status": "matched",
|
|
"slug": m["wko_slug"],
|
|
"score": round(m["score"], 2)}
|
|
# date agreement: WKO title date vs ÖGB current slice version
|
|
wdate = wko_valid_date(m["wko_title"])
|
|
odates = [s["version"] for s in v.get("current_slices", [])
|
|
if s.get("version")]
|
|
odate = max(odates) if odates else None
|
|
if wdate and odate:
|
|
cross_check["wko"]["date_agreement"] = (
|
|
"match" if wdate == odate else
|
|
f"differs (wko {wdate} vs oegb {odate})")
|
|
else:
|
|
cross_check["wko"]["date_agreement"] = "not_comparable"
|
|
elif fuzzy_m:
|
|
main_source.update({
|
|
"url": "https://www.wko.at/kollektivvertrag/"
|
|
+ fuzzy_m["wko_slug"],
|
|
"wko_title": fuzzy_m["wko_title"],
|
|
})
|
|
cross_check["wko"] = {
|
|
"status": "fuzzy_matched",
|
|
"slug": fuzzy_m["wko_slug"],
|
|
"score": round(fuzzy_m["score"], 2),
|
|
"note": "manual review"}
|
|
else:
|
|
# WKO KV without portal counterpart (or low-confidence match)
|
|
low = None
|
|
for lc in match.get("low_confidence", []):
|
|
if lc.get("best_oegb_variant_id") == vid:
|
|
low = lc
|
|
break
|
|
cross_check["wko"] = {
|
|
"status": "low_confidence" if low else "missing_on_wko",
|
|
"slug": (low or {}).get("wko_slug"),
|
|
"note": "manual review" if low else
|
|
"no WKO doc matched this portal variant",
|
|
}
|
|
else:
|
|
info = chamber_info.get(chamber)
|
|
if info:
|
|
srcs = info.get("sources", [])
|
|
public = [s for s in srcs if s.get("public")]
|
|
main_source = {
|
|
"chamber": chamber,
|
|
"type": "chamber_site",
|
|
"public": bool(public),
|
|
"sources": [s["url"] for s in srcs[:3]],
|
|
}
|
|
cross_check["wko"] = None
|
|
|
|
rows.append({
|
|
"variant_id": vid,
|
|
"variant_group_id": group,
|
|
"title": v["title"],
|
|
"chamber": chamber,
|
|
"chamber_method": method,
|
|
"employment_types": v["employment_types"],
|
|
"provinces": v["provinces"],
|
|
"unions": v["unions"],
|
|
"valid_from": v["valid_from"],
|
|
"last_content_update": v["last_content_update"],
|
|
"current_slices": v["current_slices"],
|
|
"n_topics": v["n_topics"],
|
|
"main_source": main_source,
|
|
"cross_check": cross_check,
|
|
"only_in": "oegb_and_wko" if m else "oegb",
|
|
"text_files": {
|
|
"json": f"oegb/texts/{vid}.json",
|
|
"md": f"oegb/texts/{vid}.md",
|
|
},
|
|
})
|
|
|
|
# WKO-only current documents (info pages, Lohntafeln, Empfehlungen, KVs
|
|
# not mirrored on the portal)
|
|
for slug in sorted(unmatched_wko_slugs):
|
|
d = wko_by_slug.get(slug)
|
|
if not d or not d.get("is_current"):
|
|
continue
|
|
low = low_by_slug.get(slug)
|
|
rows.append({
|
|
"variant_id": None,
|
|
"variant_group_id": None,
|
|
"title": d["title"],
|
|
"chamber": "WKO",
|
|
"chamber_method": "wko only",
|
|
"employment_types": None,
|
|
"provinces": None,
|
|
"unions": None,
|
|
"valid_from": wko_valid_date(d["title"]),
|
|
"last_content_update": d.get("wko_lastmod"),
|
|
"current_slices": None,
|
|
"n_topics": None,
|
|
"main_source": {"chamber": "WKO", "type": "wko_db",
|
|
"public": True,
|
|
"url": d["url"],
|
|
"wko_title": d["title"],
|
|
"wko_lastmod": d.get("wko_lastmod")},
|
|
"cross_check": {"wko": None},
|
|
"only_in": "wko",
|
|
"text_files": {
|
|
"json": f"wko/texts/{slug}.json" if d.get("has_text") else None,
|
|
"md": f"wko/texts/{slug}.md" if d.get("has_text") else None,
|
|
},
|
|
"wko_slug": slug,
|
|
"wko_regions_current": d.get("current_regions"),
|
|
"low_confidence_note": (low or {}).get("best_oegb_title"),
|
|
})
|
|
|
|
# ---- summary
|
|
def count(key: str) -> dict:
|
|
out: dict[str, int] = {}
|
|
for r in rows:
|
|
out[r[key]] = out.get(r[key], 0) + 1
|
|
return dict(sorted(out.items(), key=lambda kv: -kv[1]))
|
|
|
|
flagged = [r for r in rows if r["cross_check"].get("wko")
|
|
and r["cross_check"]["wko"].get("status") in
|
|
("low_confidence", "missing_on_wko")
|
|
and r["chamber"] == "WKO"]
|
|
date_diffs = [r for r in rows if r["cross_check"].get("wko")
|
|
and str(r["cross_check"]["wko"].get("date_agreement", "")
|
|
).startswith("differs")]
|
|
|
|
catalog = {
|
|
"generated_at": dt.datetime.now(dt.timezone.utc)
|
|
.isoformat(timespec="seconds"),
|
|
"sources": {
|
|
"oegb_baseline": oegb["source"],
|
|
"wko_main": wko["source"],
|
|
"chambers": str(CHAMBERS.relative_to(REPO)),
|
|
},
|
|
"n_entries": len(rows),
|
|
"n_oegb_variants": oegb["n_variants"],
|
|
"n_wko_only": sum(1 for r in rows if r["only_in"] == "wko"),
|
|
"n_matched_pairs": len(match.get("matches", [])),
|
|
"chambers": count("chamber"),
|
|
"manual_review": {
|
|
"wko_without_clear_match": [
|
|
{"title": r["title"],
|
|
"variant_id": r["variant_id"],
|
|
"status": r["cross_check"]["wko"]["status"]}
|
|
for r in flagged],
|
|
"date_disagreements": [
|
|
{"variant_id": r["variant_id"], "title": r["title"],
|
|
"detail": r["cross_check"]["wko"]["date_agreement"]}
|
|
for r in date_diffs],
|
|
"unclassified": [
|
|
{"variant_id": r["variant_id"], "title": r["title"]}
|
|
for r in rows if r["chamber"] == "unclassified"],
|
|
},
|
|
"entries": rows,
|
|
}
|
|
OUT_JSON.write_text(json.dumps(catalog, ensure_ascii=False, indent=1),
|
|
encoding="utf-8")
|
|
|
|
fields = ["variant_id", "variant_group_id", "title", "chamber",
|
|
"employment_types", "provinces", "unions", "valid_from",
|
|
"last_content_update", "only_in", "main_source",
|
|
"wko_match_status", "date_agreement", "wko_slug"]
|
|
with OUT_CSV.open("w", newline="", encoding="utf-8") as fh:
|
|
writer = csv.DictWriter(fh, fieldnames=fields)
|
|
writer.writeheader()
|
|
for r in rows:
|
|
cc = (r.get("cross_check") or {}).get("wko") or {}
|
|
writer.writerow({
|
|
"variant_id": r["variant_id"] or "",
|
|
"variant_group_id": r["variant_group_id"] or "",
|
|
"title": r["title"],
|
|
"chamber": r["chamber"],
|
|
"employment_types": ";".join(r["employment_types"] or []),
|
|
"provinces": ";".join(r["provinces"] or []),
|
|
"unions": ";".join(r["unions"] or []),
|
|
"valid_from": r["valid_from"] or "",
|
|
"last_content_update": r["last_content_update"] or "",
|
|
"only_in": r["only_in"],
|
|
"main_source": (r["main_source"] or {}).get("url")
|
|
or (r["main_source"] or {}).get("site", ""),
|
|
"wko_match_status": cc.get("status", ""),
|
|
"date_agreement": cc.get("date_agreement", ""),
|
|
"wko_slug": r.get("wko_slug") or cc.get("slug", ""),
|
|
})
|
|
|
|
print(f"catalog: {len(rows)} entries "
|
|
f"({catalog['n_wko_only']} wko-only, "
|
|
f"{len(match.get('matches', []))} matched pairs)")
|
|
print("chambers:", catalog["chambers"])
|
|
print("manual review: wko_without_clear_match="
|
|
f"{len(flagged)}, date_disagreements={len(date_diffs)}, "
|
|
f"unclassified={len(catalog['manual_review']['unclassified'])}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |