style(api): improve email template readability and clean up code formatting

This commit is contained in:
Fanilo-Nantenaina 2025-12-18 10:59:40 +03:00
parent 282ffe4898
commit 5cb9015ab5

306
api.py
View file

@ -1050,8 +1050,6 @@ class MouvementStockResponse(BaseModel):
nb_lignes: int = Field(..., description="Nombre de lignes")
templates_signature_email = {
"demande_signature": {
"id": "demande_signature",
@ -1073,7 +1071,7 @@ templates_signature_email = {
<!-- Header -->
<tr>
<td style="background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); padding: 30px; text-align: center; border-radius: 8px 8px 0 0;">
<h1 style="color: #ffffff; margin: 0; font-size: 24px; font-weight: 600;">
<h1 style="color: #000; margin: 0; font-size: 24px; font-weight: 600;">
📝 Signature Électronique Requise
</h1>
</td>
@ -1124,7 +1122,7 @@ templates_signature_email = {
<table width="100%" cellpadding="0" cellspacing="0">
<tr>
<td align="center" style="padding: 10px 0 30px;">
<a href="{{SIGNER_URL}}" style="display: inline-block; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: #ffffff; text-decoration: none; padding: 16px 40px; border-radius: 6px; font-size: 16px; font-weight: 600; box-shadow: 0 4px 12px rgba(102, 126, 234, 0.4);">
<a href="{{SIGNER_URL}}" style="display: inline-block; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: #000; text-decoration: none; padding: 16px 40px; border-radius: 6px; font-size: 16px; font-weight: 600; box-shadow: 0 4px 12px rgba(102, 126, 234, 0.4);">
Signer le document
</a>
</td>
@ -1179,10 +1177,9 @@ templates_signature_email = {
"DATE",
"MONTANT_TTC",
"SIGNER_URL",
"CONTACT_EMAIL"
]
"CONTACT_EMAIL",
],
},
"signature_confirmee": {
"id": "signature_confirmee",
"nom": "Confirmation de Signature",
@ -1290,10 +1287,9 @@ templates_signature_email = {
"NUMERO",
"DATE_SIGNATURE",
"TRANSACTION_ID",
"CONTACT_EMAIL"
]
"CONTACT_EMAIL",
],
},
"relance_signature": {
"id": "relance_signature",
"nom": "Relance Signature en Attente",
@ -1393,22 +1389,22 @@ templates_signature_email = {
"NB_JOURS",
"JOURS_RESTANTS",
"SIGNER_URL",
"CONTACT_EMAIL"
]
}
"CONTACT_EMAIL",
],
},
}
async def universign_envoyer_avec_email(
doc_id: str,
pdf_bytes: bytes,
email: str,
doc_id: str,
pdf_bytes: bytes,
email: str,
nom: str,
doc_data: Dict, # Données du document (type, montant, date, etc.)
session: AsyncSession
session: AsyncSession,
) -> Dict:
import requests
try:
api_key = settings.universign_api_key
api_url = settings.universign_api_url
@ -1419,20 +1415,26 @@ async def universign_envoyer_avec_email(
auth=auth,
json={
"name": f"{doc_data.get('type_label', 'Document')} {doc_id}",
"language": "fr"
"language": "fr",
},
timeout=30,
)
response.raise_for_status()
transaction_id = response.json().get("id")
logger.info(f"✅ Transaction Universign créée: {transaction_id}")
files = {"file": (f"{doc_data.get('type_label', 'Document')}_{doc_id}.pdf", pdf_bytes, "application/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=30)
response.raise_for_status()
file_id = response.json().get("id")
response = requests.post(
f"{api_url}/transactions/{transaction_id}/documents",
auth=auth,
@ -1450,7 +1452,7 @@ async def universign_envoyer_avec_email(
)
response.raise_for_status()
field_id = response.json().get("id")
response = requests.post(
f"{api_url}/transactions/{transaction_id}/signatures",
auth=auth,
@ -1460,35 +1462,33 @@ async def universign_envoyer_avec_email(
response.raise_for_status()
response = requests.post(
f"{api_url}/transactions/{transaction_id}/start",
auth=auth,
timeout=30
f"{api_url}/transactions/{transaction_id}/start", auth=auth, timeout=30
)
response.raise_for_status()
final_data = response.json()
signer_url = (
final_data.get("actions", [{}])[0].get("url", "")
if final_data.get("actions")
else ""
)
if not signer_url:
raise ValueError("URL de signature non retournée par Universign")
logger.info(f"✅ Signature Universign démarrée: {transaction_id}")
template = templates_signature_email["demande_signature"]
# Préparer les variables
type_labels = {
0: "Devis",
10: "Commande",
10: "Commande",
30: "Bon de Livraison",
60: "Facture",
50: "Avoir"
50: "Avoir",
}
variables = {
"NOM_SIGNATAIRE": nom,
"TYPE_DOC": type_labels.get(doc_data.get("type_doc", 0), "Document"),
@ -1496,17 +1496,17 @@ async def universign_envoyer_avec_email(
"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
"CONTACT_EMAIL": settings.smtp_from,
}
# Remplacer les variables dans le template
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))
# Créer log email
email_log = EmailLog(
id=str(uuid.uuid4()),
@ -1519,30 +1519,27 @@ async def universign_envoyer_avec_email(
date_creation=datetime.now(),
nb_tentatives=0,
)
session.add(email_log)
await session.flush()
# Enqueue l'email
email_queue.enqueue(email_log.id)
logger.info(f"📧 Email de signature envoyé en file: {email}")
return {
"transaction_id": transaction_id,
"signer_url": signer_url,
"statut": "ENVOYE",
"email_log_id": email_log.id,
"email_sent": True
"email_sent": True,
}
except Exception as e:
logger.error(f"❌ Erreur Universign+Email: {e}")
return {
"error": str(e),
"statut": "ERREUR",
"email_sent": False
}
return {"error": str(e), "statut": "ERREUR", "email_sent": False}
async def universign_statut(transaction_id: str) -> Dict:
"""Récupération statut signature"""
@ -2398,6 +2395,7 @@ async def commande_vers_facture(id: str, session: AsyncSession = Depends(get_ses
logger.error(f"Erreur transformation: {e}")
raise HTTPException(500, str(e))
def normaliser_type_doc(type_doc: int) -> int:
TYPES_AUTORISES = {0, 10, 30, 50, 60}
@ -2411,32 +2409,35 @@ def normaliser_type_doc(type_doc: int) -> int:
@app.post("/signature/universign/send", tags=["Signatures"])
async def envoyer_signature_optimise(
demande: SignatureRequest,
session: AsyncSession = Depends(get_session)
demande: SignatureRequest, session: AsyncSession = Depends(get_session)
):
try:
# Récupérer le document depuis Sage
doc = sage_client.lire_document(demande.doc_id, normaliser_type_doc(demande.type_doc))
doc = sage_client.lire_document(
demande.doc_id, normaliser_type_doc(demande.type_doc)
)
if not doc:
raise HTTPException(404, f"Document {demande.doc_id} introuvable")
# Générer PDF
pdf_bytes = email_queue._generate_pdf(demande.doc_id, normaliser_type_doc(demande.type_doc))
pdf_bytes = email_queue._generate_pdf(
demande.doc_id, normaliser_type_doc(demande.type_doc)
)
# Préparer les données du document pour l'email
doc_data = {
"type_doc": demande.type_doc,
"type_label": {
0: "Devis",
10: "Commande",
30: "Bon de Livraison",
30: "Bon de Livraison",
60: "Facture",
50: "Avoir"
50: "Avoir",
}.get(demande.type_doc, "Document"),
"montant_ttc": doc.get("total_ttc", 0),
"date": doc.get("date", datetime.now().strftime("%d/%m/%Y"))
"date": doc.get("date", datetime.now().strftime("%d/%m/%Y")),
}
# Envoi Universign + Email automatique
resultat = await universign_envoyer_avec_email(
doc_id=demande.doc_id,
@ -2444,12 +2445,12 @@ async def envoyer_signature_optimise(
email=demande.email_signataire,
nom=demande.nom_signataire,
doc_data=doc_data,
session=session
session=session,
)
if "error" in resultat:
raise HTTPException(500, resultat["error"])
# Logger en DB
signature_log = SignatureLog(
id=str(uuid.uuid4()),
@ -2462,97 +2463,98 @@ async def envoyer_signature_optimise(
statut=StatutSignatureEnum.ENVOYE,
date_envoi=datetime.now(),
)
session.add(signature_log)
await session.commit()
# MAJ champ libre Sage
sage_client.mettre_a_jour_champ_libre(
demande.doc_id,
demande.type_doc,
"UniversignID",
resultat["transaction_id"]
demande.doc_id, demande.type_doc, "UniversignID", resultat["transaction_id"]
)
logger.info(f"✅ Signature envoyée: {demande.doc_id} (Email: {resultat['email_sent']})")
logger.info(
f"✅ Signature envoyée: {demande.doc_id} (Email: {resultat['email_sent']})"
)
return {
"success": True,
"transaction_id": resultat["transaction_id"],
"signer_url": resultat["signer_url"],
"email_sent": resultat["email_sent"],
"email_log_id": resultat.get("email_log_id"),
"message": f"Demande de signature envoyée à {demande.email_signataire}"
"message": f"Demande de signature envoyée à {demande.email_signataire}",
}
except HTTPException:
raise
except Exception as e:
logger.error(f"❌ Erreur signature: {e}")
raise HTTPException(500, str(e))
@app.post("/webhooks/universign", tags=["Signatures"])
async def webhook_universign(
request: Request,
session: AsyncSession = Depends(get_session)
request: Request, session: AsyncSession = Depends(get_session)
):
try:
payload = await request.json()
event_type = payload.get("event")
transaction_id = payload.get("transaction_id")
if not transaction_id:
logger.warning("⚠️ Webhook sans transaction_id")
return {"status": "ignored"}
# Chercher la signature dans la DB
query = select(SignatureLog).where(
SignatureLog.transaction_id == transaction_id
)
result = await session.execute(query)
signature_log = result.scalar_one_or_none()
if not signature_log:
logger.warning(f"⚠️ Transaction {transaction_id} introuvable en DB")
return {"status": "not_found"}
# =============================================
# TRAITER L'EVENT SELON LE TYPE
# =============================================
if event_type == "transaction.completed":
# ✅ SIGNATURE RÉUSSIE
signature_log.statut = StatutSignatureEnum.SIGNE
signature_log.date_signature = datetime.now()
logger.info(f"✅ Signature confirmée: {signature_log.document_id}")
# ENVOYER EMAIL DE CONFIRMATION
template = templates_signature_email["signature_confirmee"]
type_labels = {
0: "Devis", 10: "Commande", 30: "Bon de Livraison",
60: "Facture", 50: "Avoir"
0: "Devis",
10: "Commande",
30: "Bon de Livraison",
60: "Facture",
50: "Avoir",
}
variables = {
"NOM_SIGNATAIRE": signature_log.nom_signataire,
"TYPE_DOC": type_labels.get(signature_log.type_document, "Document"),
"NUMERO": signature_log.document_id,
"DATE_SIGNATURE": datetime.now().strftime("%d/%m/%Y à %H:%M"),
"TRANSACTION_ID": transaction_id,
"CONTACT_EMAIL": settings.smtp_from
"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))
# Créer email de confirmation
email_log = EmailLog(
id=str(uuid.uuid4()),
@ -2565,77 +2567,80 @@ async def webhook_universign(
date_creation=datetime.now(),
nb_tentatives=0,
)
session.add(email_log)
email_queue.enqueue(email_log.id)
logger.info(f"📧 Email de confirmation envoyé: {signature_log.email_signataire}")
logger.info(
f"📧 Email de confirmation envoyé: {signature_log.email_signataire}"
)
elif event_type == "transaction.refused":
# ❌ SIGNATURE REFUSÉE
signature_log.statut = StatutSignatureEnum.REFUSE
logger.warning(f"❌ Signature refusée: {signature_log.document_id}")
elif event_type == "transaction.expired":
# ⏰ TRANSACTION EXPIRÉE
signature_log.statut = StatutSignatureEnum.EXPIRE
logger.warning(f"⏰ Transaction expirée: {signature_log.document_id}")
await session.commit()
return {
"status": "processed",
"event": event_type,
"transaction_id": transaction_id
"transaction_id": transaction_id,
}
except Exception as e:
logger.error(f"❌ Erreur webhook Universign: {e}")
return {"status": "error", "message": str(e)}
@app.get("/admin/signatures/relances-auto", tags=["Admin"])
async def relancer_signatures_automatique(
session: AsyncSession = Depends(get_session)
):
async def relancer_signatures_automatique(session: AsyncSession = Depends(get_session)):
try:
from datetime import timedelta
# Chercher signatures en attente depuis > 7 jours
date_limite = datetime.now() - timedelta(days=7)
query = select(SignatureLog).where(
SignatureLog.statut.in_([
StatutSignatureEnum.EN_ATTENTE,
StatutSignatureEnum.ENVOYE
]),
SignatureLog.statut.in_(
[StatutSignatureEnum.EN_ATTENTE, StatutSignatureEnum.ENVOYE]
),
SignatureLog.date_envoi < date_limite,
SignatureLog.nb_relances < 3 # Max 3 relances
SignatureLog.nb_relances < 3, # Max 3 relances
)
result = await session.execute(query)
signatures_a_relancer = result.scalars().all()
nb_relances = 0
for signature in signatures_a_relancer:
try:
# Calculer jours écoulés
nb_jours = (datetime.now() - signature.date_envoi).days
jours_restants = 30 - nb_jours # Lien expire après 30 jours
if jours_restants <= 0:
# Transaction expirée
signature.statut = StatutSignatureEnum.EXPIRE
continue
# Préparer email de relance
template = templates_signature_email["relance_signature"]
type_labels = {
0: "Devis", 10: "Commande", 30: "Bon de Livraison",
60: "Facture", 50: "Avoir"
0: "Devis",
10: "Commande",
30: "Bon de Livraison",
60: "Facture",
50: "Avoir",
}
variables = {
"NOM_SIGNATAIRE": signature.nom_signataire,
"TYPE_DOC": type_labels.get(signature.type_document, "Document"),
@ -2643,16 +2648,16 @@ async def relancer_signatures_automatique(
"NB_JOURS": str(nb_jours),
"JOURS_RESTANTS": str(jours_restants),
"SIGNER_URL": signature.signer_url,
"CONTACT_EMAIL": settings.smtp_from
"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))
# Créer email de relance
email_log = EmailLog(
id=str(uuid.uuid4()),
@ -2665,35 +2670,37 @@ async def relancer_signatures_automatique(
date_creation=datetime.now(),
nb_tentatives=0,
)
session.add(email_log)
email_queue.enqueue(email_log.id)
# Incrémenter compteur de relances
signature.est_relance = True
signature.nb_relances = (signature.nb_relances or 0) + 1
nb_relances += 1
logger.info(f"📧 Relance envoyée: {signature.document_id} ({signature.nb_relances}/3)")
logger.info(
f"📧 Relance envoyée: {signature.document_id} ({signature.nb_relances}/3)"
)
except Exception as e:
logger.error(f"❌ Erreur relance signature {signature.id}: {e}")
continue
await session.commit()
return {
"success": True,
"signatures_verifiees": len(signatures_a_relancer),
"relances_envoyees": nb_relances,
"message": f"{nb_relances} email(s) de relance envoyé(s)"
"message": f"{nb_relances} email(s) de relance envoyé(s)",
}
except Exception as e:
logger.error(f"❌ Erreur relances automatiques: {e}")
raise HTTPException(500, str(e))
@app.get("/signature/universign/status", tags=["Signatures"])
async def statut_signature(docId: str = Query(...)):
@ -4166,9 +4173,7 @@ async def commande_vers_livraison(
):
try:
# Étape 1: Vérifier que la commande existe
commande_existante = sage_client.lire_document(
id, TypeDocumentSQL.BON_COMMANDE
)
commande_existante = sage_client.lire_document(id, TypeDocumentSQL.BON_COMMANDE)
if not commande_existante:
raise HTTPException(404, f"Commande {id} introuvable")
@ -4549,6 +4554,7 @@ async def statistiques_utilisateurs(session: AsyncSession = Depends(get_session)
logger.error(f"❌ Erreur stats utilisateurs: {e}")
raise HTTPException(500, str(e))
@app.get("/modeles", tags=["PDF Sage-Like"])
async def get_modeles_disponibles():
"""Liste tous les modèles PDF disponibles"""
@ -4564,8 +4570,10 @@ async def get_modeles_disponibles():
async def get_document_pdf(
numero: str,
type_doc: int = Query(..., description="0=devis, 60=facture, etc."),
modele: str = Query(None, description="Nom du modèle (ex: 'Facture client logo.bgc')"),
download: bool = Query(False, description="Télécharger au lieu d'afficher")
modele: str = Query(
None, description="Nom du modèle (ex: 'Facture client logo.bgc')"
),
download: bool = Query(False, description="Télécharger au lieu d'afficher"),
):
try:
# Récupérer le PDF (en bytes)
@ -4573,28 +4581,26 @@ async def get_document_pdf(
numero=numero,
type_doc=type_doc,
modele=modele,
base64_encode=False # On veut les bytes bruts
base64_encode=False, # On veut les bytes bruts
)
# Retourner le PDF
from fastapi.responses import Response
disposition = "attachment" if download else "inline"
filename = f"{numero}.pdf"
return Response(
content=pdf_bytes,
media_type="application/pdf",
headers={
"Content-Disposition": f'{disposition}; filename="{filename}"'
}
headers={"Content-Disposition": f'{disposition}; filename="{filename}"'},
)
except Exception as e:
logger.error(f"Erreur génération PDF: {e}")
raise HTTPException(500, str(e))
# =====================================================
# LANCEMENT
# =====================================================