#!/usr/bin/env python3 """Generateur de configuration e-CF DGII (Compupar) · Sprint 4 · ERPNext Backend. Roadmap ligne 51 : « e-CF DGII integration (Compupar) ». Produit un plan de configuration de facturation electronique dominicaine cross-coherent avec les contrats CRM deja livres : - le pipeline vente (`workflow_vente_spec.json`) → quel evenement emet ; - le DocType porteur (`dossier_vente/doctype_spec.json`) → sur quel champ ; - le contrat RBAC (`rbac_50_roles.json`) → quel role Compta emet. Ce worker n'ecrit JAMAIS sur le VPS (#8) : il emet les fichiers de hand-off ; la connexion reelle au proveedor Compupar (endpoints, certificat, credentials) et l'emission en production restent cote agent ERPNext Backend. ANTI-INVENTION (#6) : aucun chiffre propre a OTO n'est fabrique. RNC emetteur, taux ITBIS, TipoCambio USD→DOP et endpoints/credentials Compupar restent `null` (a_confirmer) tant que la Compta ne les a pas confirmes AVEC source ; un invariant refuse toute valeur fixee sans `source`. Seules les donnees de reference DGII (codes de type e-CF, formes de paiement, format e-NCF) sont encodees, avec source — ce sont des identifiants normalises du standard, pas des chiffres OTO. Sous-commandes : build [-o OUT] → ecrit ecf_plan.json + MANIFEST.json validate → (re)genere en memoire, valide schema + 12 invariants de cross-coherence e-CF↔workflow↔DocType↔RBAC ; sort en erreur sinon. Sortie deterministe (tri stable, aucun horodatage) → diffable + re-generable. """ 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, "..", "..")) # 05_deliverables_mvp/ _CRM = os.path.join(_DELIVERABLES, "crm") sys.path.insert(0, _HERE) sys.path.insert(0, _CRM) sys.path.insert(0, os.path.join(_DELIVERABLES, "publiciste")) from ecflib import builder, ncf # noqa: E402 from workflow_vente.wflib.rbac import RoleResolver # noqa: E402 from lib import validator as maison # type: ignore # noqa: E402 _SPEC_PATH = os.path.join(_HERE, "ecf_spec.json") _WF_SPEC_PATH = os.path.join(_CRM, "workflow_vente", "workflow_vente_spec.json") _DT_SPEC_PATH = os.path.join(_CRM, "dossier_vente", "doctype_spec.json") _SCHEMA_PATH = os.path.join(_HERE, "ecf.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 _currency_fields(dt_spec: dict) -> set[str]: """Champs Currency du DocType Dossier Vente (bases d'emission legitimes).""" out: set[str] = set() for grp in dt_spec.get("field_groups", []): for f in grp.get("fields", []): if f.get("fieldtype") == "Currency": out.add(f["fieldname"]) return out def _all_fields(dt_spec: dict) -> set[str]: """Tous les fieldnames du DocType Dossier Vente.""" out: set[str] = set() for grp in dt_spec.get("field_groups", []): for f in grp.get("fields", []): out.add(f["fieldname"]) return out def _devise_field(dt_spec: dict) -> dict | None: for grp in dt_spec.get("field_groups", []): for f in grp.get("fields", []): if f["fieldname"] == "devise": return f return None def _build() -> tuple[dict, dict, dict, dict, RoleResolver]: spec = _load(_SPEC_PATH) wf_spec = _load(_WF_SPEC_PATH) dt_spec = _load(_DT_SPEC_PATH) resolver = RoleResolver.from_path() bundle = builder.build_bundle(spec, resolver) return bundle, spec, wf_spec, dt_spec, resolver def _validate(bundle: dict, spec: dict, wf_spec: dict, dt_spec: dict, resolver: RoleResolver) -> list[str]: """Schema de sortie + 12 invariants de cross-coherence (les 4 contrats).""" schema = _load(_SCHEMA_PATH) errors = list(maison.validate(bundle, schema)) plan = bundle["ecf_plan"] m = bundle["manifest"] events = plan["emission_events"] # Contexte derive des contrats voisins. wf_update_values = {s["update_value"] for s in wf_spec["states"]} submitted_values = {s["update_value"] for s in wf_spec["states"] if s["doc_status"] == "1"} currency_fields = _currency_fields(dt_spec) all_fields = _all_fields(dt_spec) tipo_codes = {t["code"] for t in plan["tipos_ecf"]} en_scope = set(plan["tipos_en_scope"]) forma_codes = {f["code"] for f in plan["formas_pago"]} seen: set[tuple] = set() for ev in events: tag = f"{ev['update_value']}/{ev['tipo_ecf']}" # 2 · update_value existe dans le workflow vente (anti-derive). if ev["update_value"] not in wf_update_values: errors.append(f"[{tag}] update_value absent du workflow vente") # 3 · emission uniquement sur un etat SOUMIS (doc_status=1) — jamais sur # un brouillon (lead/visite/devis/abandonne). elif ev["update_value"] not in submitted_values: errors.append(f"[{tag}] update_value n'est pas un etat soumis " f"(doc_status≠1) — pas d'e-CF sur brouillon") # 4 · base_field est un champ Currency reel du DocType Dossier Vente. if ev["base_field"] not in currency_fields: errors.append(f"[{tag}] base_field {ev['base_field']!r} n'est pas un " f"champ Currency du DocType Dossier Vente") # 5 · role resolu + portail compta (emission fiscale = concern Compta). if resolver.portail(ev["role_id"]) != "compta": errors.append(f"[{tag}] role_id hors portail compta " f"({resolver.portail(ev['role_id'])!r})") if ev["erpnext_role_name"] != resolver.erpnext_name(ev["role_id"]): errors.append(f"[{tag}] erpnext_role_name incoherent avec RBAC") # 6 · tipo_ecf : soit null + a_confirmer (jamais fabrique · #6), soit un # code du catalogue ET en_scope. if ev["tipo_ecf"] is None: if not ev["a_confirmer"]: errors.append(f"[{tag}] tipo_ecf null mais a_confirmer=false") else: if ev["tipo_ecf"] not in tipo_codes: errors.append(f"[{tag}] tipo_ecf {ev['tipo_ecf']!r} absent du catalogue DGII") elif ev["tipo_ecf"] not in en_scope: errors.append(f"[{tag}] tipo_ecf {ev['tipo_ecf']!r} hors perimetre (en_scope=false)") # 12a · unicite (update_value, tipo_ecf, role_id). key = (ev["update_value"], ev["tipo_ecf"], ev["role_id"]) if key in seen: errors.append(f"[{tag}] evenement duplique (update_value, tipo_ecf, role_id)") seen.add(key) # 7 · contrat de format e-NCF (structure DGII E+tipo(2)+seq(10) = 13 car.). encf = plan["e_ncf"] if encf["prefix"] != "E": errors.append("e_ncf.prefix doit etre 'E'") if encf["longueur"] != 13: errors.append("e_ncf.longueur doit etre 13") if any(len(c) != 2 or not c.isdigit() for c in tipo_codes): errors.append("tous les tipos_ecf.code doivent etre 2 chiffres") # Composition tracable : un e-NCF echantillon (1er type en_scope + sequence # reelle fournie) est valide et se re-parse sur le meme type. if en_scope: sample_tipo = sorted(en_scope)[0] composed = ncf.compose_encf(sample_tipo, "1") if composed["e_ncf"] is None or not ncf.is_valid_encf(composed["e_ncf"]): errors.append("composition e-NCF echantillon invalide (ncf.compose_encf)") else: parsed = ncf.parse_encf(composed["e_ncf"]) if not parsed or parsed["tipo"] != sample_tipo: errors.append("parse_encf ne retrouve pas le tipo de l'echantillon") # Anti-invention : sans sequence, aucun e-NCF n'est fabrique. if ncf.compose_encf(sample_tipo, None)["e_ncf"] is not None: errors.append("compose_encf fabrique un e-NCF sans sequence (interdit #6)") # 8 · forma_pago_defaut ∈ table DGII + reference Cardnet (#10). fpd = plan["forma_pago_defaut"] if fpd["code"] not in forma_codes: errors.append(f"forma_pago_defaut.code {fpd['code']!r} absent de la table FormaPago DGII") if "cardnet" not in fpd["source"].lower(): errors.append("forma_pago_defaut.source doit referencer Cardnet (#10)") # 9 · moneda : champ devise + options USD/DOP alignees sur le DocType (#10) ; # TipoCambio null OU source (anti-invention FX). mon = plan["moneda"] devf = _devise_field(dt_spec) if mon["tipo_moneda_field"] != "devise": errors.append("moneda.tipo_moneda_field doit etre 'devise'") if mon["options"] != ["USD", "DOP"]: errors.append("moneda.options ≠ USD/DOP (#10)") if devf is None: errors.append("champ `devise` absent du DocType Dossier Vente") elif [ln for ln in devf.get("options", "").split("\n") if ln] != ["USD", "DOP"]: errors.append("options du champ `devise` du DocType ≠ USD/DOP (#10)") if mon["tipo_cambio"] is not None and not mon["source"]: errors.append("moneda.tipo_cambio fixe sans source (interdit #6)") # 10 · ANTI-INVENTION (#6) : RNC emetteur + taux ITBIS jamais sans source. em = plan["emisor"] if (em["rnc_emisor"] is not None or em["razon_social"] is not None) and not em["source"]: errors.append("emisor renseigne sans source (interdit #6)") for tax in plan["taxes"]: if tax["taux_pct"] is not None and not tax["source"]: errors.append(f"tax {tax['code']!r} : taux_pct fixe sans source (interdit #6)") # 11 · field_map : chaque dossier_field est un champ reel du Dossier Vente. for fm in plan["field_map"]: if fm["dossier_field"] not in all_fields: errors.append(f"field_map : {fm['dossier_field']!r} n'est pas un champ du Dossier Vente") # 12b · comptes du manifeste coherents + tipos_en_scope aligne sur le catalogue. if m["counts"]["tipos_ecf"] != len(plan["tipos_ecf"]): errors.append("counts.tipos_ecf incoherent") if m["counts"]["tipos_en_scope"] != len(en_scope): errors.append("counts.tipos_en_scope incoherent") if m["counts"]["formas_pago"] != len(plan["formas_pago"]): errors.append("counts.formas_pago incoherent") if m["counts"]["emission_events"] != len(events): errors.append("counts.emission_events incoherent") if m["counts"]["roles"] != len({e["role_id"] for e in events}): errors.append("counts.roles incoherent") catalogue_en_scope = {t["code"] for t in plan["tipos_ecf"] if t["en_scope"]} if en_scope != catalogue_en_scope: errors.append("tipos_en_scope ≠ codes marques en_scope dans le catalogue") return errors def cmd_build(args: argparse.Namespace) -> int: bundle, spec, wf_spec, dt_spec, resolver = _build() errors = _validate(bundle, spec, wf_spec, dt_spec, resolver) if errors: _eprint("❌ Bundle invalide — generation refusee (anti-regression) :") 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, "ecf_plan.json"), bundle["ecf_plan"]) _write_json(os.path.join(out, "MANIFEST.json"), bundle["manifest"]) m = bundle["manifest"] print(f"✅ Plan e-CF DGII genere dans {out}") print(f" ecf_plan.json : {m['counts']['tipos_ecf']} types e-CF " f"({m['counts']['tipos_en_scope']} en perimetre) · " f"{m['counts']['emission_events']} evenements d'emission · " f"{m['counts']['valeurs_a_confirmer']} valeurs a confirmer") print(" ⚠ RNC / ITBIS / TipoCambio / endpoints Compupar cote ERPNext Backend " "(Compta renseigne avec source · VPS · #8).") return 0 def cmd_validate(args: argparse.Namespace) -> int: bundle, spec, wf_spec, dt_spec, resolver = _build() errors = _validate(bundle, spec, wf_spec, dt_spec, resolver) if errors: _eprint("❌ Validation KO :") for e in errors: _eprint(f" - {e}") return 1 m = bundle["manifest"] print(f"✅ Validation OK — config {m['config_name']!r} (provider {m['provider']!r}) : " f"{m['counts']['emission_events']} evenements, schema + 12 invariants verts.") return 0 def main(argv: list[str] | None = None) -> int: p = argparse.ArgumentParser(description="Generateur de configuration e-CF DGII (Compupar).") sub = p.add_subparsers(dest="cmd", required=True) pb = sub.add_parser("build", help="genere ecf_plan.json / MANIFEST.json") pb.add_argument("-o", "--out", default=_DEFAULT_OUT, help="dossier de sortie (defaut: ./out)") pb.set_defaults(func=cmd_build) pv = sub.add_parser("validate", help="valide le bundle (schema + 12 invariants) sans ecrire") pv.set_defaults(func=cmd_validate) args = p.parse_args(argv) return args.func(args) if __name__ == "__main__": raise SystemExit(main())