561 lines
16 KiB
Python
561 lines
16 KiB
Python
from typing import Dict, List, Optional
|
|
from config.config import settings
|
|
import logging
|
|
from enum import Enum
|
|
|
|
from datetime import datetime
|
|
import uuid
|
|
import requests
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from data.data import templates_signature_email
|
|
from database import EmailLog, StatutEmail as StatutEmailEnum
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def universign_envoyer(
|
|
doc_id: str,
|
|
pdf_bytes: bytes,
|
|
email: str,
|
|
nom: str,
|
|
doc_data: Dict,
|
|
session: AsyncSession,
|
|
) -> Dict:
|
|
from email_queue import email_queue
|
|
|
|
try:
|
|
api_key = settings.universign_api_key
|
|
api_url = settings.universign_api_url
|
|
auth = (api_key, "")
|
|
|
|
logger.info(f" Démarrage processus Universign pour {email}")
|
|
logger.info(f"Document: {doc_id} ({doc_data.get('type_label')})")
|
|
|
|
if not pdf_bytes or len(pdf_bytes) == 0:
|
|
raise Exception("Le PDF généré est vide")
|
|
|
|
logger.info(f"PDF valide : {len(pdf_bytes)} octets")
|
|
|
|
logger.info("ÉTAPE 1/6 : Création transaction")
|
|
|
|
response = requests.post(
|
|
f"{api_url}/transactions",
|
|
auth=auth,
|
|
json={
|
|
"name": f"{doc_data.get('type_label', 'Document')} {doc_id}",
|
|
"language": "fr",
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
logger.error(f"Erreur création transaction: {response.text}")
|
|
raise Exception(f"Erreur création transaction: {response.status_code}")
|
|
|
|
transaction_id = response.json().get("id")
|
|
logger.info(f"Transaction créée: {transaction_id}")
|
|
|
|
logger.info("ÉTAPE 2/6 : Upload PDF")
|
|
|
|
files = {
|
|
"file": (
|
|
f"{doc_data.get('type_label', 'Document')}_{doc_id}.pdf",
|
|
pdf_bytes,
|
|
"application/pdf",
|
|
)
|
|
}
|
|
|
|
response = requests.post(
|
|
f"{api_url}/files",
|
|
auth=auth,
|
|
files=files,
|
|
timeout=60,
|
|
)
|
|
|
|
if response.status_code not in [200, 201]:
|
|
logger.error(f"Erreur upload: {response.text}")
|
|
raise Exception(f"Erreur upload fichier: {response.status_code}")
|
|
|
|
file_id = response.json().get("id")
|
|
logger.info(f"Fichier uploadé: {file_id}")
|
|
|
|
logger.info("ÉTAPE 3/6 : Ajout document à transaction")
|
|
|
|
response = requests.post(
|
|
f"{api_url}/transactions/{transaction_id}/documents",
|
|
auth=auth,
|
|
data={"document": file_id},
|
|
timeout=30,
|
|
)
|
|
|
|
if response.status_code not in [200, 201]:
|
|
logger.error(f"Erreur ajout document: {response.text}")
|
|
raise Exception(f"Erreur ajout document: {response.status_code}")
|
|
|
|
document_id = response.json().get("id")
|
|
logger.info(f"Document ajouté: {document_id}")
|
|
|
|
logger.info("ÉTAPE 4/6 : Création champ signature")
|
|
|
|
response = requests.post(
|
|
f"{api_url}/transactions/{transaction_id}/documents/{document_id}/fields",
|
|
auth=auth,
|
|
data={
|
|
"type": "signature",
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
if response.status_code not in [200, 201]:
|
|
logger.error(f"Erreur création champ: {response.text}")
|
|
raise Exception(f"Erreur création champ: {response.status_code}")
|
|
|
|
field_id = response.json().get("id")
|
|
logger.info(f"Champ créé: {field_id}")
|
|
|
|
logger.info(" ÉTAPE 5/6 : Liaison signataire au champ")
|
|
|
|
response = requests.post(
|
|
f"{api_url}/transactions/{transaction_id}/signatures", # /signatures pas /signers
|
|
auth=auth,
|
|
data={
|
|
"signer": email,
|
|
"field": field_id,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
if response.status_code not in [200, 201]:
|
|
logger.error(f"Erreur liaison signataire: {response.text}")
|
|
raise Exception(f"Erreur liaison signataire: {response.status_code}")
|
|
|
|
logger.info(f"Signataire lié: {email}")
|
|
|
|
logger.info("ÉTAPE 6/6 : Démarrage transaction")
|
|
|
|
response = requests.post(
|
|
f"{api_url}/transactions/{transaction_id}/start", auth=auth, timeout=30
|
|
)
|
|
|
|
if response.status_code not in [200, 201]:
|
|
logger.error(f"Erreur démarrage: {response.text}")
|
|
raise Exception(f"Erreur démarrage: {response.status_code}")
|
|
|
|
final_data = response.json()
|
|
logger.info("Transaction démarrée")
|
|
|
|
logger.info("Récupération URL de signature")
|
|
|
|
signer_url = ""
|
|
|
|
if final_data.get("actions"):
|
|
for action in final_data["actions"]:
|
|
if action.get("url"):
|
|
signer_url = action["url"]
|
|
break
|
|
|
|
if not signer_url and final_data.get("signers"):
|
|
for signer in final_data["signers"]:
|
|
if signer.get("email") == email:
|
|
signer_url = signer.get("url", "")
|
|
break
|
|
|
|
if not signer_url:
|
|
logger.error(f"URL introuvable dans: {final_data}")
|
|
raise ValueError("URL de signature non retournée par Universign")
|
|
|
|
logger.info("URL récupérée")
|
|
|
|
logger.info(" Préparation email")
|
|
|
|
template = templates_signature_email["demande_signature"]
|
|
|
|
type_labels = {
|
|
0: "Devis",
|
|
10: "Commande",
|
|
30: "Bon de Livraison",
|
|
60: "Facture",
|
|
50: "Avoir",
|
|
}
|
|
|
|
variables = {
|
|
"NOM_SIGNATAIRE": nom,
|
|
"TYPE_DOC": type_labels.get(doc_data.get("type_doc", 0), "Document"),
|
|
"NUMERO": doc_id,
|
|
"DATE": doc_data.get("date", datetime.now().strftime("%d/%m/%Y")),
|
|
"MONTANT_TTC": f"{doc_data.get('montant_ttc', 0):.2f}",
|
|
"SIGNER_URL": signer_url,
|
|
"CONTACT_EMAIL": settings.smtp_from,
|
|
}
|
|
|
|
sujet = template["sujet"]
|
|
corps = template["corps_html"]
|
|
|
|
for var, valeur in variables.items():
|
|
sujet = sujet.replace(f"{{{{{var}}}}}", str(valeur))
|
|
corps = corps.replace(f"{{{{{var}}}}}", str(valeur))
|
|
|
|
email_log = EmailLog(
|
|
id=str(uuid.uuid4()),
|
|
destinataire=email,
|
|
sujet=sujet,
|
|
corps_html=corps,
|
|
document_ids=doc_id,
|
|
type_document=doc_data.get("type_doc"),
|
|
statut=StatutEmailEnum.EN_ATTENTE,
|
|
date_creation=datetime.now(),
|
|
nb_tentatives=0,
|
|
)
|
|
|
|
session.add(email_log)
|
|
await session.flush()
|
|
|
|
email_queue.enqueue(email_log.id)
|
|
|
|
logger.info(f"Email mis en file pour {email}")
|
|
logger.info("🎉 Processus terminé avec succès")
|
|
|
|
return {
|
|
"transaction_id": transaction_id,
|
|
"signer_url": signer_url,
|
|
"statut": "ENVOYE",
|
|
"email_log_id": email_log.id,
|
|
"email_sent": True,
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erreur Universign: {e}", exc_info=True)
|
|
return {
|
|
"error": str(e),
|
|
"statut": "ERREUR",
|
|
"email_sent": False,
|
|
}
|
|
|
|
|
|
async def universign_statut(transaction_id: str) -> Dict:
|
|
"""Récupération statut signature"""
|
|
import requests
|
|
|
|
try:
|
|
response = requests.get(
|
|
f"{settings.universign_api_url}/transactions/{transaction_id}",
|
|
auth=(settings.universign_api_key, ""),
|
|
timeout=10,
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
statut_map = {
|
|
"draft": "EN_ATTENTE",
|
|
"started": "EN_ATTENTE",
|
|
"completed": "SIGNE",
|
|
"refused": "REFUSE",
|
|
"expired": "EXPIRE",
|
|
"canceled": "REFUSE",
|
|
}
|
|
return {
|
|
"statut": statut_map.get(data.get("state"), "EN_ATTENTE"),
|
|
"date_signature": data.get("completed_at"),
|
|
}
|
|
else:
|
|
return {"statut": "ERREUR"}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erreur statut Universign: {e}")
|
|
return {"statut": "ERREUR", "error": str(e)}
|
|
|
|
|
|
def normaliser_type_doc(type_doc: int) -> int:
|
|
TYPES_AUTORISES = {0, 10, 30, 50, 60}
|
|
|
|
if type_doc not in TYPES_AUTORISES:
|
|
raise ValueError(
|
|
f"type_doc invalide ({type_doc}). Valeurs autorisées : {sorted(TYPES_AUTORISES)}"
|
|
)
|
|
|
|
return type_doc if type_doc == 0 else type_doc // 10
|
|
|
|
|
|
def _preparer_lignes_document(lignes: List) -> List[Dict]:
|
|
return [
|
|
{
|
|
"article_code": ligne.article_code,
|
|
"quantite": ligne.quantite,
|
|
"prix_unitaire_ht": ligne.prix_unitaire_ht,
|
|
"remise_pourcentage": ligne.remise_pourcentage or 0.0,
|
|
}
|
|
for ligne in lignes
|
|
]
|
|
|
|
|
|
# ========================================
|
|
# MAPPING UNIVERSIGN → LOCAL
|
|
# ========================================
|
|
|
|
UNIVERSIGN_TO_LOCAL: Dict[str, str] = {
|
|
# États initiaux
|
|
"draft": "EN_ATTENTE", # Transaction créée
|
|
"ready": "EN_ATTENTE", # Prête mais pas envoyée
|
|
# En cours
|
|
"started": "EN_COURS", # Envoyée, en attente de signature
|
|
# États finaux (succès)
|
|
"completed": "SIGNE", # ✓ Signé avec succès
|
|
# États finaux (échec)
|
|
"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] = {
|
|
"""
|
|
À stocker dans un champ libre Sage (ex: CB_STATUT_SIGNATURE)
|
|
"""
|
|
"EN_ATTENTE": 0, # Non envoyé
|
|
"EN_COURS": 1, # Envoyé, en attente
|
|
"SIGNE": 2, # ✓ Signé (peut déclencher workflow)
|
|
"REFUSE": 3, # ✗ Refusé
|
|
"EXPIRE": 4, # ⏰ Expiré
|
|
"ERREUR": 5, # ⚠️ Erreur
|
|
}
|
|
|
|
|
|
# ========================================
|
|
# ACTIONS MÉTIER PAR STATUT
|
|
# ========================================
|
|
|
|
STATUS_ACTIONS: Dict[str, Dict[str, any]] = {
|
|
"""
|
|
Actions automatiques à déclencher selon le statut
|
|
"""
|
|
"SIGNE": {
|
|
"update_sage_status": True, # Mettre à jour Sage
|
|
"trigger_workflow": True, # Déclencher transformation (devis→commande)
|
|
"send_notification": True, # Email de confirmation
|
|
"archive_document": True, # Archiver le PDF signé
|
|
"update_sage_field": "CB_DateSignature", # Champ libre Sage
|
|
},
|
|
"REFUSE": {
|
|
"update_sage_status": True,
|
|
"trigger_workflow": False,
|
|
"send_notification": True,
|
|
"archive_document": False,
|
|
"alert_sales": True, # Alerter commercial
|
|
},
|
|
"EXPIRE": {
|
|
"update_sage_status": True,
|
|
"trigger_workflow": False,
|
|
"send_notification": True,
|
|
"archive_document": False,
|
|
"schedule_reminder": True, # Programmer relance
|
|
},
|
|
"ERREUR": {
|
|
"update_sage_status": False,
|
|
"trigger_workflow": False,
|
|
"send_notification": False,
|
|
"log_error": True,
|
|
"retry_sync": True,
|
|
},
|
|
}
|
|
|
|
|
|
# ========================================
|
|
# RÈGLES DE TRANSITION
|
|
# ========================================
|
|
|
|
ALLOWED_TRANSITIONS: Dict[str, list] = {
|
|
"""
|
|
Transitions de statuts autorisées (validation)
|
|
"""
|
|
"EN_ATTENTE": ["EN_COURS", "ERREUR"],
|
|
"EN_COURS": ["SIGNE", "REFUSE", "EXPIRE", "ERREUR"],
|
|
"SIGNE": [], # État final, pas de retour
|
|
"REFUSE": [], # État final
|
|
"EXPIRE": [], # État final
|
|
"ERREUR": ["EN_ATTENTE", "EN_COURS"], # Retry possible
|
|
}
|
|
|
|
|
|
# ========================================
|
|
# FONCTION DE MAPPING
|
|
# ========================================
|
|
|
|
|
|
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(
|
|
universign_status.lower(),
|
|
"ERREUR", # Fallback si statut inconnu
|
|
)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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:
|
|
return True # Même statut = OK (idempotence)
|
|
|
|
allowed = ALLOWED_TRANSITIONS.get(from_status, [])
|
|
return to_status in allowed
|
|
|
|
|
|
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, {})
|
|
|
|
|
|
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"]
|
|
|
|
|
|
# ========================================
|
|
# PRIORITÉS DE STATUTS (POUR CONFLITS)
|
|
# ========================================
|
|
|
|
STATUS_PRIORITY: Dict[str, int] = {
|
|
"""
|
|
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_COURS": 2,
|
|
"EXPIRE": 3,
|
|
"REFUSE": 4,
|
|
"SIGNE": 5, # Priorité la plus élevée (état final souhaité)
|
|
}
|
|
|
|
|
|
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_b = STATUS_PRIORITY.get(status_b, 0)
|
|
|
|
return status_a if priority_a >= priority_b else status_b
|
|
|
|
|
|
# ========================================
|
|
# MESSAGES UTILISATEUR
|
|
# ========================================
|
|
|
|
STATUS_MESSAGES: Dict[str, Dict[str, str]] = {
|
|
"EN_ATTENTE": {
|
|
"fr": "Document en attente d'envoi",
|
|
"en": "Document pending",
|
|
"icon": "⏳",
|
|
"color": "gray",
|
|
},
|
|
"EN_COURS": {
|
|
"fr": "En attente de signature",
|
|
"en": "Awaiting signature",
|
|
"icon": "✍️",
|
|
"color": "blue",
|
|
},
|
|
"SIGNE": {
|
|
"fr": "Signé avec succès",
|
|
"en": "Successfully signed",
|
|
"icon": "✅",
|
|
"color": "green",
|
|
},
|
|
"REFUSE": {
|
|
"fr": "Signature refusée",
|
|
"en": "Signature refused",
|
|
"icon": "❌",
|
|
"color": "red",
|
|
},
|
|
"EXPIRE": {
|
|
"fr": "Délai de signature expiré",
|
|
"en": "Signature expired",
|
|
"icon": "⏰",
|
|
"color": "orange",
|
|
},
|
|
"ERREUR": {
|
|
"fr": "Erreur technique",
|
|
"en": "Technical error",
|
|
"icon": "⚠️",
|
|
"color": "red",
|
|
},
|
|
}
|
|
|
|
|
|
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, {})
|
|
icon = status_info.get("icon", "")
|
|
message = status_info.get(lang, local_status)
|
|
|
|
return f"{icon} {message}"
|
|
|
|
|
|
__all__ = ["_preparer_lignes_document", "normaliser_type_doc"]
|