oto-enterprise-os-dtp/05_deliverables_mvp/crm/financement_bancaire/financement_bancaire_gen.py

#!/usr/bin/env python3
"""Générateur du module Financement Bancaire — parcours hypothécaire RD.

Sprint 4 · CRM natif · roadmap L52 (« workflow vente end-to-end » — volet
financement) + DIRECTIVE_FINANCEMENT_BANCAIRE_COMPLET_20260803.md (Michel).

Transforme le contrat `financement_spec.json` en artefacts prêts à consommer
par le frontend `renderHypotheque` et le backend `/api/hypotheque/*` :
  - banques.json             → profils des banques partenaires (taux/LTV/durée)
  - sections.json            → structure accordéon (apport → info → … → envoi)
  - documents.json           → catalogue des documents exigés par catégorie
  - autorisations.json       → autorisations à signer (OTO Sign)
  - gate_spec.json           → gate check 4 conditions + bannière critique
  - gate_status_example.json → exemple CALCULÉ (dossier bloqué) auditable
  - MANIFEST.json            → traçabilité + comptes dérivés

Le CŒUR métier (calcul d'apport 20/30 + gate check 4 conditions) vit dans
`finlib/gate.py` (fonctions pures, testées). Ce fichier ne fait qu'orchestrer :
charger le contrat, assembler le bundle, valider (schéma + invariants), écrire.

Anti-invention (#6) : aucune valeur (taux d'apport, documents, autorisations,
noms de rôle) n'est codée en dur — tout vient du contrat + du contrat RBAC
(`rbac_50_roles.json`, résolu par `RoleResolver.from_path`, invariant 4). Ce
worker n'écrit JAMAIS sur le VPS (l'application réelle est côté serveur).

Sortie déterministe (tri stable, aucun horodatage) → diffable + re-générable.

Sous-commandes :
  build    [-o OUT]  → écrit les 7 fichiers d'artefacts
  validate           → (re)génère en mémoire, valide schéma + 12 invariants ;
                       sort en erreur si un invariant casse (anti-régression).
"""

from __future__ import annotations

import argparse
import json
import os
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
_DELIVERABLES = os.path.normpath(os.path.join(_HERE, "..", ".."))

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

from finlib import builder, gate  # noqa: E402
from finlib.rbac import RoleResolver  # noqa: E402
from lib import validator as maison  # type: ignore  # noqa: E402

_SPEC_PATH = os.path.join(_HERE, "financement_spec.json")
_SCHEMA_PATH = os.path.join(_HERE, "financement.schema.json")
_DEFAULT_OUT = os.path.join(_HERE, "out")

# Fichiers écrits par `build` (nom → clé du bundle).
_ARTIFACTS = [
    ("banques.json", "banques"),
    ("sections.json", "sections"),
    ("documents.json", "documents"),
    ("autorisations.json", "autorisations"),
    ("gate_spec.json", "gate_spec"),
    ("gate_status_example.json", "gate_status_example"),
    ("MANIFEST.json", "manifest"),
]


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() -> tuple[dict, dict]:
    spec = _load(_SPEC_PATH)
    resolver = RoleResolver.from_path()
    return spec, builder.build_bundle(spec, resolver)


def _dups(ids: list[str]) -> list[str]:
    seen: set[str] = set()
    dups: set[str] = set()
    for i in ids:
        if i in seen:
            dups.add(i)
        seen.add(i)
    return sorted(dups)


