oto-enterprise-os-dtp/05_deliverables_mvp/frontend/chat_otoia/chat_otoia_gen.py

#!/usr/bin/env python3
"""Générateur du Chat OTOIA embarqué par portail (Sprint 6 · ERPNext Backend).

Livrable ERPNext Backend de la roadmap Sprint 6 (l.63 « Chat OTOIA embedded dans
chaque portail »). Contrainte #1 « ERPNext natif = priorité absolue » : le bloc de
contenu réutilisable inséré dans un `Workspace` v15 EST le DocType `Custom Block`.
On produit donc un `Custom Block` par portail + sa config runtime — aucun framework
de chat externe côté worker.

Transforme (rbac_50_roles.json + chat_spec.json + portails_spec.json +
portails/out/workspace.json + seo_spec.json) en :
  - custom_block.json -> 1 fixture `Custom Block` (carrier natif) par portail
  - chat_mount.json   -> 1 config runtime par portail (persona, langues, rôles
                         autorisés, portée de connaissance, capabilities, endpoint=null)
  - MANIFEST.json     -> traçabilité (comptes, résumé par portail, marque, hand-off)

Sous-commandes :
  build    [-o OUT]   -> écrit custom_block.json / chat_mount.json / MANIFEST.json
  validate [-o OUT]   -> (re)génère en mémoire, valide vs chat.schema.json + invariants

Sortie déterministe (tri stable, aucun horodatage) -> diffable + re-générable.
Anti-invention (#6) : portails = portails métier ; rôles + portée de connaissance =
surface RBAC EXACTE du portail (mêmes valeurs que les Has Role des Workspaces) ;
endpoint OTOIA = null (a_confirmer · VPS) ; persona/capabilities/langues sourcés.
Ce worker n'écrit JAMAIS sur le VPS (#8) : hand-off à l'agent ERPNext Backend.
"""

from __future__ import annotations

import argparse
import json
import os
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _HERE)

from chatlib import builder, deps  # noqa: E402

_DEFAULT_OUT = os.path.join(_HERE, "out")

_EXPECTED_CAPS = ["aec.py", "knowledge.py", "prompt_engine.py", "chat.py"]


def _eprint(*args) -> None:
    print(*args, file=sys.stderr)


def _write_json(path: str, data) -> None:
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(data, fh, ensure_ascii=False, indent=2)
        fh.write("\n")


def _build_bundle() -> dict:
    contract = deps.load_contract()
    spec = deps.load_spec()
    portails_spec = deps.load_portails_spec()
    langs, default_lang = deps.load_langs()
    return builder.build_bundle(contract, spec, portails_spec, langs, default_lang)


def _workspace_roles(workspaces: list) -> dict[str, list[str]]:
    """Has Role réels par Workspace (label -> rôles triés), depuis le hand-off portails."""
    out: dict[str, list[str]] = {}
    for ws in workspaces:
        out[ws["label"]] = sorted(row["role"] for row in ws.get("roles", []))
    return out


