#!/usr/bin/env python3
"""Fetch the WKO Kollektivvertrags-Datenbank (wko.at) as machine-readable data.
The Wirtschaftskammer Österreich publishes the Kollektivverträge of its
own sectors (Gewerbe/Handwerk, Industrie, Handel, ...) at
https://www.wko.at/oe/kollektivvertraege. Per the project's source
policy the WKO database is the MAIN source for WKO-negotiated KVs; the
ÖGB portal (fetch_kv_portal.py) is the cross-check baseline.
Site semantics (verified 2026-09-09):
- Per region there is a collectiveAgreementsOverview sitemap listing the
branch pages /{region}/kollektivvertraege/{branch-slug}.
- Each branch page lists the KV documents as entries; the Zeitraum filter on the
site treats archive=false as "aktuell".
- Detail pages live under /kollektivvertrag/{slug} (region-neutral) and
carry the full text, the räumlicher/fachlicher Geltungsbereich, and
links to Lohntafel-PDFs (linked, never downloaded - the wage tables
that are part of the KV text are converted with the text).
- The collectiveAgreementsDV sitemap provides a lastmod stamp per
detail page, which drives the --refresh diff.
Run from the repository root:
python3 personalverrechnung/tools/fetch_wko_kv.py # full fetch
python3 personalverrechnung/tools/fetch_wko_kv.py --limit 3 # smoke test
python3 personalverrechnung/tools/fetch_wko_kv.py --refresh
Outputs (curated, versioned):
personalverrechnung/quellen/kv/wko/catalog.json
personalverrechnung/quellen/kv/wko/texts/{slug}.json|.md
personalverrechnung/quellen/kv/wko/match-report.json
Raw mirror (unversioned): .firecrawl/wko-kv/
"""
from __future__ import annotations
import argparse
import concurrent.futures as cf
import csv
import datetime as dt
import html
import json
import re
import sys
import time
import urllib.request
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from fetch_kv_portal import FetchError, html_to_md, log # noqa: E402
REPO = Path(__file__).resolve().parents[2]
RAW = REPO / ".firecrawl" / "wko-kv"
OUT = REPO / "personalverrechnung" / "quellen" / "kv" / "wko"
TEXTS = OUT / "texts"
BASE = "https://www.wko.at"
REGIONS = ["oe", "bgld", "ktn", "noe", "ooe", "sbg", "stmk", "tirol",
"vlbg", "wien"]
REGION_NAMES = {"oe": "Österreich", "bgld": "Burgenland", "ktn": "Kärnten",
"noe": "Niederösterreich", "ooe": "Oberösterreich",
"sbg": "Salzburg", "stmk": "Steiermark", "tirol": "Tirol",
"vlbg": "Vorarlberg", "wien": "Wien"}
OVERVIEW_SITEMAP = BASE + "/sitemap/{region}-collectiveAgreementsOverview-sitemap-0.xml"
DV_SITEMAP = BASE + "/sitemap/oe-collectiveAgreementsDV-sitemap-0.xml"
DETAIL = BASE + "/kollektivvertrag/{slug}"
HEADERS = {
"User-Agent": "odoo-at-payroll-kv-library/1.0 (legal-source mirror; "
"contact: repo maintainer)",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.5",
"Accept-Language": "de-AT,de;q=0.9",
"Accept-Encoding": "identity",
}
WORKERS = 4
RETRIES = 3
TIMEOUT = 40
ENTRY_RE = re.compile(
r''
r']*>(.*?)', re.S)
# ------------------------------------------------------------- transport
def fetch(url: str, cache: Path | None = None, force: bool = False) -> str:
if cache is not None and cache.exists() and not force:
return cache.read_text(encoding="utf-8")
last_err: Exception | None = None
for attempt in range(RETRIES):
try:
req = urllib.request.Request(url, headers=HEADERS)
with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
data = resp.read().decode("utf-8", "replace")
if cache is not None:
cache.parent.mkdir(parents=True, exist_ok=True)
cache.write_text(data, encoding="utf-8")
return data
except Exception as exc: # noqa: BLE001
last_err = exc
time.sleep(2 * (attempt + 1))
raise FetchError(f"{url}: {last_err}")
def fetch_json_cached(url: str, cache: Path, force: bool = False) -> dict:
text = fetch(url, cache=cache, force=force)
if text.lstrip().startswith(" list[str]:
cache = RAW / "sitemaps" / f"{region}-overview.xml"
text = fetch(OVERVIEW_SITEMAP.format(region=region), cache=cache,
force=force)
urls = []
for loc in re.findall(r"([^<]+)", text):
url = html.unescape(loc).strip()
# skip the region landing page itself (no branch selected)
if url.rstrip("/").endswith("/kollektivvertraege"):
continue
urls.append(url)
return urls
def parse_overview(page: str) -> dict:
"""Extract branch title and document entries from a branch page."""
title = ""
m = re.search(r"(.*?)(?: - WKO)?", page, re.S)
if m:
title = re.sub(r"\s+", " ", html.unescape(m.group(1))).strip()
title = re.sub(r"^Kollektivverträge für\s*", "", title)
entries = []
for years, archive, href, inner in ENTRY_RE.findall(page):
text = re.sub(r"<[^>]+>", " ", inner)
text = re.sub(r"\s+", " ", html.unescape(text)).strip()
slug = href.rstrip("/").rsplit("/kollektivvertrag/", 1)[-1]
entries.append({
"slug": slug,
"href": href if href.startswith("http") else BASE + href,
"title": text,
"years": [y.strip() for y in years.split(",") if y.strip()],
"archived": archive == "true",
})
return {"branch_title": title, "entries": entries}
def fetch_overview(url: str, region: str, force: bool = False) -> dict:
slug = url.rstrip("/").rsplit("/", 1)[-1]
cache = RAW / "overviews" / region / f"{slug}.html"
page = fetch(url, cache=cache, force=force)
data = parse_overview(page)
data["url"] = url
data["region"] = region
return data
# ---------------------------------------------------------------- details
def dv_lastmods(force: bool = False) -> dict[str, str]:
"""Detail-page lastmod stamps from the DV sitemap (region-neutral)."""
cache = RAW / "sitemaps" / "oe-dv.xml"
text = fetch(DV_SITEMAP, cache=cache, force=force)
out = {}
for block in re.findall(r".*?", text, re.S):
loc = re.search(r"([^<]+)", block)
mod = re.search(r"([^<]+)", block)
if loc and mod:
slug = html.unescape(loc.group(1)).rstrip("/").rsplit(
"/kollektivvertrag/", 1)[-1]
out[slug] = mod.group(1)[:10]
return out
def parse_detail(page: str) -> dict:
title = ""
m = re.search(r"]*class=\"[^\"]*article-head-title[^\"]*\"[^>]*>"
r"(.*?)", page, re.S)
if m:
title = re.sub(r"\s+", " ", html.unescape(
re.sub(r"<[^>]+>", "", m.group(1)))).strip()
if not title:
m = re.search(r"(.*?)(?: - WKO)?", page, re.S)
if m:
title = re.sub(r"\s+", " ", html.unescape(m.group(1))).strip()
content_html = ""
m = re.search(r'data-gtm-section="Content"(.*?)', page, re.S)
if m:
content_html = m.group(1)
scope = {}
for key in ("Räumlicher Geltungsbereich", "Fachlicher Geltungsbereich"):
sm = re.search(key + r"[^:<]*:\s*([^<\n]{2,200})", content_html)
if sm:
scope[key.split(" ")[0].lower()] = html.unescape(
sm.group(1)).strip()
archived_banner = "ARCHIVIERT" in content_html or "archiviert" in title
pdf_links = []
for href in re.findall(r'href="([^"]+\.pdf[^"]*)"', content_html,
re.I):
url = href if href.startswith("http") else BASE + href
if url not in pdf_links:
pdf_links.append(url)
external: dict[str, str] = {}
for href, label in re.findall(
r']*href="(https?://[^"]+)"[^>]*>(.*?)',
content_html, re.S):
host = re.sub(r"^https?://([^/]+).*", r"\1", href)
if "wko.at" in host:
continue
label = re.sub(r"\s+", " ", html.unescape(
re.sub(r"<[^>]+>", "", label))).strip()
external.setdefault(host, label)
external_links = [f"{label} | {host}" for host, label in
external.items()]
return {
"title": title,
"scope": scope,
"archived_banner": archived_banner,
"pdf_links": pdf_links,
"external_links": external_links,
"content_html": content_html,
}
def fetch_detail(slug: str, force: bool = False) -> dict:
cache = RAW / "docs" / f"{slug}.html"
page = fetch(DETAIL.format(slug=slug), cache=cache, force=force)
data = parse_detail(page)
data["slug"] = slug
data["url"] = DETAIL.format(slug=slug)
return data
# --------------------------------------------------------------- pipeline
def collect(force: bool, limit: int | None, refresh: bool) -> None:
# refresh runs always re-check the overviews/sitemaps (archive status
# and listings must be fresh); a plain run honours --force only.
listing_force = force or refresh
overviews: list[dict] = []
failures: list[str] = []
jobs = []
for region in REGIONS:
for url in overview_urls(region, force=listing_force):
jobs.append((region, url))
if limit:
jobs = jobs[:limit]
log(f"overview pages to fetch: {len(jobs)}")
with cf.ThreadPoolExecutor(max_workers=WORKERS) as pool:
futs = {pool.submit(fetch_overview, url, region, listing_force):
(region, url) for region, url in jobs}
done = 0
for fut in cf.as_completed(futs):
region, url = futs[fut]
try:
overviews.append(fut.result())
done += 1
if done % 100 == 0:
log(f" overviews done: {done}/{len(jobs)}")
except Exception as exc: # noqa: BLE001
failures.append(f"overview {region} {url}: {exc}")
log(f"overviews fetched: {len(overviews)} (failed: {len(failures)})")
# ---- aggregate documents across regions
docs: dict[str, dict] = {}
for ov in overviews:
for e in ov["entries"]:
slug = e["slug"]
d = docs.setdefault(slug, {
"slug": slug, "url": e["href"], "title": e["title"],
"regions": {}, "branch_titles": set(),
})
d["regions"][ov["region"]] = {
"archived": e["archived"], "years": e["years"],
}
if ov["branch_title"]:
d["branch_titles"].add(ov["branch_title"])
if e["title"]:
d["title"] = e["title"]
for d in docs.values():
d["branch_titles"] = sorted(d["branch_titles"])
d["current_regions"] = sorted(
r for r, v in d["regions"].items() if not v["archived"])
d["is_current"] = bool(d["current_regions"])
lastmods = dv_lastmods(force=listing_force)
for d in docs.values():
d["wko_lastmod"] = lastmods.get(d["slug"])
log(f"unique documents listed: {len(docs)} "
f"(current somewhere: {sum(d['is_current'] for d in docs.values())})")
# ---- phase 2: fetch current detail pages
if refresh:
prev = _previous_catalog()
todo = [d["slug"] for d in docs.values() if d["is_current"] and (
d["slug"] not in prev
or prev[d["slug"]].get("wko_lastmod") != d["wko_lastmod"])]
log(f"refresh: {len(todo)} changed/new detail pages")
else:
todo = [d["slug"] for d in docs.values() if d["is_current"]]
if limit:
todo = todo[:limit]
details: dict[str, dict] = {}
with cf.ThreadPoolExecutor(max_workers=WORKERS) as pool:
futs = {pool.submit(fetch_detail, slug, force): slug for slug in todo}
done = 0
for fut in cf.as_completed(futs):
slug = futs[fut]
try:
details[slug] = fut.result()
done += 1
if done % 50 == 0:
log(f" details done: {done}/{len(todo)}")
except Exception as exc: # noqa: BLE001
failures.append(f"detail {slug}: {exc}")
log(f"details fetched: {len(details)} (failed: {len(failures)})")
# ---- write curated output
OUT.mkdir(parents=True, exist_ok=True)
TEXTS.mkdir(parents=True, exist_ok=True)
rows = []
for d in sorted(docs.values(), key=lambda x: x["slug"]):
det = details.get(d["slug"])
md = ""
if det:
content_html = det.pop("content_html", "")
md = html_to_md(content_html)
(TEXTS / f"{d['slug']}.json").write_text(
json.dumps(det, ensure_ascii=False, indent=1),
encoding="utf-8")
(TEXTS / f"{d['slug']}.md").write_text(md, encoding="utf-8")
rows.append({
"slug": d["slug"],
"url": d["url"],
"title": d["title"],
"branch_titles": d["branch_titles"],
"is_current": d["is_current"],
"current_regions": d["current_regions"],
"regions": d["regions"],
"wko_lastmod": d["wko_lastmod"],
"scope": det["scope"] if det else None,
"archived_banner": det["archived_banner"] if det else None,
"pdf_links": det["pdf_links"] if det else None,
"external_links": det["external_links"] if det else None,
"has_text": bool(det and md.strip()),
})
catalog = {
"generated_at": dt.datetime.now(dt.timezone.utc)
.isoformat(timespec="seconds"),
"source": BASE + "/oe/kollektivvertraege",
"n_docs": len(rows),
"n_current": sum(r["is_current"] for r in rows),
"docs": rows,
}
(OUT / "catalog.json").write_text(
json.dumps(catalog, ensure_ascii=False, indent=1), encoding="utf-8")
_write_csv(rows)
(RAW / "failures.log").write_text("\n".join(failures) + "\n",
encoding="utf-8")
log(f"done: {len(rows)} docs in catalog, {len(details)} texts, "
f"{len(failures)} failures")
# ---- phase 3: match against the ÖGB baseline catalog
match_report = _match_oegb(rows)
(OUT / "match-report.json").write_text(
json.dumps(match_report, ensure_ascii=False, indent=1),
encoding="utf-8")
log(f"match report: {match_report['n_matched']} matched, "
f"{match_report['n_low_confidence']} low-confidence, "
f"{match_report['n_unmatched_wko_current']} current WKO docs "
"without ÖGB match")
def _write_csv(rows: list[dict]) -> None:
fields = ["slug", "title", "is_current", "current_regions",
"wko_lastmod", "has_text", "url"]
with (OUT / "catalog.csv").open("w", newline="", encoding="utf-8") as fh:
writer = csv.DictWriter(fh, fieldnames=fields, extrasaction="ignore")
writer.writeheader()
for r in rows:
r2 = dict(r)
r2["current_regions"] = ";".join(r2["current_regions"])
writer.writerow(r2)
def _previous_catalog() -> dict[str, dict]:
path = OUT / "catalog.json"
if not path.exists():
return {}
return {d["slug"]: d for d in
json.loads(path.read_text(encoding="utf-8"))["docs"]}
STOPWORDS = {
"kollektivvertrag", "rahmenkollektivvertrag", "zusatzkollektivvertrag",
"rahmen", "fuer", "die", "der", "das", "und", "sowie", "im",
"in", "des", "dem", "den", "gueltig", "ab", "kv", "lohn",
"gehaltsordnung", "lohnordnung", "gehalt", "lohntafel", "ordnung",
"tafel", "information", "ergebnis", "abschluss",
}
def _tokens(title: str) -> set[str]:
t = html.unescape(title or "").lower()
t = (t.replace("ä", "ae").replace("ö", "oe").replace("ü", "ue")
.replace("ß", "ss"))
words = re.findall(r"[a-z0-9]+", t)
return {w for w in words if w not in STOPWORDS and len(w) > 1}
def _fuzzy_overlap(a: set[str], b: set[str]) -> int:
"""Count token pairs where one is a prefix of the other (min len 4),
so German morphology (Florist/Floristen, Gutsbetrieb/-betrieben)
does not defeat matching."""
used: set[str] = set()
n = 0
for x in a:
for y in b:
if y in used:
continue
if (len(x) >= 4 and len(y) >= 4
and (x.startswith(y) or y.startswith(x))):
used.add(y)
n += 1
break
return n
def _match_oegb(wko_rows: list[dict]) -> dict:
oegb_path = (REPO / "personalverrechnung" / "quellen" / "kv" / "oegb"
/ "catalog.json")
if not oegb_path.exists():
log("no ÖGB baseline catalog yet - skipping match step "
"(run fetch_kv_portal.py first)")
return {"n_matched": 0, "n_low_confidence": 0,
"n_unmatched_wko_current": 0, "matches": [],
"note": "oegb catalog missing"}
oegb = json.loads(oegb_path.read_text(encoding="utf-8"))["variants"]
index = [(v, _tokens(v["title"])) for v in oegb]
matches = []
matched_oegb: set[str] = set()
low, unmatched = [], []
for row in wko_rows:
if not row["is_current"]:
continue
tw = _tokens(row["title"])
if not tw:
unmatched.append({"slug": row["slug"], "title": row["title"],
"reason": "no title tokens"})
continue
best, best_score, best_inter, best_vid = None, 0.0, 0, None
for v, tv in index:
if not tv:
continue
inter = _fuzzy_overlap(tw, tv)
score = inter / min(len(tw), len(tv))
if score > best_score:
best, best_score, best_inter, best_vid = v, score, inter, \
v["variant_id"]
# matched: strong multi-token overlap, or full containment of a
# very short title (e.g. 'Floristen' in 'Floristen und
# Blumeneinzelhaendler')
if (best_score >= 0.75 and best_inter >= 2) or best_score >= 0.999:
matches.append({"wko_slug": row["slug"], "wko_title": row["title"],
"oegb_variant_id": best_vid,
"oegb_title": best["title"], "score": best_score})
matched_oegb.add(best_vid)
elif best_score >= 0.5 and best_inter >= 2:
low.append({"wko_slug": row["slug"], "wko_title": row["title"],
"best_oegb_variant_id": best_vid,
"best_oegb_title": best["title"] if best else None,
"score": best_score})
else:
unmatched.append({"slug": row["slug"], "title": row["title"],
"reason": f"best score {best_score:.2f}"})
oegb_unmatched = [v["variant_id"] for v, _ in index
if v["variant_id"] not in matched_oegb]
return {
"n_matched": len(matches),
"n_low_confidence": len(low),
"n_unmatched_wko_current": len(unmatched),
"n_oegb_without_wko_match": len(oegb_unmatched),
"matches": matches,
"low_confidence": low,
"unmatched_wko_current": unmatched,
"oegb_without_wko_match": oegb_unmatched,
}
def main() -> None:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--force", action="store_true")
ap.add_argument("--limit", type=int, default=None)
ap.add_argument("--refresh", action="store_true",
help="only refetch detail pages whose wko lastmod "
"changed; overviews are always re-checked")
args = ap.parse_args()
collect(force=args.force, limit=args.limit, refresh=args.refresh)
if __name__ == "__main__":
main()