def _validate_bundle(spec: dict, bundle: dict) -> list[str]:
    """Schéma de sortie + 12 invariants métier du module Financement Bancaire."""
    schema = _load(_SCHEMA_PATH)
    errors = list(maison.validate(bundle, schema))

    banques = bundle["banques"]
    sections = bundle["sections"]
    documents = bundle["documents"]
    autorisations = bundle["autorisations"]
    gate_spec = bundle["gate_spec"]
    manifest = bundle["manifest"]
    cfg = gate.GateConfig.from_spec(spec)

    # Invariant 1 · unicité des identifiants (banques, documents, autorisations, sections).
    for coll_name, coll in (
        ("banques", banques),
        ("documents", documents),
        ("autorisations", autorisations),
        ("sections", sections),
    ):
        d = _dups([x["id"] for x in coll])
        if d:
            errors.append(f"Identifiant dupliqué dans {coll_name} : {d}")

    # Invariant 2 · taux d'apport : 0 < résident < étranger ≤ 100 (amendement Michel).
    taux = cfg.taux_apport_pct
    tr = taux.get("resident_rd")
    te = taux.get("etranger")
    if tr is None or te is None:
        errors.append("Taux d'apport résident/étranger manquant dans le contrat.")
    elif not (0 < tr < te <= 100):
        errors.append(
            f"Taux d'apport invalides : résident={tr}, étranger={te} "
            f"(attendu 0 < résident < étranger ≤ 100)."
        )

    # Invariant 3 · fourchettes banque cohérentes (min ≤ max quand les deux présents).
    for b in banques:
        for lo, hi, nom in (
            ("taux_min_pct", "taux_max_pct", "taux"),
            ("ltv_min_pct", "ltv_max_pct", "LTV"),
        ):
            a, z = b.get(lo), b.get(hi)
            if a is not None and z is not None and a > z:
                errors.append(f"Banque {b['id']!r} : {nom} min {a} > max {z}.")

    # Invariant 4 · rôles du contrat résolus dans RBAC (anti-invention #6).
    resolver = RoleResolver.from_path()
    for cle, rid in spec["roles"].items():
        try:
            resolver.erpnext_name(rid)
        except (KeyError, ValueError) as exc:
            errors.append(f"Rôle {cle}={rid!r} irrésoluble dans RBAC : {exc}")

    # Invariant 5 · sections : `ordre` contigu 1..N, unique.
    ordres = [s["ordre"] for s in sections]
    if sorted(ordres) != list(range(1, len(sections) + 1)):
        errors.append(f"Ordres de sections non contigus 1..N : {sorted(ordres)}")

    # Invariant 6 · exactement UNE section `gating`, en position 1 (apport_initial).
    gating = [s for s in sections if s.get("gating")]
    if len(gating) != 1:
        errors.append(f"Attendu exactement 1 section gating, trouvé {len(gating)}.")
    elif gating[0]["ordre"] != 1 or gating[0]["id"] != "apport_initial":
        errors.append(
            "La section gating doit être `apport_initial` en position 1 "
            f"(trouvé id={gating[0]['id']!r}, ordre={gating[0]['ordre']})."
        )

    # Invariant 7 · gate : exactement 4 conditions, clés == CONDITION_KEYS, ordre 1..4.
    conds = gate_spec["conditions"]
    keys = [c["key"] for c in conds]
    if set(keys) != set(gate.CONDITION_KEYS):
        errors.append(
            f"Clés de condition {sorted(set(keys))} ≠ clés canoniques "
            f"{sorted(gate.CONDITION_KEYS)}."
        )
    if [c["ordre"] for c in conds] != [1, 2, 3, 4]:
        errors.append(f"Ordres des conditions du gate ≠ 1..4 : {[c['ordre'] for c in conds]}.")
    if gate_spec["condition_keys"] != list(gate.CONDITION_KEYS):
        errors.append("gate_spec.condition_keys ≠ ordre canonique CONDITION_KEYS.")

    # Invariant 8 · bannière : 4 lignes, alignées 1-pour-1 sur les 4 conditions.
    lignes = gate_spec["banniere"]["lignes"]
    if len(lignes) != len(gate.CONDITION_KEYS):
        errors.append(
            f"Bannière : {len(lignes)} lignes ≠ {len(gate.CONDITION_KEYS)} conditions."
        )

    # Invariant 9 · documents « étrangers » exclusifs + cohérence categorie⟺applies_to.
    req_res = set(gate.required_document_ids("resident_rd", cfg))
    req_etr = set(gate.required_document_ids("etranger", cfg))
    if not req_res < req_etr:
        errors.append(
            "Les documents requis résident doivent être un sous-ensemble STRICT "
            "des documents requis étranger (les pièces étrangers ajoutent des exigences)."
        )
    for d in documents:
        etr_cat = d["categorie"] == "etrangers"
        etr_app = d["applies_to"] == "etranger"
        if etr_cat != etr_app:
            errors.append(
                f"Document {d['id']!r} : categorie/applies_to incohérents "
                f"(categorie={d['categorie']}, applies_to={d['applies_to']})."
            )

    # Invariant 10 · statuts recevables ⊆ statuts déclarés.
    statuses = set(spec["document_statuses"])
    recevables = set(spec["document_statuses_recevables"])
    if not recevables <= statuses:
        errors.append(
            f"document_statuses_recevables {sorted(recevables)} ⊄ "
            f"document_statuses {sorted(statuses)}."
        )

    # Invariant 11 · exemple de gate CALCULÉ cohérent (auto-audit du cœur métier).
    ex = bundle["gate_status_example"]
    gs = ex["gate_status"]
    recomputed_can = gate.gate_status(spec["dossier_exemple"], cfg)["can_submit"]
    if ex["can_submit"] != recomputed_can or gs["can_submit"] != recomputed_can:
        errors.append("gate_status_example.can_submit incohérent avec le recalcul.")
    if bool(ex["reasons"]) == ex["can_submit"]:
        errors.append("Incohérence : `reasons` non vide DOIT équivaloir à ¬can_submit.")
    completed = sum(1 for k in gate.CONDITION_KEYS if gs["conditions"][k]["ok"])
    if gs["conditions_completed"] != completed:
        errors.append("gate_status_example.conditions_completed mal compté.")

    # Invariant 12 · cohérence des comptes du manifeste avec les collections.
    c = manifest["counts"]
    expected = {
        "banques": len(banques),
        "sections": len(sections),
        "documents": len(documents),
        "documents_requis_resident": len(req_res),
        "documents_requis_etranger": len(req_etr),
        "autorisations": len(autorisations),
        "gate_conditions": len(conds),
        "banniere_lignes": len(lignes),
        "workflow_suivi_etapes": len(gate_spec["workflow_suivi_banque"]),
    }
    for k, v in expected.items():
        if c.get(k) != v:
            errors.append(f"manifest.counts.{k}={c.get(k)} ≠ {v} (dérive de compte).")

    return errors


def cmd_build(args: argparse.Namespace) -> int:
    spec, bundle = _build_bundle()
    errors = _validate_bundle(spec, 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)
    for fname, key in _ARTIFACTS:
        _write_json(os.path.join(out, fname), bundle[key])

    m = bundle["manifest"]
    c = m["counts"]
    print(f"✅ Artefacts Financement Bancaire générés dans {out}")
    print(f"   banques.json       : {c['banques']} banques partenaires")
    print(f"   sections.json      : {c['sections']} sections (accordéon)")
    print(
        f"   documents.json     : {c['documents']} documents "
        f"(requis résident {c['documents_requis_resident']} · étranger {c['documents_requis_etranger']})"
    )
    print(f"   autorisations.json : {c['autorisations']} autorisations (OTO Sign)")
    print(f"   gate_spec.json     : {c['gate_conditions']} conditions + bannière critique")
    ex = bundle["gate_status_example"]
    print(
        f"   gate_status_example: dossier {ex['dossier_id']} · "
        f"can_submit={ex['can_submit']} · {len(ex['reasons'])} raison(s) de blocage"
    )
    return 0


def cmd_validate(args: argparse.Namespace) -> int:
    spec, bundle = _build_bundle()
    errors = _validate_bundle(spec, bundle)
    if errors:
        _eprint("❌ Validation KO :")
        for e in errors:
            _eprint(f"   - {e}")
        return 1
    c = bundle["manifest"]["counts"]
    print(
        f"✅ Validation OK — {c['banques']} banques, {c['sections']} sections, "
        f"{c['documents']} documents, {c['autorisations']} autorisations, "
        f"gate 4 conditions, schéma + 12 invariants verts."
    )
    return 0


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(
        description="Générateur du module Financement Bancaire (parcours hypothécaire RD)."
    )
    sub = p.add_subparsers(dest="cmd", required=True)

    pb = sub.add_parser("build", help="génère les 7 artefacts out/")
    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 + 12 invariants) sans écrire")
    pv.set_defaults(func=cmd_validate)

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


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