refactor(universign): simplify status mapping and remove redundant comments
This commit is contained in:
Fanilo-Nantenaina
parent
f3fc32c89f
commit
bac8cc6017
1 changed files with 30 additions and 151 deletions
|
|
@ -1,76 +1,46 @@
|
||||||
"""
|
from typing import Dict
|
||||||
Mapping exhaustif : Universign ↔ Local ↔ Sage
|
|
||||||
"""
|
|
||||||
|
|
||||||
from typing import Dict, Optional
|
|
||||||
from enum import Enum
|
|
||||||
|
|
||||||
|
|
||||||
# ========================================
|
|
||||||
# MAPPING UNIVERSIGN → LOCAL
|
|
||||||
# ========================================
|
|
||||||
|
|
||||||
UNIVERSIGN_TO_LOCAL: Dict[str, str] = {
|
UNIVERSIGN_TO_LOCAL: Dict[str, str] = {
|
||||||
# États initiaux
|
"draft": "EN_ATTENTE",
|
||||||
"draft": "EN_ATTENTE", # Transaction créée
|
"ready": "EN_ATTENTE",
|
||||||
"ready": "EN_ATTENTE", # Prête mais pas envoyée
|
"started": "EN_COURS",
|
||||||
# En cours
|
"completed": "SIGNE",
|
||||||
"started": "EN_COURS", # Envoyée, en attente de signature
|
"refused": "REFUSE",
|
||||||
# États finaux (succès)
|
"expired": "EXPIRE",
|
||||||
"completed": "SIGNE", # ✓ Signé avec succès
|
"canceled": "REFUSE",
|
||||||
# États finaux (échec)
|
"failed": "ERREUR",
|
||||||
"refused": "REFUSE", # ✗ Refusé par un signataire
|
|
||||||
"expired": "EXPIRE", # ⏰ Délai expiré
|
|
||||||
"canceled": "REFUSE", # ✗ Annulé manuellement
|
|
||||||
"failed": "ERREUR", # ⚠️ Erreur technique
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
# ========================================
|
|
||||||
# MAPPING LOCAL → SAGE (CHAMP LIBRE)
|
|
||||||
# ========================================
|
|
||||||
|
|
||||||
LOCAL_TO_SAGE_STATUS: Dict[str, int] = {
|
LOCAL_TO_SAGE_STATUS: Dict[str, int] = {
|
||||||
"""
|
"EN_ATTENTE": 0,
|
||||||
À stocker dans un champ libre Sage (ex: CB_STATUT_SIGNATURE)
|
"EN_COURS": 1,
|
||||||
"""
|
"SIGNE": 2,
|
||||||
"EN_ATTENTE": 0, # Non envoyé
|
"REFUSE": 3,
|
||||||
"EN_COURS": 1, # Envoyé, en attente
|
"EXPIRE": 4,
|
||||||
"SIGNE": 2, # ✓ Signé (peut déclencher workflow)
|
"ERREUR": 5,
|
||||||
"REFUSE": 3, # ✗ Refusé
|
|
||||||
"EXPIRE": 4, # ⏰ Expiré
|
|
||||||
"ERREUR": 5, # ⚠️ Erreur
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
# ========================================
|
|
||||||
# ACTIONS MÉTIER PAR STATUT
|
|
||||||
# ========================================
|
|
||||||
|
|
||||||
STATUS_ACTIONS: Dict[str, Dict[str, any]] = {
|
STATUS_ACTIONS: Dict[str, Dict[str, any]] = {
|
||||||
"""
|
|
||||||
Actions automatiques à déclencher selon le statut
|
|
||||||
"""
|
|
||||||
"SIGNE": {
|
"SIGNE": {
|
||||||
"update_sage_status": True, # Mettre à jour Sage
|
"update_sage_status": True,
|
||||||
"trigger_workflow": True, # Déclencher transformation (devis→commande)
|
"trigger_workflow": True,
|
||||||
"send_notification": True, # Email de confirmation
|
"send_notification": True,
|
||||||
"archive_document": True, # Archiver le PDF signé
|
"archive_document": True,
|
||||||
"update_sage_field": "CB_DateSignature", # Champ libre Sage
|
"update_sage_field": "CB_DateSignature",
|
||||||
},
|
},
|
||||||
"REFUSE": {
|
"REFUSE": {
|
||||||
"update_sage_status": True,
|
"update_sage_status": True,
|
||||||
"trigger_workflow": False,
|
"trigger_workflow": False,
|
||||||
"send_notification": True,
|
"send_notification": True,
|
||||||
"archive_document": False,
|
"archive_document": False,
|
||||||
"alert_sales": True, # Alerter commercial
|
"alert_sales": True,
|
||||||
},
|
},
|
||||||
"EXPIRE": {
|
"EXPIRE": {
|
||||||
"update_sage_status": True,
|
"update_sage_status": True,
|
||||||
"trigger_workflow": False,
|
"trigger_workflow": False,
|
||||||
"send_notification": True,
|
"send_notification": True,
|
||||||
"archive_document": False,
|
"archive_document": False,
|
||||||
"schedule_reminder": True, # Programmer relance
|
"schedule_reminder": True,
|
||||||
},
|
},
|
||||||
"ERREUR": {
|
"ERREUR": {
|
||||||
"update_sage_status": False,
|
"update_sage_status": False,
|
||||||
|
|
@ -81,141 +51,60 @@ STATUS_ACTIONS: Dict[str, Dict[str, any]] = {
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
# ========================================
|
|
||||||
# RÈGLES DE TRANSITION
|
|
||||||
# ========================================
|
|
||||||
|
|
||||||
ALLOWED_TRANSITIONS: Dict[str, list] = {
|
ALLOWED_TRANSITIONS: Dict[str, list] = {
|
||||||
"""
|
|
||||||
Transitions de statuts autorisées (validation)
|
|
||||||
"""
|
|
||||||
"EN_ATTENTE": ["EN_COURS", "ERREUR"],
|
"EN_ATTENTE": ["EN_COURS", "ERREUR"],
|
||||||
"EN_COURS": ["SIGNE", "REFUSE", "EXPIRE", "ERREUR"],
|
"EN_COURS": ["SIGNE", "REFUSE", "EXPIRE", "ERREUR"],
|
||||||
"SIGNE": [], # État final, pas de retour
|
"SIGNE": [],
|
||||||
"REFUSE": [], # État final
|
"REFUSE": [],
|
||||||
"EXPIRE": [], # État final
|
"EXPIRE": [],
|
||||||
"ERREUR": ["EN_ATTENTE", "EN_COURS"], # Retry possible
|
"ERREUR": ["EN_ATTENTE", "EN_COURS"],
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
# ========================================
|
|
||||||
# FONCTION DE MAPPING
|
|
||||||
# ========================================
|
|
||||||
|
|
||||||
|
|
||||||
def map_universign_to_local(universign_status: str) -> str:
|
def map_universign_to_local(universign_status: str) -> str:
|
||||||
"""
|
|
||||||
Convertit un statut Universign en statut local
|
|
||||||
|
|
||||||
Args:
|
|
||||||
universign_status: Statut brut Universign
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Statut local normalisé
|
|
||||||
"""
|
|
||||||
return UNIVERSIGN_TO_LOCAL.get(
|
return UNIVERSIGN_TO_LOCAL.get(
|
||||||
universign_status.lower(),
|
universign_status.lower(),
|
||||||
"ERREUR", # Fallback si statut inconnu
|
"ERREUR",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def get_sage_status_code(local_status: str) -> int:
|
def get_sage_status_code(local_status: str) -> int:
|
||||||
"""
|
|
||||||
Obtient le code numérique pour Sage
|
|
||||||
|
|
||||||
Args:
|
|
||||||
local_status: Statut local (EN_ATTENTE, SIGNE, etc.)
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Code numérique pour champ libre Sage
|
|
||||||
"""
|
|
||||||
return LOCAL_TO_SAGE_STATUS.get(local_status, 5)
|
return LOCAL_TO_SAGE_STATUS.get(local_status, 5)
|
||||||
|
|
||||||
|
|
||||||
def is_transition_allowed(from_status: str, to_status: str) -> bool:
|
def is_transition_allowed(from_status: str, to_status: str) -> bool:
|
||||||
"""
|
|
||||||
Vérifie si une transition de statut est valide
|
|
||||||
|
|
||||||
Args:
|
|
||||||
from_status: Statut actuel
|
|
||||||
to_status: Nouveau statut
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
True si la transition est autorisée
|
|
||||||
"""
|
|
||||||
if from_status == to_status:
|
if from_status == to_status:
|
||||||
return True # Même statut = OK (idempotence)
|
return True
|
||||||
|
|
||||||
allowed = ALLOWED_TRANSITIONS.get(from_status, [])
|
allowed = ALLOWED_TRANSITIONS.get(from_status, [])
|
||||||
return to_status in allowed
|
return to_status in allowed
|
||||||
|
|
||||||
|
|
||||||
def get_status_actions(local_status: str) -> Dict[str, any]:
|
def get_status_actions(local_status: str) -> Dict[str, any]:
|
||||||
"""
|
|
||||||
Obtient les actions à exécuter pour un statut
|
|
||||||
|
|
||||||
Args:
|
|
||||||
local_status: Statut local
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Dictionnaire des actions à exécuter
|
|
||||||
"""
|
|
||||||
return STATUS_ACTIONS.get(local_status, {})
|
return STATUS_ACTIONS.get(local_status, {})
|
||||||
|
|
||||||
|
|
||||||
def is_final_status(local_status: str) -> bool:
|
def is_final_status(local_status: str) -> bool:
|
||||||
"""
|
|
||||||
Détermine si le statut est final (pas de synchronisation nécessaire)
|
|
||||||
|
|
||||||
Args:
|
|
||||||
local_status: Statut local
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
True si statut final
|
|
||||||
"""
|
|
||||||
return local_status in ["SIGNE", "REFUSE", "EXPIRE"]
|
return local_status in ["SIGNE", "REFUSE", "EXPIRE"]
|
||||||
|
|
||||||
|
|
||||||
# ========================================
|
|
||||||
# PRIORITÉS DE STATUTS (POUR CONFLITS)
|
|
||||||
# ========================================
|
|
||||||
|
|
||||||
STATUS_PRIORITY: Dict[str, int] = {
|
STATUS_PRIORITY: Dict[str, int] = {
|
||||||
"""
|
"ERREUR": 0,
|
||||||
En cas de conflit de synchronisation, prendre le statut
|
|
||||||
avec la priorité la plus élevée
|
|
||||||
"""
|
|
||||||
"ERREUR": 0, # Priorité la plus basse
|
|
||||||
"EN_ATTENTE": 1,
|
"EN_ATTENTE": 1,
|
||||||
"EN_COURS": 2,
|
"EN_COURS": 2,
|
||||||
"EXPIRE": 3,
|
"EXPIRE": 3,
|
||||||
"REFUSE": 4,
|
"REFUSE": 4,
|
||||||
"SIGNE": 5, # Priorité la plus élevée (état final souhaité)
|
"SIGNE": 5,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
def resolve_status_conflict(status_a: str, status_b: str) -> str:
|
def resolve_status_conflict(status_a: str, status_b: str) -> str:
|
||||||
"""
|
|
||||||
Résout un conflit entre deux statuts (prend le plus prioritaire)
|
|
||||||
|
|
||||||
Args:
|
|
||||||
status_a: Premier statut
|
|
||||||
status_b: Second statut
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Statut prioritaire
|
|
||||||
"""
|
|
||||||
priority_a = STATUS_PRIORITY.get(status_a, 0)
|
priority_a = STATUS_PRIORITY.get(status_a, 0)
|
||||||
priority_b = STATUS_PRIORITY.get(status_b, 0)
|
priority_b = STATUS_PRIORITY.get(status_b, 0)
|
||||||
|
|
||||||
return status_a if priority_a >= priority_b else status_b
|
return status_a if priority_a >= priority_b else status_b
|
||||||
|
|
||||||
|
|
||||||
# ========================================
|
|
||||||
# MESSAGES UTILISATEUR
|
|
||||||
# ========================================
|
|
||||||
|
|
||||||
STATUS_MESSAGES: Dict[str, Dict[str, str]] = {
|
STATUS_MESSAGES: Dict[str, Dict[str, str]] = {
|
||||||
"EN_ATTENTE": {
|
"EN_ATTENTE": {
|
||||||
"fr": "Document en attente d'envoi",
|
"fr": "Document en attente d'envoi",
|
||||||
|
|
@ -257,16 +146,6 @@ STATUS_MESSAGES: Dict[str, Dict[str, str]] = {
|
||||||
|
|
||||||
|
|
||||||
def get_status_message(local_status: str, lang: str = "fr") -> str:
|
def get_status_message(local_status: str, lang: str = "fr") -> str:
|
||||||
"""
|
|
||||||
Obtient le message utilisateur pour un statut
|
|
||||||
|
|
||||||
Args:
|
|
||||||
local_status: Statut local
|
|
||||||
lang: Langue (fr, en)
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Message formaté
|
|
||||||
"""
|
|
||||||
status_info = STATUS_MESSAGES.get(local_status, {})
|
status_info = STATUS_MESSAGES.get(local_status, {})
|
||||||
icon = status_info.get("icon", "")
|
icon = status_info.get("icon", "")
|
||||||
message = status_info.get(lang, local_status)
|
message = status_info.get(lang, local_status)
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue