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.
793 lines
30 KiB
Python
793 lines
30 KiB
Python
#!/usr/bin/env python3
|
|
"""Build the machine-readable KV library from kollektivvertrag.at (ÖGB-Verlag).
|
|
|
|
The portal of the Austrian trade-union federation (ÖGB) publishes every
|
|
currently valid Kollektivvertrag as consolidated text. It is the only
|
|
complete, all-chambers source and therefore serves as the cross-check
|
|
baseline for the chamber-specific main sources (WKO, Ärztekammer,
|
|
Apoothekerkammer, Rechtsanwaltskammer, Notariatskammer, LWK, public
|
|
sector) that are matched against it in kv-catalog.json.
|
|
|
|
Run from the repository root:
|
|
|
|
python3 personalverrechnung/tools/fetch_kv_portal.py # full fetch
|
|
python3 personalverrechnung/tools/fetch_kv_portal.py --limit 3 # smoke test
|
|
python3 personalverrechnung/tools/fetch_kv_portal.py --build-only
|
|
python3 personalverrechnung/tools/fetch_kv_portal.py --refresh # update run
|
|
|
|
Outputs (curated, versioned):
|
|
personalverrechnung/quellen/kv/oegb/catalog.json|.csv
|
|
personalverrechnung/quellen/kv/oegb/texts/<variant-id>.json|.md
|
|
personalverrechnung/quellen/kv/oegb/update-reports/update-<date>.json
|
|
|
|
Raw API mirror (unversioned, .firecrawl/ is gitignored):
|
|
.firecrawl/kv-portal/sitemap.xml
|
|
.firecrawl/kv-portal/groups/<group-id>.json|html
|
|
.firecrawl/kv-portal/variants/<variant-id>/structure.json
|
|
.firecrawl/kv-portal/variants/<variant-id>/slices/<slice-id>.json
|
|
.firecrawl/kv-portal/variants/<variant-id>/topics/<topic-id>.json
|
|
|
|
Semantics of the portal API (verified 2026-09-09):
|
|
- sitemap.xml lists one dashboard URL per variant-group/variant.
|
|
- Each dashboard page embeds data-variant-group-json with every variant
|
|
of the group: provinces, employmentTypes (arb/ang), unions, validity
|
|
(validFrom/lastContentUpdate) and the topic ids.
|
|
- kvfulltext.kv-structure.json?structure-id=<variant-id> returns the
|
|
doc-set tree; its children are KV_TIME_SLICE parts whose ids embed the
|
|
version date (<paraid>_<YYYYMMDD>). The latest date per slice name is
|
|
the current version of that part; older entries are history.
|
|
- kvfulltext.kv-slices.json?slice-id=<id> returns the paragraph tree
|
|
(titleHtml/contentHtml) of one slice.
|
|
- kvdashboard.kv-topics.json?topic-id=<id> returns the dashboard topic
|
|
content (the portal's own "current" view per theme).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import concurrent.futures as cf
|
|
import csv
|
|
import datetime as dt
|
|
import html
|
|
import html.parser
|
|
import json
|
|
import re
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
RAW = REPO / ".firecrawl" / "kv-portal"
|
|
OUT = REPO / "personalverrechnung" / "quellen" / "kv" / "oegb"
|
|
TEXTS = OUT / "texts"
|
|
REPORTS = OUT / "update-reports"
|
|
|
|
BASE = "https://www.kollektivvertrag.at"
|
|
DASHBOARD = BASE + "/kv?variant-group-id={group}&variant-id={variant}&topic-id=1"
|
|
STRUCTURE = (BASE + "/content/kollektivvertrag/at/de/system/kv/fulltext/"
|
|
"jcr:content/root/container/container/kvfulltext.kv-structure.json"
|
|
"?structure-id={variant}")
|
|
SLICES = (BASE + "/content/kollektivvertrag/at/de/system/kv/fulltext/"
|
|
"jcr:content/root/container/container/kvfulltext.kv-slices.json"
|
|
"?slice-id={slice}")
|
|
TOPICS = (BASE + "/content/kollektivvertrag/at/de/system/kv/dashboard/"
|
|
"jcr:content/root/container/container/kvdashboard.kv-topics.json"
|
|
"?topic-id={topic}")
|
|
|
|
HEADERS = {
|
|
"User-Agent": "odoo-at-payroll-kv-library/1.0 (legal-source mirror; "
|
|
"contact: repo maintainer)",
|
|
"Accept": "application/json,text/html;q=0.9,*/*;q=0.5",
|
|
"Accept-Language": "de-AT,de;q=0.9",
|
|
"Accept-Encoding": "identity",
|
|
}
|
|
|
|
PROVINCE_NAMES = {
|
|
"b": "Burgenland", "k": "Kärnten", "nö": "Niederösterreich",
|
|
"oö": "Oberösterreich", "s": "Salzburg", "st": "Steiermark",
|
|
"t": "Tirol", "v": "Vorarlberg", "w": "Wien",
|
|
}
|
|
EMPLOYMENT_TYPE_NAMES = {"arb": "Arbeiter:innen", "ang": "Angestellte"}
|
|
UNION_NAMES = {
|
|
"gpa": "GPA-djp", "gpa-djp": "GPA-djp", "younion": "younion",
|
|
"vida": "vida", "gbh": "Gewerkschaft Bau-Holz",
|
|
"proge": "PRO-GE", "pro-ge": "PRO-GE",
|
|
"gpf": "Gewerkschaft der Post- und Fernmeldebediensteten",
|
|
"goed": "Gewerkschaft Öffentlicher Dienst (GÖD)",
|
|
}
|
|
|
|
|
|
def union_name(u: str) -> str:
|
|
return UNION_NAMES.get(u, UNION_NAMES.get(u.replace("-", ""), u))
|
|
# Employer-side chambers that can be recognised in the consolidated texts.
|
|
CHAMBER_PATTERNS = [
|
|
("Aerztekammer", re.compile(r"Ärztekammer|Ärzt(innen)?kammer", re.I)),
|
|
("Apoothekerkammer", re.compile(r"Apoothekerkammer", re.I)),
|
|
("Rechtsanwaltskammer", re.compile(r"Rechtsanw(a|ä)ltskammer", re.I)),
|
|
("Notariatskammer", re.compile(r"Notariatskammer", re.I)),
|
|
("Landwirtschaftskammer", re.compile(r"Landwirtschaftskammer|L(a|ä)ndliche\s+Genossenschaften", re.I)),
|
|
("Staedtebund_Gemeindebund", re.compile(r"Städtebund|Gemeindebund|Stadtgemeinden", re.I)),
|
|
("WKO", re.compile(r"Wirtschaftskammer|Wirtschaftskammern|Fachverband|Bundesinnung|Landesinnung|Landessektion|Bundesgruppe", re.I)),
|
|
]
|
|
|
|
WORKERS = 4
|
|
RETRIES = 3
|
|
TIMEOUT = 40
|
|
|
|
|
|
class FetchError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def log(msg: str) -> None:
|
|
print(msg, flush=True)
|
|
|
|
|
|
def fetch(url: str, binary: bool = False, force: bool = False,
|
|
cache: Path | None = None) -> bytes | str:
|
|
"""GET a URL with disk cache; returns text (or bytes for binary=True)."""
|
|
if cache is not None and cache.exists() and not force:
|
|
return cache.read_bytes() if binary else 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()
|
|
if cache is not None:
|
|
cache.parent.mkdir(parents=True, exist_ok=True)
|
|
cache.write_bytes(data)
|
|
return data if binary else data.decode("utf-8", "replace")
|
|
except Exception as exc: # noqa: BLE001 - retry any transport error
|
|
last_err = exc
|
|
time.sleep(2 * (attempt + 1))
|
|
raise FetchError(f"{url}: {last_err}")
|
|
|
|
|
|
def fetch_json(url: str, cache: Path, force: bool = False) -> dict:
|
|
raw = fetch(url, cache=cache, force=force)
|
|
text = raw if isinstance(raw, str) else raw.decode("utf-8", "replace")
|
|
stripped = text.lstrip()
|
|
if stripped.startswith("<!DOCTYPE") or stripped.startswith("<html"):
|
|
raise FetchError(f"{url}: HTML error page instead of JSON")
|
|
return json.loads(text)
|
|
|
|
|
|
# ---------------------------------------------------------------- sitemap
|
|
|
|
|
|
def parse_sitemap(force: bool = False) -> list[dict]:
|
|
"""Return [{'group': ..., 'variant': ...}] from the portal sitemap."""
|
|
cache = RAW / "sitemap.xml"
|
|
text = fetch(BASE + "/sitemap.xml", cache=cache, force=force)
|
|
entries = []
|
|
for loc in re.findall(r"<loc>([^<]+)</loc>", text):
|
|
url = html.unescape(loc)
|
|
if "/kv?variant-group-id=" not in url:
|
|
continue
|
|
m = re.search(r"variant-group-id=([^&]+)&variant-id=([^&]+)", url)
|
|
if m:
|
|
entries.append({"group": m.group(1), "variant": m.group(2),
|
|
"url": url})
|
|
return entries
|
|
|
|
|
|
# ------------------------------------------------------------- dashboard
|
|
|
|
|
|
def fetch_group(group_id: str, variant_id: str, force: bool = False) -> dict:
|
|
"""Fetch one dashboard page and return its data-variant-group-json."""
|
|
cache_html = RAW / "groups" / f"{group_id}.html"
|
|
cache_json = RAW / "groups" / f"{group_id}.json"
|
|
if cache_json.exists() and not force:
|
|
return json.loads(cache_json.read_text(encoding="utf-8"))
|
|
page = fetch(DASHBOARD.format(group=group_id, variant=variant_id),
|
|
cache=cache_html, force=force)
|
|
m = re.search(r"<ex-kv-dashboard[^>]*>", page)
|
|
if not m:
|
|
raise FetchError(f"dashboard {group_id}: no ex-kv-dashboard element")
|
|
tag = m.group(0)
|
|
m = re.search(r'data-variant-group-json="([^"]*)"', tag)
|
|
if not m:
|
|
raise FetchError(f"dashboard {group_id}: no variant-group json")
|
|
data = json.loads(html.unescape(m.group(1)))
|
|
cache_json.parent.mkdir(parents=True, exist_ok=True)
|
|
cache_json.write_text(json.dumps(data, ensure_ascii=False, indent=1),
|
|
encoding="utf-8")
|
|
return data
|
|
|
|
|
|
# ---------------------------------------------------- structure & slices
|
|
|
|
|
|
def slice_version_date(slice_id: str) -> str:
|
|
"""'328261_20191201' -> '20191201' (may be empty for odd ids)."""
|
|
m = re.search(r"_(\d{8})$", slice_id)
|
|
return m.group(1) if m else ""
|
|
|
|
|
|
def fetch_structure(variant_id: str, force: bool = False) -> dict:
|
|
return fetch_json(
|
|
STRUCTURE.format(variant=variant_id),
|
|
RAW / "variants" / variant_id / "structure.json", force)
|
|
|
|
|
|
def fetch_slice(variant_id: str, slice_id: str, force: bool = False) -> dict:
|
|
return fetch_json(
|
|
SLICES.format(slice=slice_id),
|
|
RAW / "variants" / variant_id / "slices" / f"{slice_id}.json", force)
|
|
|
|
|
|
def fetch_topic(variant_id: str, topic_id: str, force: bool = False) -> dict:
|
|
return fetch_json(
|
|
TOPICS.format(topic=topic_id),
|
|
RAW / "variants" / variant_id / "topics" / f"{topic_id}.json", force)
|
|
|
|
|
|
# ------------------------------------------------------ html -> markdown
|
|
|
|
|
|
class _MD(html.parser.HTMLParser):
|
|
"""Conservative HTML-to-Markdown renderer for ÖGB-Verlag fragments.
|
|
|
|
The JSON keeps the original htmlContent; this rendering exists for
|
|
review/diffing, not as the primary machine format.
|
|
"""
|
|
|
|
HEADINGS = {"h1": "# ", "h2": "## ", "h3": "### ", "h4": "#### ",
|
|
"h5": "##### ", "h6": "###### "}
|
|
BLOCKS = {"p", "div", "li", "tr", "table", "ul", "ol", "h1", "h2", "h3",
|
|
"h4", "h5", "h6", "br", "blockquote"}
|
|
SKIP = {"script", "style", "svg"}
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__(convert_charrefs=True)
|
|
self.out: list[str] = []
|
|
self._buf: list[str] = []
|
|
self._list_depth = 0
|
|
self._cell: list[str] | None = None
|
|
self._row: list[str] | None = None
|
|
self._table: list[list[str]] | None = None
|
|
self._skip = 0
|
|
|
|
# -- buffer handling ------------------------------------------------
|
|
def _emit(self, text: str) -> None:
|
|
if self._cell is not None:
|
|
self._cell.append(text)
|
|
else:
|
|
self._buf.append(text)
|
|
|
|
def _flush_block(self, prefix: str = "") -> None:
|
|
text = "".join(self._buf)
|
|
self._buf = []
|
|
text = re.sub(r"[ \t\u00a0]+", " ", text).strip()
|
|
if text or prefix:
|
|
self.out.append(prefix + text)
|
|
|
|
def handle_starttag(self, tag, attrs):
|
|
if tag in self.SKIP:
|
|
self._skip += 1
|
|
return
|
|
if self._skip:
|
|
return
|
|
if tag in ("b", "strong"):
|
|
self._emit("**")
|
|
elif tag in ("i", "em"):
|
|
self._emit("*")
|
|
elif tag == "u":
|
|
self._emit("__")
|
|
elif tag == "br":
|
|
if self._cell is None:
|
|
self._flush_block()
|
|
elif tag == "a":
|
|
href = dict(attrs).get("href", "")
|
|
if href:
|
|
self._emit("\x01")
|
|
self._emit(f"\x02{href}\x02")
|
|
elif tag in ("ul", "ol"):
|
|
self._list_depth += 1
|
|
elif tag == "li":
|
|
self._flush_block()
|
|
elif tag == "table":
|
|
self._flush_block()
|
|
self._table = []
|
|
elif tag == "tr":
|
|
self._row = []
|
|
elif tag in ("td", "th"):
|
|
self._cell = []
|
|
elif tag in self.HEADINGS:
|
|
self._flush_block(self.HEADINGS[tag])
|
|
elif tag in self.BLOCKS:
|
|
self._flush_block()
|
|
|
|
def handle_endtag(self, tag):
|
|
if tag in self.SKIP:
|
|
self._skip = max(0, self._skip - 1)
|
|
return
|
|
if self._skip:
|
|
return
|
|
if tag in ("b", "strong"):
|
|
self._emit("**")
|
|
elif tag in ("i", "em"):
|
|
self._emit("*")
|
|
elif tag == "u":
|
|
self._emit("__")
|
|
elif tag in ("ul", "ol"):
|
|
self._list_depth = max(0, self._list_depth - 1)
|
|
elif tag == "li":
|
|
bullet = " " * (self._list_depth - 1) + "- "
|
|
self._flush_block(bullet)
|
|
elif tag in ("td", "th"):
|
|
if self._cell is not None and self._row is not None:
|
|
cell = re.sub(r"\s+", " ", "".join(self._cell)).strip()
|
|
cell = cell.replace("|", "\\|")
|
|
self._row.append(cell)
|
|
self._cell = None
|
|
elif tag == "tr":
|
|
if self._row is not None and self._table is not None:
|
|
self._table.append(self._row)
|
|
self._row = None
|
|
elif tag == "table":
|
|
if self._table:
|
|
width = max(len(r) for r in self._table)
|
|
for r in self._table:
|
|
r.extend([""] * (width - len(r)))
|
|
lines = ["| " + " | ".join(self._table[0]) + " |",
|
|
"|" + "---|" * width]
|
|
lines += ["| " + " | ".join(r) + " |" for r in self._table[1:]]
|
|
self.out.extend(lines)
|
|
self._table = None
|
|
elif tag in self.HEADINGS:
|
|
self._flush_block()
|
|
elif tag in self.BLOCKS:
|
|
self._flush_block()
|
|
|
|
def handle_data(self, data):
|
|
if not self._skip:
|
|
self._emit(data)
|
|
|
|
def result(self) -> str:
|
|
self._flush_block()
|
|
text = "\n\n".join(p for p in self.out if p.strip())
|
|
# resolve link markers produced by handle_starttag('a')
|
|
text = re.sub(r"\x01([^\x01\x02]*)\x02([^\x02]*)\x02", r"\1 (\2)", text)
|
|
text = text.replace("\x01", "").replace("\x02", "")
|
|
return re.sub(r"\n{3,}", "\n\n", text).strip()
|
|
|
|
|
|
def html_to_md(fragment: str) -> str:
|
|
if not fragment or not fragment.strip():
|
|
return ""
|
|
parser = _MD()
|
|
try:
|
|
parser.feed(fragment)
|
|
parser.close()
|
|
return parser.result()
|
|
except Exception: # noqa: BLE001 - never fail a render
|
|
return re.sub(r"<[^>]+>", " ", fragment).strip()
|
|
|
|
|
|
def title_of(title_html: str) -> str:
|
|
text = re.sub(r"<[^>]+>", " ", title_html or "")
|
|
return re.sub(r"\s+", " ", html.unescape(text)).strip()
|
|
|
|
|
|
# --------------------------------------------------------- curation
|
|
|
|
|
|
def current_slices(structure: dict) -> list[dict]:
|
|
"""Latest version per slice name, in structure order."""
|
|
main = structure.get("main", structure)
|
|
latest: dict[str, dict] = {}
|
|
order: list[str] = []
|
|
for child in (main.get("children") or []):
|
|
link = child.get("link", {})
|
|
if link.get("type") != "KV_TIME_SLICE":
|
|
continue
|
|
name = link.get("name") or link.get("id")
|
|
node = {
|
|
"slice_id": link.get("id"),
|
|
"name": name,
|
|
"valid_from": (child.get("validFrom") or "")[:10],
|
|
"version": slice_version_date(link.get("id") or ""),
|
|
}
|
|
if name not in latest:
|
|
order.append(name)
|
|
latest[name] = node
|
|
elif node["version"] >= latest[name]["version"]:
|
|
latest[name] = node
|
|
return [latest[n] for n in order]
|
|
|
|
|
|
def all_slice_ids(structure: dict) -> list[dict]:
|
|
main = structure.get("main", structure)
|
|
out = []
|
|
for child in (main.get("children") or []):
|
|
link = child.get("link", {})
|
|
if link.get("type") == "KV_TIME_SLICE":
|
|
out.append({"slice_id": link.get("id"), "name": link.get("name"),
|
|
"valid_from": (child.get("validFrom") or "")[:10]})
|
|
return out
|
|
|
|
|
|
def classify_chamber(texts: list[str]) -> str:
|
|
for chamber, pattern in CHAMBER_PATTERNS:
|
|
for t in texts:
|
|
if t and pattern.search(t):
|
|
return chamber
|
|
return "unclassified"
|
|
|
|
|
|
def collect_text(node: dict, acc: list[str], limit: int = 6000) -> None:
|
|
acc.append(title_of(node.get("titleHtml") or ""))
|
|
acc.append(re.sub(r"<[^>]+>", " ", html.unescape(
|
|
node.get("contentHtml") or "")))
|
|
for child in (node.get("children") or [])[:6]:
|
|
collect_text(child, acc, limit)
|
|
if sum(len(a) for a in acc) > limit:
|
|
return
|
|
|
|
|
|
def build_variant_doc(group_data: dict, variant: dict, structure: dict,
|
|
slices: dict[str, dict], topics: dict[str, dict],
|
|
fetched_at: str) -> dict:
|
|
fm = variant.get("facetMapping", {})
|
|
props = variant.get("properties", {})
|
|
validity = props.get("validity", {})
|
|
cur = current_slices(structure)
|
|
current_ids = {c["slice_id"] for c in cur}
|
|
|
|
doc = {
|
|
"variant_id": variant.get("id"),
|
|
"variant_group_id": group_data.get("id"),
|
|
"title": variant.get("name") or group_data.get("title"),
|
|
"employment_types": fm.get("employmentTypes", []),
|
|
"employment_types_names": [EMPLOYMENT_TYPE_NAMES.get(e, e)
|
|
for e in fm.get("employmentTypes", [])],
|
|
"provinces": fm.get("provinces", []),
|
|
"provinces_names": [PROVINCE_NAMES.get(p, p)
|
|
for p in fm.get("provinces", [])],
|
|
"unions": props.get("unions", []),
|
|
"unions_names": [union_name(u) for u in props.get("unions", [])],
|
|
"validity": {
|
|
"valid_from": (validity.get("validFrom") or "")[:10],
|
|
"last_content_update": (validity.get("lastContentUpdate") or "")[:10],
|
|
},
|
|
"fetched_at": fetched_at,
|
|
"source_urls": {
|
|
"dashboard": DASHBOARD.format(group=group_data.get("id"),
|
|
variant=variant.get("id")),
|
|
"volltext": BASE + "/volltext?doc-set-id=" + str(variant.get("id")),
|
|
"portal": BASE,
|
|
},
|
|
"chamber": None, # filled by the cross-check/catalog step
|
|
"slices": cur,
|
|
"slice_history": [s for s in all_slice_ids(structure)
|
|
if s["slice_id"] not in current_ids],
|
|
"topics": [{"id": t.get("id"), "title": t.get("title")}
|
|
for t in variant.get("topics", [])],
|
|
}
|
|
|
|
# chamber heuristic from the head of the current texts
|
|
head_texts = []
|
|
for cid in list(current_ids)[:2]:
|
|
sl = slices.get(cid, {})
|
|
acc: list[str] = []
|
|
for child in (sl.get("children") or [])[:4]:
|
|
collect_text(child, acc)
|
|
head_texts.append(" ".join(acc))
|
|
doc["chamber_heuristic"] = classify_chamber(head_texts)
|
|
|
|
# full content
|
|
doc["content"] = {
|
|
"slices": {cid: slices[cid] for cid in sorted(current_ids)
|
|
if cid in slices},
|
|
"topics": {tid: topics[tid] for tid in sorted(topics)
|
|
if tid in topics},
|
|
}
|
|
return doc
|
|
|
|
|
|
def render_markdown(doc: dict) -> str:
|
|
lines = [
|
|
f"# {doc['title']}",
|
|
"",
|
|
f"- **Variante:** `{doc['variant_id']}` "
|
|
f"(Gruppe `{doc['variant_group_id']}`)",
|
|
f"- **Anstellungsart:** {', '.join(doc['employment_types_names']) or 'n/a'}",
|
|
f"- **Bundesländer:** {', '.join(doc['provinces_names']) or 'Österreichweit'}",
|
|
f"- **Gewerkschaft:** {', '.join(doc['unions_names']) or 'n/a'}",
|
|
f"- **Gültig ab / Stand:** {doc['validity']['valid_from']} / "
|
|
f"{doc['validity']['last_content_update']}",
|
|
f"- **Quelle:** {doc['source_urls']['dashboard']}",
|
|
f"- **Abgerufen:** {doc['fetched_at']}",
|
|
"",
|
|
]
|
|
for s in doc["slices"]:
|
|
lines += [f"## Teil: {s['name']} "
|
|
f"(Version {s['version']}, gültig ab {s['valid_from']})",
|
|
""]
|
|
sl = doc["content"]["slices"].get(s["slice_id"], {})
|
|
lines.append(_render_slice_children(sl))
|
|
lines.append("")
|
|
for t in doc["topics"]:
|
|
body = doc["content"]["topics"].get(t["id"], {})
|
|
items = body.get("items") or []
|
|
if not items:
|
|
continue
|
|
lines += [f"## Thema: {t['title']}", ""]
|
|
for item in items:
|
|
md = html_to_md(item.get("htmlContent") or "")
|
|
if md:
|
|
lines += [md, ""]
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def _render_slice_children(slice_doc: dict, depth: int = 0) -> str:
|
|
out = []
|
|
for child in (slice_doc.get("children") or []):
|
|
title = title_of(child.get("titleHtml") or "")
|
|
if title:
|
|
level = "#" * min(depth + 2, 6)
|
|
out.append(f"{level} {title}")
|
|
out.append("")
|
|
body = html_to_md(child.get("contentHtml") or "")
|
|
if body:
|
|
out.append(body)
|
|
out.append("")
|
|
sub = _render_slice_children(child, depth + 1)
|
|
if sub:
|
|
out.append(sub)
|
|
return "\n".join(out)
|
|
|
|
|
|
# ------------------------------------------------------------- pipeline
|
|
|
|
|
|
def variant_jobs(group_datas: dict[str, dict]) -> list[dict]:
|
|
"""All variants to process, deduped by variant id."""
|
|
jobs: dict[str, dict] = {}
|
|
for group_id, data in group_datas.items():
|
|
variants = list((data.get("variants") or {}).get("items") or [])
|
|
if data.get("variant"):
|
|
variants.append(data["variant"])
|
|
for v in variants:
|
|
vid = v.get("id")
|
|
if vid and vid not in jobs:
|
|
jobs[vid] = {"group": group_id, "variant": v}
|
|
return list(jobs.values())
|
|
|
|
|
|
def process_variant(job: dict, group_data: dict, force: bool) -> dict:
|
|
variant = job["variant"]
|
|
vid = variant["id"]
|
|
structure = fetch_structure(vid, force)
|
|
slice_docs = {}
|
|
for s in all_slice_ids(structure):
|
|
slice_docs[s["slice_id"]] = fetch_slice(vid, s["slice_id"], force)
|
|
topic_docs = {}
|
|
for t in variant.get("topics", []):
|
|
tid = t.get("id")
|
|
if tid:
|
|
topic_docs[tid] = fetch_topic(vid, tid, force)
|
|
fetched_at = dt.date.today().isoformat()
|
|
return build_variant_doc(group_data, variant, structure, slice_docs,
|
|
topic_docs, fetched_at)
|
|
|
|
|
|
def write_outputs(docs: list[dict]) -> None:
|
|
if not docs:
|
|
raise SystemExit("no variant docs to write (fetch failures?)")
|
|
TEXTS.mkdir(parents=True, exist_ok=True)
|
|
catalog = []
|
|
for doc in docs:
|
|
vid = doc["variant_id"]
|
|
(TEXTS / f"{vid}.json").write_text(
|
|
json.dumps(doc, ensure_ascii=False, indent=1), encoding="utf-8")
|
|
(TEXTS / f"{vid}.md").write_text(render_markdown(doc), encoding="utf-8")
|
|
catalog.append({
|
|
"variant_group_id": doc["variant_group_id"],
|
|
"variant_id": vid,
|
|
"title": doc["title"],
|
|
"employment_types": doc["employment_types"],
|
|
"provinces": doc["provinces"],
|
|
"unions": doc["unions"],
|
|
"valid_from": doc["validity"]["valid_from"],
|
|
"last_content_update": doc["validity"]["last_content_update"],
|
|
"chamber_heuristic": doc["chamber_heuristic"],
|
|
"current_slices": [{"name": s["name"], "slice_id": s["slice_id"],
|
|
"version": s["version"]}
|
|
for s in doc["slices"]],
|
|
"n_slice_history": len(doc["slice_history"]),
|
|
"n_topics": len(doc["topics"]),
|
|
"dashboard_url": doc["source_urls"]["dashboard"],
|
|
})
|
|
catalog.sort(key=lambda e: (e["variant_group_id"], e["variant_id"]))
|
|
OUT.mkdir(parents=True, exist_ok=True)
|
|
(OUT / "catalog.json").write_text(
|
|
json.dumps({"generated_at": dt.datetime.now(dt.timezone.utc)
|
|
.isoformat(timespec="seconds"),
|
|
"source": BASE,
|
|
"n_variants": len(catalog),
|
|
"variants": catalog},
|
|
ensure_ascii=False, indent=1), encoding="utf-8")
|
|
with (OUT / "catalog.csv").open("w", newline="",
|
|
encoding="utf-8") as fh:
|
|
writer = csv.DictWriter(fh, fieldnames=list(catalog[0].keys()))
|
|
writer.writeheader()
|
|
writer.writerows(catalog)
|
|
|
|
|
|
def run(force: bool, limit: int | None, build_only: bool) -> None:
|
|
if build_only:
|
|
docs = _load_docs_from_raw()
|
|
write_outputs(docs)
|
|
log(f"built {len(docs)} variant docs from raw mirror")
|
|
return
|
|
entries = parse_sitemap(force)
|
|
groups: dict[str, str] = {}
|
|
for e in entries:
|
|
groups.setdefault(e["group"], e["variant"])
|
|
log(f"sitemap: {len(entries)} urls, {len(groups)} unique groups")
|
|
ids = list(groups.items())
|
|
if limit:
|
|
ids = ids[:limit]
|
|
group_datas = {}
|
|
failures: list[str] = []
|
|
|
|
with cf.ThreadPoolExecutor(max_workers=WORKERS) as pool:
|
|
futs = {pool.submit(fetch_group, g, v, force): g for g, v in ids}
|
|
for fut in cf.as_completed(futs):
|
|
g = futs[fut]
|
|
try:
|
|
group_datas[g] = fut.result()
|
|
except Exception as exc: # noqa: BLE001
|
|
failures.append(f"group {g}: {exc}")
|
|
log(f"groups fetched: {len(group_datas)} (failed: {len(failures)})")
|
|
|
|
jobs = variant_jobs(group_datas)
|
|
if limit:
|
|
jobs = jobs[:limit]
|
|
log(f"variants to process: {len(jobs)}")
|
|
docs = []
|
|
with cf.ThreadPoolExecutor(max_workers=WORKERS) as pool:
|
|
futs = {}
|
|
for job in jobs:
|
|
group_data = group_datas[job["group"]]
|
|
futs[pool.submit(process_variant, job, group_data, force)] = job
|
|
done = 0
|
|
for fut in cf.as_completed(futs):
|
|
job = futs[fut]
|
|
try:
|
|
docs.append(fut.result())
|
|
done += 1
|
|
if done % 25 == 0:
|
|
log(f" variants done: {done}/{len(jobs)}")
|
|
except Exception as exc: # noqa: BLE001
|
|
failures.append(f"variant {job['variant']['id']}: {exc}")
|
|
docs.sort(key=lambda d: (d["variant_group_id"], d["variant_id"]))
|
|
write_outputs(docs)
|
|
(RAW / "failures.log").write_text("\n".join(failures) + "\n",
|
|
encoding="utf-8")
|
|
log(f"done: {len(docs)} docs, {len(failures)} failures "
|
|
f"(see .firecrawl/kv-portal/failures.log)")
|
|
|
|
|
|
def _load_docs_from_raw() -> list[dict]:
|
|
"""Rebuild curated outputs from the raw mirror without network."""
|
|
docs = []
|
|
for cache in sorted((RAW / "groups").glob("*.json")):
|
|
group_data = json.loads(cache.read_text(encoding="utf-8"))
|
|
for job in variant_jobs({group_data["id"]: group_data}):
|
|
vid = job["variant"]["id"]
|
|
vdir = RAW / "variants" / vid
|
|
struct_cache = vdir / "structure.json"
|
|
if not struct_cache.exists():
|
|
continue
|
|
structure = json.loads(struct_cache.read_text(encoding="utf-8"))
|
|
slices = {}
|
|
for s in all_slice_ids(structure):
|
|
p = vdir / "slices" / f"{s['slice_id']}.json"
|
|
if p.exists():
|
|
slices[s["slice_id"]] = json.loads(
|
|
p.read_text(encoding="utf-8"))
|
|
topics = {}
|
|
for t in job["variant"].get("topics", []):
|
|
p = vdir / "topics" / f"{t['id']}.json"
|
|
if p.exists():
|
|
topics[t["id"]] = json.loads(
|
|
p.read_text(encoding="utf-8"))
|
|
docs.append(build_variant_doc(
|
|
group_data, job["variant"], structure, slices, topics,
|
|
dt.date.today().isoformat()))
|
|
return docs
|
|
|
|
|
|
def run_refresh() -> None:
|
|
"""Diff sitemap + group metadata against the catalog; refetch changes."""
|
|
catalog_path = OUT / "catalog.json"
|
|
if not catalog_path.exists():
|
|
log("no catalog yet -- run a full fetch first")
|
|
sys.exit(2)
|
|
old = {v["variant_id"]: v for v in
|
|
json.loads(catalog_path.read_text(encoding="utf-8"))["variants"]}
|
|
entries = parse_sitemap(force=True)
|
|
groups: dict[str, str] = {}
|
|
for e in entries:
|
|
groups.setdefault(e["group"], e["variant"])
|
|
log(f"sitemap: {len(entries)} urls, {len(groups)} groups "
|
|
f"(catalog has {len(old)})")
|
|
|
|
group_datas = {}
|
|
with cf.ThreadPoolExecutor(max_workers=WORKERS) as pool:
|
|
futs = {pool.submit(fetch_group, g, v, True): g for g, v in groups.items()}
|
|
for fut in cf.as_completed(futs):
|
|
g = futs[fut]
|
|
try:
|
|
group_datas[g] = fut.result()
|
|
except Exception as exc: # noqa: BLE001
|
|
log(f" WARN group {g}: {exc}")
|
|
|
|
changed, new, removed = [], [], []
|
|
for job in variant_jobs(group_datas):
|
|
vid = job["variant"]["id"]
|
|
v = job["variant"]
|
|
validity = v.get("properties", {}).get("validity", {})
|
|
stamp = ((validity.get("validFrom") or "")[:10],
|
|
(validity.get("lastContentUpdate") or "")[:10])
|
|
if vid not in old:
|
|
new.append(vid)
|
|
elif (old[vid]["valid_from"], old[vid]["last_content_update"]) != stamp:
|
|
changed.append(vid)
|
|
removed = [vid for vid in old if vid not in
|
|
{j["variant"]["id"] for j in variant_jobs(group_datas)}]
|
|
|
|
log(f"changed: {len(changed)}, new: {len(new)}, removed: {len(removed)}")
|
|
docs = []
|
|
for job in variant_jobs(group_datas):
|
|
vid = job["variant"]["id"]
|
|
if vid in changed or vid in new:
|
|
try:
|
|
docs.append(process_variant(job, group_datas[job["group"]],
|
|
force=True))
|
|
except Exception as exc: # noqa: BLE001
|
|
log(f" WARN variant {vid}: {exc}")
|
|
if docs:
|
|
# changed variants are already re-fetched into the raw mirror;
|
|
# rebuild the full curated set from raw so nothing gets lost
|
|
write_outputs(_load_docs_from_raw())
|
|
report = {
|
|
"run_at": dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"),
|
|
"groups_seen": len(group_datas),
|
|
"changed": changed,
|
|
"new": new,
|
|
"removed": removed,
|
|
}
|
|
REPORTS.mkdir(parents=True, exist_ok=True)
|
|
path = REPORTS / f"update-{dt.date.today().isoformat()}.json"
|
|
path.write_text(json.dumps(report, ensure_ascii=False, indent=1),
|
|
encoding="utf-8")
|
|
log(f"update report: {path}")
|
|
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
ap.add_argument("--force", action="store_true",
|
|
help="refetch even when a raw file exists")
|
|
ap.add_argument("--limit", type=int, default=None,
|
|
help="only process the first N groups (smoke test)")
|
|
ap.add_argument("--build-only", action="store_true",
|
|
help="rebuild curated output from the raw mirror")
|
|
ap.add_argument("--refresh", action="store_true",
|
|
help="diff against the catalog and update changed KVs")
|
|
args = ap.parse_args()
|
|
if args.refresh:
|
|
run_refresh()
|
|
else:
|
|
run(force=args.force, limit=args.limit, build_only=args.build_only)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |