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