def _validate_bundle(bundle: dict) -> list[str]:
    schema = deps.load_json(deps.SCHEMA_PATH)
    errors = deps.validate(bundle, schema)

    contract = deps.load_contract()
    spec = deps.load_spec()
    portails_spec = deps.load_portails_spec()
    langs, default_lang = deps.load_langs()
    workspaces = deps.load_workspaces()

    custom_blocks = bundle["custom_block"]
    mounts = bundle["chat_mount"]
    manifest = bundle["manifest"]

    portails_metier = set(contract.get("portails_business", []))
    spec_keys = [p["key"] for p in portails_spec["portails"]]
    roles_map = deps.roles_by_portail(contract)
    dt_map = deps.doctypes_by_portail(contract)
    ws_roles = _workspace_roles(workspaces)
    labels = {p["key"]: p["label"] for p in portails_spec["portails"]}

    # Invariant 1 · EXACTEMENT les portails métier (roadmap « chaque portail »),
    # la console technique `plateforme` étant exclue (comme les Workspaces).
    mount_keys = [m["portail"] for m in mounts]
    if set(mount_keys) != portails_metier:
        errors.append(
            f"portails du chat {sorted(set(mount_keys))} != portails_business {sorted(portails_metier)}."
        )
    if mount_keys != spec_keys:
        errors.append(f"ordre/contenu portails {mount_keys} != portails_spec {spec_keys}.")

    # Invariant 2 · 1 Custom Block ⇔ 1 mount ⇔ 1 portail ; block_name unique ; ordre stable.
    if not (len(custom_blocks) == len(mounts) == len(portails_metier)):
        errors.append(
            f"comptes divergents : {len(custom_blocks)} blocks / {len(mounts)} mounts "
            f"/ {len(portails_metier)} portails métier."
        )
    bn = [cb["block_name"] for cb in custom_blocks]
    if len(bn) != len(set(bn)):
        errors.append("block_name en doublon.")

    # Invariant 8 (global) · capabilities = les 4 capacités OTOIA de la spec, sourcées.
    spec_caps = [c["module"] for c in spec["capabilities"]]
    if spec_caps != _EXPECTED_CAPS:
        errors.append(f"capabilities spec {spec_caps} != {_EXPECTED_CAPS} (CLAUDE.md).")
    for c in spec["capabilities"]:
        if not c.get("source"):
            errors.append(f"capability {c.get('module')!r} sans source (anti-invention).")

    # Invariant 9 (global) · langues = seo_spec (source unique), aucune langue inventée.
    if list(langs) != ["fr", "en", "es"] or default_lang != "fr":
        errors.append(f"langues seo inattendues: {langs}/{default_lang}.")

    mount_by_key = {m["portail"]: m for m in mounts}
    cb_by_bn = {cb["block_name"]: cb for cb in custom_blocks}

    for key in spec_keys:
        m = mount_by_key[key]
        label = labels[key]

        # Invariant 3 · ANTI-INVENTION endpoint : jamais fabriqué (null · a_confirmer).
        if m["endpoint"] is not None:
            errors.append(f"[{key}] endpoint doit rester null (a_confirmer · VPS #8).")

        # Invariant 4 · portée de connaissance = surface RBAC EXACTE du portail.
        expected_scope = sorted(dt_map.get(key, set()))
        if m["knowledge_scope"] != expected_scope:
            errors.append(f"[{key}] knowledge_scope != DocTypes autorisés RBAC.")
        if not expected_scope:
            errors.append(f"[{key}] portée de connaissance vide (portail sans DocType ?).")

        # Invariant 5 · rôles autorisés = rôles du portail (contrat), triés.
        expected_roles = sorted(roles_map.get(key, []))
        if m["roles_allowed"] != expected_roles:
            errors.append(f"[{key}] roles_allowed != rôles du portail (contrat).")

        # Invariant 6 · COHÉRENCE INTER-LIVRABLES : rôles du chat == Has Role du
        # Workspace correspondant (frontend/portails/out/workspace.json).
        if label not in ws_roles:
            errors.append(f"[{key}] aucun Workspace '{label}' dans le hand-off portails.")
        elif m["roles_allowed"] != ws_roles[label]:
            errors.append(f"[{key}] roles_allowed != Has Role du Workspace '{label}'.")

        # Invariant 7 · persona sourcée, identique partout.
        if m["persona"] != {"nom": "Amélie", "voix": "multilingual_v2"}:
            errors.append(f"[{key}] persona != Amélie/multilingual_v2 (CLAUDE.md).")

        # Invariant 8 · capabilities du mount = les 4 capacités, sans ajout.
        if m["capabilities"] != _EXPECTED_CAPS:
            errors.append(f"[{key}] capabilities mount != {_EXPECTED_CAPS}.")

        # Invariant 9 · langues + défaut du mount = source seo (aucune invention).
        if m["langues"] != list(langs) or m["lang_defaut"] != default_lang:
            errors.append(f"[{key}] langues/défaut != seo_spec.")

        # Invariant 13 · html_id déterministe.
        if m["html_id"] != f"otoia-chat-{key}":
            errors.append(f"[{key}] html_id non déterministe: {m['html_id']!r}.")

        # Invariant 11 · Custom Block : html de montage bien formé, sans chiffre,
        # sans URL inventée ; block_name = préfixe + label.
        cb_name = m["custom_block"]
        if cb_name != f"OTOIA Chat · {label}":
            errors.append(f"[{key}] block_name non déterministe: {cb_name!r}.")
        if cb_name not in cb_by_bn:
            errors.append(f"[{key}] Custom Block '{cb_name}' absent du bundle.")
        else:
            html = cb_by_bn[cb_name]["html"]
            for token in (m["html_id"], f'data-portail="{key}"', 'data-assistant="Amélie"'):
                if token not in html:
                    errors.append(f"[{key}] html de montage sans {token!r}.")
            if "http://" in html or "https://" in html:
                errors.append(f"[{key}] html contient une URL (endpoint jamais fabriqué #6).")
            if any(ch.isdigit() for ch in html):
                errors.append(f"[{key}] html contient un chiffre (aucun chiffre émis #6).")

    # Invariant 12 · comptes du manifeste cohérents.
    c = manifest["counts"]
    knowledge_union = sorted({dt for m in mounts for dt in m["knowledge_scope"]})
    checks = {
        "portails": len(mounts),
        "custom_blocks": len(custom_blocks),
        "mounts": len(mounts),
        "roles_couverts": sum(len(m["roles_allowed"]) for m in mounts),
        "knowledge_doctypes_uniques": len(knowledge_union),
    }
    for k, v in checks.items():
        if c.get(k) != v:
            errors.append(f"counts.{k} incohérent ({c.get(k)} != {v}).")
    if manifest["endpoint_statut"] != "a_confirmer":
        errors.append("manifest.endpoint_statut != a_confirmer.")

    # Invariant 10 · tokens de marque verbatim (CLAUDE.md #4).
    brand = manifest["brand"]
    if brand["accent"]["valeur"] != "#f0b429" or brand["fond"]["valeur"] != "#0a0a12":
        errors.append("brand fond/accent != CLAUDE.md #4.")

    # Invariant 14 · déterminisme : deux builds successifs identiques.
    again = _build_bundle()
    if json.dumps(again, sort_keys=True, ensure_ascii=False) != json.dumps(bundle, sort_keys=True, ensure_ascii=False):
        errors.append("build non déterministe (deux passes divergent).")

    return errors


