""" Endpoints API Flask pour la gestion des paiements Stripe """ from flask import Blueprint, request, jsonify, redirect import logging from typing import Dict, Any from web_sentinel.payment import StripeService, StripeConfig from .license_store_factory import build_license_store from .email_service import EmailService logger = logging.getLogger(__name__) # Blueprint pour les routes de paiement payment_bp = Blueprint('payment', __name__, url_prefix='/api/v1/payment') # Instances des services stripe_service = StripeService() license_store = build_license_store() email_service = EmailService() @payment_bp.route('/config', methods=['GET']) def get_payment_config(): """ Retourne la configuration publique de Stripe (clé publique, plans) GET /api/v1/payment/config """ try: plans_info = {} for key, plan in StripeConfig.PLANS.items(): plans_info[key] = { 'name': plan.name, 'tier': plan.tier, 'features': plan.features, 'sast_limit': plan.sast_limit, 'lookup_key_monthly': plan.lookup_key_monthly, 'lookup_key_yearly': plan.lookup_key_yearly } return jsonify({ 'success': True, 'data': { 'publishable_key': StripeConfig.STRIPE_PUBLISHABLE_KEY, 'plans': plans_info, 'configured': StripeConfig.is_configured() } }) except Exception as e: logger.error(f"Erreur lors de la récupération de la config: {e}") return jsonify({ 'success': False, 'error': str(e) }), 500 @payment_bp.route('/create-checkout-session', methods=['POST']) def create_checkout_session(): """ Crée une session Checkout Stripe POST /api/v1/payment/create-checkout-session Body: { "lookup_key": "web_sentinel_pro_monthly", "customer_email": "user@example.com", "success_url": "https://example.com/success", "cancel_url": "https://example.com/cancel", "metadata": {"user_id": "123"} // Optionnel } """ try: data = request.get_json() # Validation des données required_fields = ['lookup_key', 'customer_email'] for field in required_fields: if field not in data: return jsonify({ 'success': False, 'error': f'Champ requis manquant: {field}' }), 400 # URLs par défaut si non fournies success_url = data.get('success_url', f'{StripeConfig.BASE_URL}/success') cancel_url = data.get('cancel_url', f'{StripeConfig.BASE_URL}/cancel') # Créer la session session_data = stripe_service.create_checkout_session( lookup_key=data['lookup_key'], customer_email=data['customer_email'], success_url=success_url, cancel_url=cancel_url, metadata=data.get('metadata') ) return jsonify({ 'success': True, 'data': session_data }) except ValueError as e: logger.error(f"Validation error: {e}") return jsonify({ 'success': False, 'error': str(e) }), 400 except Exception as e: logger.error(f"Erreur lors de la création de session Checkout: {e}") return jsonify({ 'success': False, 'error': 'Erreur lors de la création de la session de paiement' }), 500 @payment_bp.route('/create-portal-session', methods=['POST']) def create_portal_session(): """ Crée une session de portail client Stripe POST /api/v1/payment/create-portal-session Body: { "customer_id": "cus_xxxxx", "return_url": "https://example.com/account" } """ try: data = request.get_json() # Validation if 'customer_id' not in data: return jsonify({ 'success': False, 'error': 'customer_id requis' }), 400 return_url = data.get('return_url', f'{StripeConfig.BASE_URL}/account') # Créer la session portail portal_data = stripe_service.create_customer_portal_session( customer_id=data['customer_id'], return_url=return_url ) return jsonify({ 'success': True, 'data': portal_data }) except Exception as e: logger.error(f"Erreur lors de la création du portail: {e}") return jsonify({ 'success': False, 'error': 'Erreur lors de la création du portail client' }), 500 @payment_bp.route('/session/', methods=['GET']) def get_session(session_id: str): """ Récupère les détails d'une session Checkout GET /api/v1/payment/session/ """ try: session_data = stripe_service.get_checkout_session(session_id) return jsonify({ 'success': True, 'data': session_data }) except Exception as e: logger.error(f"Erreur lors de la récupération de session: {e}") return jsonify({ 'success': False, 'error': 'Session non trouvée' }), 404 @payment_bp.route('/subscription/', methods=['GET']) def get_subscription(subscription_id: str): """ Récupère les détails d'un abonnement GET /api/v1/payment/subscription/ """ try: subscription_data = stripe_service.get_subscription(subscription_id) return jsonify({ 'success': True, 'data': subscription_data }) except Exception as e: logger.error(f"Erreur lors de la récupération de l'abonnement: {e}") return jsonify({ 'success': False, 'error': 'Abonnement non trouvé' }), 404 @payment_bp.route('/subscription//cancel', methods=['POST']) def cancel_subscription(subscription_id: str): """ Annule un abonnement POST /api/v1/payment/subscription//cancel Body: { "at_period_end": true // Optionnel, défaut: true } """ try: data = request.get_json() or {} at_period_end = data.get('at_period_end', True) result = stripe_service.cancel_subscription( subscription_id=subscription_id, at_period_end=at_period_end ) return jsonify({ 'success': True, 'data': result }) except Exception as e: logger.error(f"Erreur lors de l'annulation de l'abonnement: {e}") return jsonify({ 'success': False, 'error': 'Erreur lors de l\'annulation' }), 500 @payment_bp.route('/license/validate', methods=['POST']) def validate_license(): """ Valide une clé API et retourne les détails de la licence POST /api/v1/payment/license/validate Body: { "api_key": "ws_live_xxxxx" } Headers: Authorization: Bearer ws_live_xxxxx (alternative) """ try: # Récupérer la clé API depuis le body ou le header data = request.get_json() or {} api_key = data.get('api_key') # Fallback : vérifier le header Authorization if not api_key: auth_header = request.headers.get('Authorization', '') if auth_header.startswith('Bearer '): api_key = auth_header[7:] if not api_key: return jsonify({ 'valid': False, 'error': 'API key requise' }), 400 # Valider la licence validation_result = license_store.validate_license(api_key) if validation_result['valid']: return jsonify({ 'success': True, 'data': validation_result }), 200 else: return jsonify({ 'success': False, 'data': validation_result }), 403 except Exception as e: logger.error(f"Erreur lors de la validation de licence: {e}") return jsonify({ 'success': False, 'error': 'Erreur lors de la validation' }), 500 @payment_bp.route('/license/info', methods=['GET']) def get_license_info(): """ Récupère les informations d'une licence via API key GET /api/v1/payment/license/info Headers: Authorization: Bearer ws_live_xxxxx """ try: # Récupérer la clé API depuis le header auth_header = request.headers.get('Authorization', '') if not auth_header.startswith('Bearer '): return jsonify({ 'success': False, 'error': 'Authorization header requis (Bearer token)' }), 401 api_key = auth_header[7:] # Récupérer la licence license_data = license_store.get_license_by_api_key(api_key) if not license_data: return jsonify({ 'success': False, 'error': 'Licence non trouvée' }), 404 # Retourner les infos (sans l'API key complète pour sécurité) return jsonify({ 'success': True, 'data': { 'email': license_data['email'], 'tier': license_data['tier'], 'status': license_data['status'], 'max_domains': license_data['max_domains'], 'max_users': license_data['max_users'], 'allow_source_scan': license_data.get('allow_source_scan'), 'max_source_files': license_data.get('max_source_files'), 'max_source_size_mb': license_data.get('max_source_size_mb'), 'advanced_rules': license_data.get('advanced_rules'), 'created_at': license_data['created_at'], 'expires_at': license_data['expires_at'], 'usage_count': license_data['usage_count'], 'last_used_at': license_data['last_used_at'] } }) except Exception as e: logger.error(f"Erreur lors de la récupération de licence: {e}") return jsonify({ 'success': False, 'error': 'Erreur lors de la récupération' }), 500 @payment_bp.route('/webhook', methods=['POST']) def stripe_webhook(): """ Endpoint pour recevoir les webhooks Stripe POST /api/v1/payment/webhook Gère les événements: - checkout.session.completed: Paiement réussi - customer.subscription.updated: Abonnement mis à jour - customer.subscription.deleted: Abonnement annulé - invoice.payment_succeeded: Renouvellement réussi - invoice.payment_failed: Échec de paiement """ payload = request.data sig_header = request.headers.get('Stripe-Signature') try: # Vérifier la signature event = stripe_service.verify_webhook_signature(payload, sig_header) logger.info(f"Webhook reçu: {event['type']}") # Traiter l'événement event_type = event['type'] event_data = event['data']['object'] if event_type == 'checkout.session.completed': # Nouveau paiement/abonnement handle_checkout_completed(event_data) elif event_type == 'customer.subscription.updated': # Abonnement mis à jour handle_subscription_updated(event_data) elif event_type == 'customer.subscription.deleted': # Abonnement annulé handle_subscription_deleted(event_data) elif event_type == 'invoice.payment_succeeded': # Paiement réussi (renouvellement) handle_payment_succeeded(event_data) elif event_type == 'invoice.payment_failed': # Échec de paiement handle_payment_failed(event_data) return jsonify({'success': True}), 200 except Exception as e: logger.error(f"Erreur lors du traitement du webhook: {e}") return jsonify({'success': False, 'error': str(e)}), 400 def handle_checkout_completed(session): """Traite l'événement checkout.session.completed""" logger.info(f"Checkout complété pour session: {session['id']}") try: # Récupérer les détails de la session customer_email = session.get('customer_details', {}).get('email') customer_id = session.get('customer') subscription_id = session.get('subscription') if not customer_email: logger.error("Email client manquant dans la session") return # Extraire le tier depuis les metadata (vérifier 'tier' ou 'plan') metadata = session.get('metadata', {}) tier = metadata.get('tier') or metadata.get('plan', 'STARTER') tier = tier.upper() # Définir les limites selon le tier limits = { 'FREE': {'max_domains': 5, 'max_users': 0}, 'STARTER': {'max_domains': -1, 'max_users': 1}, 'PRO': {'max_domains': -1, 'max_users': 1}, 'ENTERPRISE': {'max_domains': -1, 'max_users': 10} } tier_limits = limits.get(tier, limits['STARTER']) # Créer la licence dans la base de données license_data = license_store.create_license( email=customer_email, tier=tier, stripe_customer_id=customer_id, stripe_subscription_id=subscription_id, max_domains=tier_limits['max_domains'], max_users=tier_limits['max_users'] ) logger.info(f"✅ Licence créée: {license_data['api_key']} pour {customer_email}") # Envoyer l'email de bienvenue avec la clé API customer_name = session.get('customer_details', {}).get('name') email_sent = email_service.send_license_email( to_email=customer_email, api_key=license_data['api_key'], tier=tier, customer_name=customer_name ) if email_sent: logger.info(f"✅ Email de licence envoyé à {customer_email}") else: logger.warning(f"⚠️ Email de licence non envoyé à {customer_email}") except Exception as e: logger.error(f"❌ Erreur lors du traitement du checkout: {e}", exc_info=True) def handle_subscription_updated(subscription): """Traite l'événement customer.subscription.updated""" logger.info(f"Abonnement mis à jour: {subscription['id']}") try: subscription_id = subscription['id'] status = subscription['status'] # Récupérer la licence correspondante license_data = license_store.get_license_by_stripe_subscription(subscription_id) if not license_data: logger.warning(f"Aucune licence trouvée pour subscription {subscription_id}") return # Mettre à jour le statut de la licence selon le statut Stripe license_status_map = { 'active': 'active', 'past_due': 'suspended', 'canceled': 'cancelled', 'unpaid': 'suspended' } new_status = license_status_map.get(status, 'suspended') license_store.update_license_status( api_key=license_data['api_key'], status=new_status, event_type='subscription_updated', event_data=f"Stripe status: {status}" ) logger.info(f"✅ Licence {license_data['api_key']} mise à jour: {new_status}") except Exception as e: logger.error(f"❌ Erreur lors de la mise à jour de l'abonnement: {e}", exc_info=True) def handle_subscription_deleted(subscription): """Traite l'événement customer.subscription.deleted""" logger.info(f"Abonnement supprimé: {subscription['id']}") try: subscription_id = subscription['id'] # Récupérer la licence correspondante license_data = license_store.get_license_by_stripe_subscription(subscription_id) if not license_data: logger.warning(f"Aucune licence trouvée pour subscription {subscription_id}") return # Désactiver la licence license_store.update_license_status( api_key=license_data['api_key'], status='cancelled', event_type='subscription_deleted', event_data=f"Stripe subscription deleted: {subscription_id}" ) # Envoyer email de confirmation email_service.send_subscription_cancelled_email( to_email=license_data['email'], tier=license_data['tier'] ) logger.info(f"✅ Licence {license_data['api_key']} annulée") except Exception as e: logger.error(f"❌ Erreur lors de la suppression de l'abonnement: {e}", exc_info=True) def handle_payment_succeeded(invoice): """Traite l'événement invoice.payment_succeeded""" logger.info(f"Paiement réussi pour facture: {invoice['id']}") try: subscription_id = invoice.get('subscription') if not subscription_id: return # Récupérer la licence license_data = license_store.get_license_by_stripe_subscription(subscription_id) if not license_data: logger.warning(f"Aucune licence trouvée pour subscription {subscription_id}") return # S'assurer que la licence est active if license_data['status'] != 'active': license_store.update_license_status( api_key=license_data['api_key'], status='active', event_type='payment_succeeded', event_data=f"Invoice: {invoice['id']}" ) logger.info(f"✅ Licence {license_data['api_key']} réactivée après paiement") except Exception as e: logger.error(f"❌ Erreur lors du traitement du paiement réussi: {e}", exc_info=True) def handle_payment_failed(invoice): """Traite l'événement invoice.payment_failed""" logger.error(f"Échec de paiement pour facture: {invoice['id']}") try: subscription_id = invoice.get('subscription') if not subscription_id: return # Récupérer la licence license_data = license_store.get_license_by_stripe_subscription(subscription_id) if not license_data: logger.warning(f"Aucune licence trouvée pour subscription {subscription_id}") return # Suspendre la licence license_store.update_license_status( api_key=license_data['api_key'], status='suspended', event_type='payment_failed', event_data=f"Invoice: {invoice['id']}" ) logger.warning(f"⚠️ Licence {license_data['api_key']} suspendue suite à l'échec de paiement") # TODO: Envoyer email d'alerte au client except Exception as e: logger.error(f"❌ Erreur lors du traitement de l'échec de paiement: {e}", exc_info=True)