#!/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/.json|.md personalverrechnung/quellen/kv/oegb/update-reports/update-.json Raw API mirror (unversioned, .firecrawl/ is gitignored): .firecrawl/kv-portal/sitemap.xml .firecrawl/kv-portal/groups/.json|html .firecrawl/kv-portal/variants//structure.json .firecrawl/kv-portal/variants//slices/.json .firecrawl/kv-portal/variants//topics/.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= returns the doc-set tree; its children are KV_TIME_SLICE parts whose ids embed the version date (_). The latest date per slice name is the current version of that part; older entries are history. - kvfulltext.kv-slices.json?slice-id= returns the paragraph tree (titleHtml/contentHtml) of one slice. - kvdashboard.kv-topics.json?topic-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(" 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"([^<]+)", 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"]*>", 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()