def cmd_build(args: argparse.Namespace) -> int:
    bundle = _build_bundle()
    errors = _validate_bundle(bundle)
    if errors:
        _eprint("❌ Bundle invalide — génération refusée (anti-régression) :")
        for e in errors:
            _eprint(f"   - {e}")
        return 1

    out = os.path.abspath(args.out)
    os.makedirs(out, exist_ok=True)
    _write_json(os.path.join(out, "custom_block.json"), bundle["custom_block"])
    _write_json(os.path.join(out, "chat_mount.json"), bundle["chat_mount"])
    _write_json(os.path.join(out, "MANIFEST.json"), bundle["manifest"])

    c = bundle["manifest"]["counts"]
    print(f"✅ Chat OTOIA généré dans {out}")
    print(
        f"   {c['portails']} portails · {c['custom_blocks']} Custom Blocks · "
        f"{c['roles_couverts']} rôles couverts · {c['knowledge_doctypes_uniques']} "
        f"DocTypes de connaissance (endpoint a_confirmer)"
    )
    for m in bundle["manifest"]["portails"]:
        print(
            f"     - {m['workspace']} : {m['nb_roles']} rôles / "
            f"{m['nb_knowledge_doctypes']} DocTypes · block « {m['custom_block']} »"
        )
    return 0


def cmd_validate(args: argparse.Namespace) -> int:
    bundle = _build_bundle()
    errors = _validate_bundle(bundle)
    if errors:
        _eprint("❌ Validation KO :")
        for e in errors:
            _eprint(f"   - {e}")
        return 1
    c = bundle["manifest"]["counts"]
    print(
        f"✅ Validation OK — {c['portails']} portails, {c['roles_couverts']} rôles, "
        f"schéma + 14 invariants verts."
    )
    return 0


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(
        description="Générateur du Chat OTOIA embarqué par portail (Custom Block + config runtime)."
    )
    sub = p.add_subparsers(dest="cmd", required=True)

    pb = sub.add_parser("build", help="génère custom_block.json / chat_mount.json / MANIFEST.json")
    pb.add_argument("-o", "--out", default=_DEFAULT_OUT, help="dossier de sortie (défaut: ./out)")
    pb.set_defaults(func=cmd_build)

    pv = sub.add_parser("validate", help="valide le bundle (schéma + invariants) sans écrire")
    pv.add_argument("-o", "--out", default=_DEFAULT_OUT, help="ignoré (compat)")
    pv.set_defaults(func=cmd_validate)

    args = p.parse_args(argv)
    return args.func(args)


if __name__ == "__main__":
    raise SystemExit(main())