feat(tools): map kv library variants to kb ids

tools/build_kv_variant_map.py

@@ -0,0 +1,143 @@
+"""Erzeugt `tools/catalogs/kv_variant_map.json`: KV-Library-Varianten
+(``SI-xxxx_de`` — identisch mit `library_variant_id` im Odoo-Katalog
+``l10n.at.payroll.kv``) ↔ KB-Einträge ``kv-kvt-NNN``.
+
+Brücke ist der Match-Report der KV-Library (``wko/match-report.json``:
+wko_slug → oegb_variant_id mit Score). Unsere KB-Einträge sind über den
+WKO-Dokument-Slug verknüpft (Katalog-Key ``<slug>.html``, URL-Slug identisch).
+
+Deterministisch re-runnable; der KV-Library-Pfad liegt im Schwesterprojekt:
+
+    python3 tools/build_kv_variant_map.py [--kv-library ../odoo-at-payroll/personalverrechnung/quellen/kv]
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import time
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_KV_LIBRARY = REPO_ROOT.parent / "odoo-at-payroll" / "personalverrechnung" / "quellen" / "kv"
+
+
+def build(kv_library: Path) -> dict:
+    kb_catalog = json.loads(
+        (REPO_ROOT / "tools" / "catalogs" / "kv_catalog.json").read_text(encoding="utf-8")
+    )
+    lib_catalog = json.loads((kv_library / "kv-catalog.json").read_text(encoding="utf-8"))
+    match = json.loads((kv_library / "wko" / "match-report.json").read_text(encoding="utf-8"))
+
+    variant_meta = {e["variant_id"]: e for e in lib_catalog["entries"]}
+
+    # slug -> KB-Eintrag (Katalog-Key ist "<slug>.html"; URL-Slug als Kontrolle)
+    slug_to_kb: dict[str, dict] = {}
+    for filename, entry in kb_catalog["entries"].items():
+        slug = entry["url"].rstrip("/").rsplit("/", 1)[-1]
+        if not slug:
+            slug = filename.removesuffix(".html")
+        slug_to_kb[slug] = entry
+
+    variants: dict[str, dict] = {}
+    unmatched: dict[str, dict] = {}
+
+    def bucket(variant_id: str, confidence: str) -> dict:
+        if variant_id not in variants:
+            meta = variant_meta.get(variant_id, {})
+            variants[variant_id] = {
+                "variant_id": variant_id,
+                "variant_group_id": meta.get("variant_group_id"),
+                "title": meta.get("title"),
+                "chamber": meta.get("chamber"),
+                "docs": [],
+                "confidence": confidence,
+            }
+        return variants[variant_id]
+
+    for row in match["matches"]:
+        slug = row["wko_slug"]
+        kb = slug_to_kb.get(slug)
+        if kb is None:
+            continue
+        b = bucket(row["oegb_variant_id"], "matched")
+        b["docs"].append(
+            {"kv_kvt_id": kb["id"], "slug": slug, "doctype": kb.get("doctype")}
+        )
+    for row in match.get("low_confidence", []):
+        slug = row["wko_slug"]
+        kb = slug_to_kb.get(slug)
+        if kb is None:
+            continue
+        #matched schlägt low_confidence für denselben Slug
+        already = any(
+            kb["id"] in {d["kv_kvt_id"] for d in v["docs"]}
+            for v in variants.values()
+        )
+        if already:
+            continue
+        b = bucket(row["best_oegb_variant_id"], "low")
+        b["docs"].append(
+            {"kv_kvt_id": kb["id"], "slug": slug, "doctype": kb.get("doctype")}
+        )
+
+    covered_ids = {
+        d["kv_kvt_id"] for v in variants.values() for d in v["docs"]
+    }
+    for filename, entry in kb_catalog["entries"].items():
+        if entry["id"] not in covered_ids:
+            slug = entry["url"].rstrip("/").rsplit("/", 1)[-1] or filename.removesuffix(".html")
+            unmatched[entry["id"]] = {"slug": slug, "title": entry["title"]}
+
+    by_id = {v["id"]: v for v in kb_catalog["entries"].values()}
+    for v in variants.values():
+        v["docs"] = sorted(
+            v["docs"], key=lambda d: by_id[d["kv_kvt_id"]]["id"]
+        )
+
+    return {
+        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        "sources": {
+            "kb_catalog": "tools/catalogs/kv_catalog.json",
+            "kv_library": str(kv_library),
+        },
+        "stats": {
+            "n_variants": len(variants),
+            "n_variants_matched": sum(
+                1 for v in variants.values() if v["confidence"] == "matched"
+            ),
+            "n_variants_low_confidence": sum(
+                1 for v in variants.values() if v["confidence"] == "low"
+            ),
+            "n_kb_entries_covered": len(covered_ids),
+            "n_kb_entries_total": len(kb_catalog["entries"]),
+            "n_kb_entries_unmatched": len(unmatched),
+            "kv_library_report": {
+                "n_matched": match["n_matched"],
+                "n_low_confidence": match["n_low_confidence"],
+                "n_unmatched_wko_current": match["n_unmatched_wko_current"],
+            },
+        },
+        "variants": dict(sorted(variants.items())),
+        "unmatched_kb_entries": dict(sorted(unmatched.items())),
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--kv-library", type=Path, default=DEFAULT_KV_LIBRARY,
+        help="KV-Library-Verzeichnis (kv-catalog.json + wko/match-report.json)",
+    )
+    args = parser.parse_args()
+    out = build(args.kv_library)
+    target = REPO_ROOT / "tools" / "catalogs" / "kv_variant_map.json"
+    target.write_text(
+        json.dumps(out, ensure_ascii=False, indent=1) + "\n", encoding="utf-8"
+    )
+    print(json.dumps(out["stats"], ensure_ascii=False, indent=1))
+    print("geschrieben:", target)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())