pv-agent/tools/build_kv_variant_map.py

"""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())