refactor(utils): reorganize status mapping functions and add docstrings
This commit is contained in:
parent
bac8cc6017
commit
0deb178bc6
1 changed files with 37 additions and 39 deletions
|
|
@ -1,4 +1,4 @@
|
|||
from typing import Dict
|
||||
from typing import Dict, Any
|
||||
|
||||
UNIVERSIGN_TO_LOCAL: Dict[str, str] = {
|
||||
"draft": "EN_ATTENTE",
|
||||
|
|
@ -20,7 +20,7 @@ LOCAL_TO_SAGE_STATUS: Dict[str, int] = {
|
|||
"ERREUR": 5,
|
||||
}
|
||||
|
||||
STATUS_ACTIONS: Dict[str, Dict[str, any]] = {
|
||||
STATUS_ACTIONS: Dict[str, Dict[str, Any]] = {
|
||||
"SIGNE": {
|
||||
"update_sage_status": True,
|
||||
"trigger_workflow": True,
|
||||
|
|
@ -60,34 +60,6 @@ ALLOWED_TRANSITIONS: Dict[str, list] = {
|
|||
"ERREUR": ["EN_ATTENTE", "EN_COURS"],
|
||||
}
|
||||
|
||||
|
||||
def map_universign_to_local(universign_status: str) -> str:
|
||||
return UNIVERSIGN_TO_LOCAL.get(
|
||||
universign_status.lower(),
|
||||
"ERREUR",
|
||||
)
|
||||
|
||||
|
||||
def get_sage_status_code(local_status: str) -> int:
|
||||
return LOCAL_TO_SAGE_STATUS.get(local_status, 5)
|
||||
|
||||
|
||||
def is_transition_allowed(from_status: str, to_status: str) -> bool:
|
||||
if from_status == to_status:
|
||||
return True
|
||||
|
||||
allowed = ALLOWED_TRANSITIONS.get(from_status, [])
|
||||
return to_status in allowed
|
||||
|
||||
|
||||
def get_status_actions(local_status: str) -> Dict[str, any]:
|
||||
return STATUS_ACTIONS.get(local_status, {})
|
||||
|
||||
|
||||
def is_final_status(local_status: str) -> bool:
|
||||
return local_status in ["SIGNE", "REFUSE", "EXPIRE"]
|
||||
|
||||
|
||||
STATUS_PRIORITY: Dict[str, int] = {
|
||||
"ERREUR": 0,
|
||||
"EN_ATTENTE": 1,
|
||||
|
|
@ -97,14 +69,6 @@ STATUS_PRIORITY: Dict[str, int] = {
|
|||
"SIGNE": 5,
|
||||
}
|
||||
|
||||
|
||||
def resolve_status_conflict(status_a: str, status_b: str) -> str:
|
||||
priority_a = STATUS_PRIORITY.get(status_a, 0)
|
||||
priority_b = STATUS_PRIORITY.get(status_b, 0)
|
||||
|
||||
return status_a if priority_a >= priority_b else status_b
|
||||
|
||||
|
||||
STATUS_MESSAGES: Dict[str, Dict[str, str]] = {
|
||||
"EN_ATTENTE": {
|
||||
"fr": "Document en attente d'envoi",
|
||||
|
|
@ -145,9 +109,43 @@ STATUS_MESSAGES: Dict[str, Dict[str, str]] = {
|
|||
}
|
||||
|
||||
|
||||
def map_universign_to_local(universign_status: str) -> str:
|
||||
"""Convertit un statut Universign en statut local."""
|
||||
return UNIVERSIGN_TO_LOCAL.get(universign_status.lower(), "ERREUR")
|
||||
|
||||
|
||||
def get_sage_status_code(local_status: str) -> int:
|
||||
"""Obtient le code numérique pour Sage."""
|
||||
return LOCAL_TO_SAGE_STATUS.get(local_status, 5)
|
||||
|
||||
|
||||
def is_transition_allowed(from_status: str, to_status: str) -> bool:
|
||||
"""Vérifie si une transition de statut est valide."""
|
||||
if from_status == to_status:
|
||||
return True
|
||||
return to_status in ALLOWED_TRANSITIONS.get(from_status, [])
|
||||
|
||||
|
||||
def get_status_actions(local_status: str) -> Dict[str, Any]:
|
||||
"""Obtient les actions à exécuter pour un statut."""
|
||||
return STATUS_ACTIONS.get(local_status, {})
|
||||
|
||||
|
||||
def is_final_status(local_status: str) -> bool:
|
||||
"""Détermine si le statut est final."""
|
||||
return local_status in ["SIGNE", "REFUSE", "EXPIRE"]
|
||||
|
||||
|
||||
def resolve_status_conflict(status_a: str, status_b: str) -> str:
|
||||
"""Résout un conflit entre deux statuts (prend le plus prioritaire)."""
|
||||
priority_a = STATUS_PRIORITY.get(status_a, 0)
|
||||
priority_b = STATUS_PRIORITY.get(status_b, 0)
|
||||
return status_a if priority_a >= priority_b else status_b
|
||||
|
||||
|
||||
def get_status_message(local_status: str, lang: str = "fr") -> str:
|
||||
"""Obtient le message utilisateur pour un statut."""
|
||||
status_info = STATUS_MESSAGES.get(local_status, {})
|
||||
icon = status_info.get("icon", "")
|
||||
message = status_info.get(lang, local_status)
|
||||
|
||||
return f"{icon} {message}"
|
||||
|
|
|
|||
Loading…
Reference in a new issue