""" Service Stripe pour gérer les paiements et abonnements """ import stripe import logging from typing import Optional, Dict, Any from datetime import datetime, timedelta from .config import StripeConfig logger = logging.getLogger(__name__) class StripeService: """Service de gestion des paiements Stripe""" def __init__(self): """Initialise le service Stripe""" if not StripeConfig.is_configured(): logger.warning("Stripe n'est pas configuré. Définissez STRIPE_SECRET_KEY dans les variables d'environnement.") return stripe.api_key = StripeConfig.STRIPE_SECRET_KEY logger.info("Service Stripe initialisé") def create_checkout_session( self, lookup_key: str, customer_email: str, success_url: str, cancel_url: str, metadata: Optional[Dict[str, str]] = None ) -> Dict[str, Any]: """ Crée une session Checkout Stripe Args: lookup_key: Clé de recherche du prix (ex: web_sentinel_pro_monthly) customer_email: Email du client success_url: URL de redirection en cas de succès cancel_url: URL de redirection en cas d'annulation metadata: Métadonnées additionnelles Returns: Dict contenant l'ID de session et l'URL de paiement """ try: # Récupérer le plan et le Price ID depuis la configuration plan = StripeConfig.get_plan_by_lookup_key(lookup_key) # Déterminer le Price ID selon la périodicité if plan.lookup_key_monthly == lookup_key: price_id = plan.price_monthly else: price_id = plan.price_yearly # Préparer les métadonnées session_metadata = metadata or {} session_metadata.update({ 'lookup_key': lookup_key, 'plan': plan.tier }) # Créer la session Checkout session = stripe.checkout.Session.create( customer_email=customer_email, line_items=[{ 'price': price_id, 'quantity': 1, }], mode='subscription', success_url=success_url + '?session_id={CHECKOUT_SESSION_ID}', cancel_url=cancel_url, metadata=session_metadata, subscription_data={ 'metadata': session_metadata, }, # Options additionnelles allow_promotion_codes=True, billing_address_collection='required', payment_method_types=['card'], locale='fr' ) logger.info(f"Session Checkout créée: {session.id} pour {customer_email}") return { 'session_id': session.id, 'url': session.url, 'customer_email': customer_email, 'plan': session_metadata['plan'] } except stripe.error.StripeError as e: logger.error(f"Erreur Stripe lors de la création de session: {e}") raise except Exception as e: logger.error(f"Erreur lors de la création de session Checkout: {e}") raise def create_customer_portal_session( self, customer_id: str, return_url: str ) -> Dict[str, str]: """ Crée une session de portail client Args: customer_id: ID du client Stripe return_url: URL de retour après gestion Returns: Dict contenant l'URL du portail """ try: session = stripe.billing_portal.Session.create( customer=customer_id, return_url=return_url, ) logger.info(f"Session portail créée pour le client: {customer_id}") return { 'url': session.url } except stripe.error.StripeError as e: logger.error(f"Erreur Stripe lors de la création du portail: {e}") raise def get_checkout_session(self, session_id: str) -> Dict[str, Any]: """ Récupère les détails d'une session Checkout Args: session_id: ID de la session Checkout Returns: Dict contenant les détails de la session """ try: session = stripe.checkout.Session.retrieve( session_id, expand=['customer', 'subscription'] ) return { 'id': session.id, 'customer_id': session.customer, 'customer_email': session.customer_email if hasattr(session, 'customer_email') else None, 'subscription_id': session.subscription, 'payment_status': session.payment_status, 'status': session.status, 'metadata': session.metadata } except stripe.error.StripeError as e: logger.error(f"Erreur Stripe lors de la récupération de session: {e}") raise def get_subscription(self, subscription_id: str) -> Dict[str, Any]: """ Récupère les détails d'un abonnement Args: subscription_id: ID de l'abonnement Returns: Dict contenant les détails de l'abonnement """ try: subscription = stripe.Subscription.retrieve(subscription_id) return { 'id': subscription.id, 'customer_id': subscription.customer, 'status': subscription.status, 'current_period_start': datetime.fromtimestamp(subscription.current_period_start), 'current_period_end': datetime.fromtimestamp(subscription.current_period_end), 'cancel_at_period_end': subscription.cancel_at_period_end, 'metadata': subscription.metadata } except stripe.error.StripeError as e: logger.error(f"Erreur Stripe lors de la récupération de l'abonnement: {e}") raise def verify_webhook_signature(self, payload: bytes, sig_header: str) -> Dict[str, Any]: """ Vérifie la signature d'un webhook Stripe Args: payload: Corps de la requête webhook sig_header: Header Stripe-Signature Returns: Event Stripe vérifié """ try: event = stripe.Webhook.construct_event( payload, sig_header, StripeConfig.STRIPE_WEBHOOK_SECRET ) return event except ValueError as e: logger.error(f"Payload webhook invalide: {e}") raise except stripe.error.SignatureVerificationError as e: logger.error(f"Signature webhook invalide: {e}") raise def cancel_subscription(self, subscription_id: str, at_period_end: bool = True) -> Dict[str, Any]: """ Annule un abonnement Args: subscription_id: ID de l'abonnement at_period_end: Si True, annule à la fin de la période de facturation Returns: Dict contenant les détails de l'abonnement annulé """ try: if at_period_end: subscription = stripe.Subscription.modify( subscription_id, cancel_at_period_end=True ) else: subscription = stripe.Subscription.delete(subscription_id) logger.info(f"Abonnement annulé: {subscription_id}") return { 'id': subscription.id, 'status': subscription.status, 'cancel_at_period_end': subscription.cancel_at_period_end } except stripe.error.StripeError as e: logger.error(f"Erreur lors de l'annulation de l'abonnement: {e}") raise