fanilo/Sage100-vps
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(signer-status): add new signer statuses and improve status handling

Browse source
This commit is contained in:
Fanilo-Nantenaina 2026-01-06 20:24:29 +03:00
parent fbaa43e3fd
commit bcaa621432
3 changed files with 48 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
database/models/universign.py
View file

@ -29,10 +29,14 @@ class UniversignTransactionStatus(str, Enum):
class UniversignSignerStatus(str, Enum):
WAITING = "waiting"
OPEN = "open"
VIEWED = "viewed"
SIGNED = "signed"
COMPLETED = "completed"
REFUSED = "refused"
EXPIRED = "expired"
STALLED = "stalled"
UNKNOWN = "unknown"
class LocalDocumentStatus(str, Enum):

62
services/universign_sync.py
View file

@ -251,46 +251,38 @@ class UniversignSyncService:
transaction: UniversignTransaction,
universign_data: Dict,
):
"""
CORRECTION : Synchronise les signataires sans perdre les données locales
"""
# Récupérer les participants depuis différents endroits possibles
signers_data = universign_data.get("participants", [])
if not signers_data:
signers_data = universign_data.get("signers", [])
# ⚠️ IMPORTANT : Ne pas toucher aux signers si Universign n'en retourne pas
if not signers_data:
logger.debug(
"Aucun signataire dans les données Universign, conservation des données locales"
)
logger.debug("Aucun signataire dans les données Universign")
return
# Créer un mapping email -> signer existant
existing_signers = {s.email: s for s in transaction.signers}
for idx, signer_data in enumerate(signers_data):
email = signer_data.get("email", "")
if not email:
logger.warning(f"Signataire sans email à l'index {idx}, ignoré")
continue
# ✅ PROTECTION : gérer les statuts inconnus
raw_status = signer_data.get("status") or signer_data.get(
"state", "waiting"
)
try:
status = UniversignSignerStatus(raw_status)
except ValueError:
logger.warning(
f"Statut inconnu pour signer {email}: {raw_status}, utilisation de 'unknown'"
)
status = UniversignSignerStatus.UNKNOWN
if email in existing_signers:
# ✅ Mise à jour du signer existant (ne pas écraser si None)
signer = existing_signers[email]
signer.status = status
# Mise à jour du statut
new_status = signer_data.get("status") or signer_data.get("state")
if new_status:
try:
signer.status = UniversignSignerStatus(new_status)
except ValueError:
logger.warning(
f"Statut inconnu pour signer {email}: {new_status}"
)
# Mise à jour des dates (ne pas écraser si déjà renseignées)
viewed_at = self._parse_date(signer_data.get("viewed_at"))
if viewed_at and not signer.viewed_at:
signer.viewed_at = viewed_at
@ -303,29 +295,26 @@ class UniversignSyncService:
if refused_at and not signer.refused_at:
signer.refused_at = refused_at
# Mise à jour du nom si manquant
if signer_data.get("name") and not signer.name:
signer.name = signer_data.get("name")
else:
# ✅ Nouveau signer
# ✅ Nouveau signer avec gestion d'erreur intégrée
try:
status = signer_data.get("status") or signer_data.get(
"state", "waiting"
)
signer = UniversignSigner(
id=f"{transaction.id}_signer_{idx}_{int(datetime.now().timestamp())}",
transaction_id=transaction.id,
email=email,
name=signer_data.get("name"),
status=UniversignSignerStatus(status),
status=status,
order_index=idx,
viewed_at=self._parse_date(signer_data.get("viewed_at")),
signed_at=self._parse_date(signer_data.get("signed_at")),
refused_at=self._parse_date(signer_data.get("refused_at")),
)
session.add(signer)
logger.info(f"➕ Nouveau signataire ajouté: {email}")
logger.info(
f"➕ Nouveau signataire ajouté: {email} (statut: {status.value})"
)
except Exception as e:
logger.error(f"Erreur création signer {email}: {e}")
@ -534,15 +523,22 @@ class UniversignSyncService:
self, session: AsyncSession, transaction: UniversignTransaction, new_status: str
):
actions = get_status_actions(new_status)
if not actions:
return
if actions.get("update_sage_status"):
if actions.get("update_sage_status") and self.sage_client:
await self._update_sage_status(transaction, new_status)
elif actions.get("update_sage_status"):
logger.debug(
f"sage_client non configuré, skip MAJ Sage pour {transaction.sage_document_id}"
)
if actions.get("send_notification"):
if actions.get("send_notification") and self.email_queue and self.settings:
await self._send_notification(session, transaction, new_status)
elif actions.get("send_notification"):
logger.debug(
f"email_queue/settings non configuré, skip notification pour {transaction.transaction_id}"
)
async def _update_sage_status(
self, transaction: UniversignTransaction, status: str

17
utils/universign_status_mapping.py
View file

@ -1,4 +1,8 @@
from typing import Dict, Any
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
UNIVERSIGN_TO_LOCAL: Dict[str, str] = {
"draft": "EN_ATTENTE",
@ -111,8 +115,17 @@ 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")
"""Convertit un statut Universign en statut local avec fallback robuste."""
normalized = universign_status.lower().strip()
mapped = UNIVERSIGN_TO_LOCAL.get(normalized)
if not mapped:
logger.warning(
f"Statut Universign inconnu: '{universign_status}', mapping vers ERREUR"
)
return "ERREUR"
return mapped
def get_sage_status_code(local_status: str) -> int: