oto-enterprise-os-dtp/05_deliverables_mvp/rbac/roleprofile_gen/roleprofile_gen.py

#!/usr/bin/env python3
"""Générateur de `Role Profile` par portail · RBAC 50 rôles (Sprint 2).

Troisième et dernier maillon RBAC en-repo, complément de `../fixtures_gen`
(Role + Custom DocPerm) et `../userperm_gen` (row-level) : il produit les
**bundles assignables** — un `Role Profile` ERPNext v15 natif par portail — qui
permettent à l'agent ERPNext d'attribuer à un utilisateur, en un seul geste,
tous les rôles de SON portail (champ `User.role_profile_name`).

Cible de portage VPS : agent ERPNext Backend (SPEC §7). Ce worker n'écrit JAMAIS
sur le VPS — il produit en-repo un bundle de fixtures que l'agent dépose dans
`fixtures/` de l'app OTO puis applique via `bench migrate`.

Transforme `rbac_50_roles.json` (contrat validé par `rbac.schema.json`) en :
  - role_profile.json → 1 fixture `Role Profile` par portail (+ ses `Has Role`)
  - MANIFEST.json      → traçabilité : comptes, portails métier vs technique,
                         mapping portail → profil

Sous-commandes :
  build    [-o OUT]   → écrit role_profile.json / MANIFEST.json
  validate [-o OUT]   → (re)génère en mémoire, valide vs roleprofile.schema.json
                        + invariants — sort en erreur si un invariant casse.

Sortie déterministe (tri stable, aucun horodatage) → diffable + re-générable.
Anti-invention (#6) : 100 % des rôles proviennent du contrat ; le nom du profil
est une convention déterministe dérivée de la clé `portail` (aucun libellé
métier fabriqué) ; aucun DocType custom (que du natif v15).
"""

from __future__ import annotations

import argparse
import json
import os
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
_RBAC_DIR = os.path.normpath(os.path.join(_HERE, ".."))          # rbac/
_DELIVERABLES = os.path.normpath(os.path.join(_RBAC_DIR, ".."))  # 05_deliverables_mvp/

sys.path.insert(0, _HERE)
# Réutilisation (workflow #5 : zéro duplication) du validateur maison Publiciste.
sys.path.insert(0, os.path.join(_DELIVERABLES, "publiciste"))

from profilelib import builder, frappe  # noqa: E402
from lib import validator as maison  # type: ignore  # noqa: E402

_CONTRACT_PATH = os.path.join(_RBAC_DIR, "rbac_50_roles.json")
_SCHEMA_PATH = os.path.join(_HERE, "roleprofile.schema.json")
_DEFAULT_OUT = os.path.join(_HERE, "out")


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


def _load(path: str) -> dict:
    with open(path, encoding="utf-8") as fh:
        return json.load(fh)


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:
    return builder.build_bundle(_load(_CONTRACT_PATH))


