#!/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())