sepehr/office_translator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add Telegram notifications for user signup and Stripe events
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 2m34s
Details

Browse Source
This commit is contained in:
sepehr
2026-06-07 11:37:01 +02:00
parent 8659b6761f
commit 29753881a6
3 changed files with 122 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
services/payment_service.py
View File

@@ -336,6 +336,21 @@ async def handle_checkout_completed(session: Dict):
)
except Exception as e:
logger.error("Failed to send credit purchase email: %s", e)
# Send Telegram notification
try:
from utils.telegram import send_telegram_notification
import asyncio
msg = (
f"💰 *Achat de crédits !*\n\n"
f"• *Utilisateur* : `{user.name or 'Non renseigné'}` (`{user.email}`)\n"
f"• *Crédits achetés* : `{credits}`\n"
f"• *Montant* : `{session.get('amount_total', 0) / 100:.2f} {session.get('currency', 'eur').upper()}`\n"
f"• *ID Transaction* : `{session_id}`"
)
asyncio.create_task(send_telegram_notification(msg))
except Exception as tel_err:
logger.error(f"Failed to send telegram notification for credits checkout: {tel_err}")
return
# It's a subscription
@@ -420,6 +435,22 @@ async def handle_checkout_completed(session: Dict):
except Exception as e:
logger.error("Failed to send activation email: %s", e)
# Send Telegram notification
try:
from utils.telegram import send_telegram_notification
import asyncio
ends_str = subscription_ends_at.strftime("%d/%m/%Y") if subscription_ends_at else "Non spécifiée"
msg = (
f"⭐ *Nouvel Abonnement Activé !*\n\n"
f"• *Utilisateur* : `{user.name or 'Non renseigné'}` (`{user.email}`)\n"
f"• *Forfait* : `{plan.upper()}`\n"
f"• *Montant* : `{session.get('amount_total', 0) / 100:.2f} {session.get('currency', 'eur').upper()}`\n"
f"• *Date de fin* : {ends_str}"
)
asyncio.create_task(send_telegram_notification(msg))
except Exception as tel_err:
logger.error(f"Failed to send telegram notification for subscription activation: {tel_err}")
async def handle_subscription_updated(subscription: Dict):
"""Handle subscription updates"""
@@ -474,6 +505,19 @@ async def handle_subscription_updated(subscription: Dict):
except Exception as e:
logger.error("Failed to send cancellation email in handle_subscription_updated: %s", e)
# Send Telegram notification for cancellation request
try:
from utils.telegram import send_telegram_notification
import asyncio
msg = (
f"🔕 *Désinscription programmée (Stripe)*\n\n"
f"• *Utilisateur* : `{user.name or 'Non renseigné'}` (`{user.email}`)\n"
f"• *Date de fin d'accès* : {ends_str}"
)
asyncio.create_task(send_telegram_notification(msg))
except Exception as tel_err:
logger.error(f"Failed to send telegram notification for subscription cancellation request: {tel_err}")
async def handle_subscription_deleted(subscription: Dict):
"""Handle subscription cancellation (actually ended)"""
@@ -509,6 +553,19 @@ async def handle_subscription_deleted(subscription: Dict):
except Exception as e:
logger.error("Failed to send subscription ended email: %s", e)
# Send Telegram notification for ended subscription
try:
from utils.telegram import send_telegram_notification
import asyncio
msg = (
f"❌ *Abonnement Terminé / Expiré (Stripe)*\n\n"
f"• *Utilisateur* : `{user.name or 'Non renseigné'}` (`{user.email}`)\n"
f"• *Statut* : Retour au forfait `FREE`"
)
asyncio.create_task(send_telegram_notification(msg))
except Exception as tel_err:
logger.error(f"Failed to send telegram notification for subscription ended: {tel_err}")
async def handle_payment_failed(invoice: Dict):
"""Handle failed payment — set subscription status to PAST_DUE and send email"""
@@ -547,6 +604,19 @@ async def handle_payment_failed(invoice: Dict):
)
except Exception as e:
logger.error("Failed to send payment failed email: %s", e)
# Send Telegram notification
try:
from utils.telegram import send_telegram_notification
import asyncio
msg = (
f"⚠️ *Échec de paiement (Stripe)*\n\n"
f"• *Utilisateur* : `{db_user.name or 'Non renseigné'}` (`{db_user.email}`)\n"
f"• *Statut* : Facture impayée, forfait passé en `PAST_DUE`"
)
asyncio.create_task(send_telegram_notification(msg))
except Exception as tel_err:
logger.error(f"Failed to send telegram notification for payment failed: {tel_err}")
except Exception as exc:
logger.error("handle_payment_failed DB error: %s", exc)