def _validate_bundle(bundle: dict) -> list[str]:
    """Valide le bundle contre roleprofile.schema.json + invariants RBAC."""
    schema = _load(_SCHEMA_PATH)
    errors = list(maison.validate(bundle, schema))

    manifest = bundle["manifest"]
    profiles = bundle["role_profile"]
    contract = _load(_CONTRACT_PATH)
    contract_roles = {r["erpnext_role_name"] for r in contract["roles"]}
    portail_of_role = {r["erpnext_role_name"]: r["portail"] for r in contract["roles"]}
    portails_metier = set(contract.get("portails_business", []))

    # Invariant 1 · cible non-négociable « RBAC 50 rôles » : couverture exhaustive.
    if manifest["cible_rbac_roles"] != len(contract_roles):
        errors.append(
            f"cible_rbac_roles={manifest['cible_rbac_roles']} ≠ rôles contrat={len(contract_roles)}"
        )

    # Invariant 2 · couverture BIJECTIVE : chaque rôle du contrat dans EXACTEMENT
    # un profil, aucun rôle inventé, aucun manquant.
    assigned: list[str] = []
    for prof in profiles:
        for row in prof["roles"]:
            assigned.append(row["role"])
    if len(assigned) != len(set(assigned)):
        errors.append("Un rôle apparaît dans plusieurs Role Profile (couverture non bijective).")
    if set(assigned) != contract_roles:
        missing = sorted(contract_roles - set(assigned))
        extra = sorted(set(assigned) - contract_roles)
        errors.append(f"Couverture ≠ contrat (manquants={missing}, en trop={extra}).")

    # Invariant 3 · un profil par portail distinct du contrat, nom déterministe.
    contract_portails = {r["portail"] for r in contract["roles"]}
    seen_names: set[str] = set()
    for prof in profiles:
        name = prof["name"]
        if name != prof["role_profile"]:
            errors.append(f"{name}: name ≠ role_profile.")
        if name in seen_names:
            errors.append(f"Role Profile dupliqué : {name}.")
        seen_names.add(name)
        # Chaque rôle du profil appartient bien au même portail que le profil.
        expected = {n: p for n, p in portail_of_role.items()}
        portails_in_prof = {expected[row["role"]] for row in prof["roles"]}
        if len(portails_in_prof) != 1:
            errors.append(f"{name}: mélange de portails {sorted(portails_in_prof)}.")
        else:
            portail = portails_in_prof.pop()
            if name != frappe.role_profile_name(portail):
                errors.append(f"{name}: nom ≠ convention pour portail {portail!r}.")

    # Invariant 4 · le manifeste liste EXACTEMENT les portails du contrat.
    meta_portails = {m["portail"] for m in manifest["profiles"]}
    if meta_portails != contract_portails:
        errors.append(f"manifest.profiles portails {sorted(meta_portails)} ≠ contrat {sorted(contract_portails)}.")

    # Invariant 5 · métadonnées fidèles (flag métier + comptes par profil).
    nb_by_portail: dict[str, int] = {}
    for prof in profiles:
        p = portail_of_role[prof["roles"][0]["role"]]
        nb_by_portail[p] = len(prof["roles"])
    for m in manifest["profiles"]:
        p = m["portail"]
        if m["metier"] != (p in portails_metier):
            errors.append(f"profil {p!r}: flag metier incohérent avec portails_business.")
        if m["role_profile"] != frappe.role_profile_name(p):
            errors.append(f"profil {p!r}: role_profile ≠ convention de nommage.")
        if m["nb_roles"] != nb_by_portail.get(p):
            errors.append(f"profil {p!r}: nb_roles {m['nb_roles']} ≠ réel {nb_by_portail.get(p)}.")

    # Invariant 6 · cohérence des comptages agrégés du manifeste.
    counts = manifest["counts"]
    if counts["role_profiles"] != len(profiles):
        errors.append("counts.role_profiles incohérent.")
    if counts["roles_couverts"] != len(assigned):
        errors.append("counts.roles_couverts incohérent avec les profils.")
    nb_metier = sum(1 for m in manifest["profiles"] if m["metier"])
    if counts["portails_metier"] != nb_metier:
        errors.append("counts.portails_metier incohérent.")
    if counts["portails_techniques"] != len(profiles) - nb_metier:
        errors.append("counts.portails_techniques incohérent.")

    # Invariant 7 · déterminisme : profils triés par portail, rôles triés par nom.
    if [m["portail"] for m in manifest["profiles"]] != sorted(meta_portails):
        errors.append("manifest.profiles non trié par portail (déterminisme).")
    for prof in profiles:
        role_seq = [row["role"] for row in prof["roles"]]
        if role_seq != sorted(role_seq):
            errors.append(f"{prof['name']}: Has Role non triés (déterminisme).")

    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, "role_profile.json"), bundle["role_profile"])
    _write_json(os.path.join(out, "MANIFEST.json"), bundle["manifest"])

    m = bundle["manifest"]
    c = m["counts"]
    print(f"✅ Role Profile générés dans {out}")
    print(
        f"   role_profile.json : {c['role_profiles']} profils "
        f"({c['portails_metier']} métier + {c['portails_techniques']} technique) "
        f"· {c['roles_couverts']}/50 rôles couverts"
    )
    for meta in m["profiles"]:
        tag = "métier" if meta["metier"] else "technique"
        print(f"     - {meta['role_profile']} : {meta['nb_roles']} rôles ({tag})")
    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['role_profiles']} Role Profile, "
        f"{c['roles_couverts']}/50 rôles couverts (bijectif), "
        f"schéma + invariants verts."
    )
    return 0


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(
        description="Générateur de Role Profile (bundles de rôles) depuis le contrat RBAC."
    )
    sub = p.add_subparsers(dest="cmd", required=True)

    pb = sub.add_parser("build", help="génère role_profile.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())