#! /usr/bin/env python3.6 """ Python 3.6 or newer required. """ import stripe import json import os from flask import Flask, render_template, jsonify, request from dotenv import load_dotenv, find_dotenv import GlobalVariable as MYSY_GV from pymongo import ReturnDocument import prj_common as mycommon import inspect import sys, os import time from datetime import datetime # Setup Stripe python client library load_dotenv(find_dotenv()) # For sample support and debugging, not required for production: stripe.set_app_info( 'stripe-samples/subscription-use-cases/fixed-price', version='0.0.1', url='https://github.com/stripe-samples/subscription-use-cases/fixed-price') stripe.api_version = '2020-08-27' stripe.api_key = "sk_test_51LUUfAAbmaEugrFTrWsfcBWZtbBh9r3HCa2sgeyikG808LjSk3bAdFhV6KxgRZ3vFxooa6RE0c5zBkTuOUrKkyjy00BrsIXAPs" #static_dir = str(os.path.abspath(os.path.join(__file__, "..", os.getenv("STATIC_DIR")))) static_dir = "./" app = Flask(__name__, static_folder=static_dir, static_url_path="", template_folder=static_dir) @app.route('/', methods=['GET']) def get_index(): return render_template('register.html') @app.route('/config', methods=['GET']) def get_config(): # Retrieves two prices with the lookup_keys # `sample_basic` and `sample_premium`. To # create these prices, you can use the Stripe # CLI fixtures command with the supplied # `seed.json` fixture file like so: # # stripe fixtures seed.json # prices = stripe.Price.list( ) return jsonify( publishableKey="pk_test_51LUUfAAbmaEugrFTI25uZBD3IFjbtaL6jUfRV83diDf7nco8worna4NGKhMHbPP71WCwT5EHFRdDNatxPrJWwgZ300kgH5EO4p", prices=prices.data, ) """ Creation d'un client dans Stripe """ def create_customer(diction): # Reads application/json and returns a response #data = json.loads(request.data) try: field_list = ['name', 'email', 'country', 'city'] # recuperation des infos du partenaire cust_name = "" if ("name" in diction.keys()): if diction['name']: cust_name = diction['name'] cust_email = "" if ("email" in diction.keys()): if diction['email']: cust_email = diction['email'] cust_country = "" if ("country" in diction.keys()): if diction['country']: cust_country = diction['country'] cust_city = "" if ("city" in diction.keys()): if diction['city']: cust_city = diction['city'] # Create a new customer object #customer = stripe.Customer.create(email=data['email']) customer = stripe.Customer.create( email=cust_email, name=cust_name, address={ "city": cust_city, "country": cust_country, }, ) if( customer and customer.id ): mycommon.myprint( str(inspect.stack()[0][3]) + " - le partenaire "+str(cust_email)+" a été correctement créé dans Stripe. Son Id = "+str(customer.id)) return True, str(customer.id) else: mycommon.myprint( str(inspect.stack()[0][3]) + " WARNING - Impossble de créer le partenaire " + str( cust_email) + " Stripe.") return False, False except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, False """ Recuperation des mode de payement d'un customer """ def get_customer_payement_cards(diction): try: field_list = ['stripe_account_id'] # recuperation des infos du partenaire stripe_account_id = "" if ("stripe_account_id" in diction.keys()): if diction['stripe_account_id']: stripe_account_id = diction['stripe_account_id'] customer_payments = stripe.Customer.list_payment_methods( str(stripe_account_id), type="card", ) #print(" ### customer_payments = "+str(customer_payments.data)) RetObject = [] for val in customer_payments.data : my_card = {} my_card['brand'] = str(val.card.brand) my_card['exp_month'] = str(val.card.exp_month) my_card['exp_year'] = str(val.card.exp_year) my_card['last4'] = str(val.card.last4) my_card['createdatetime'] = str( datetime.fromtimestamp(int(str(val.created))) ) RetObject.append(mycommon.JSONEncoder().encode(my_card)) print(" ### RetObject = "+str( RetObject )) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, False """ Creation d'une carte de payement """ def create_update_payment_card(diction): try: field_list_obligatoire = ['number', 'exp_month', 'exp_year', 'cvc', 'customerid', 'stripe_paymentmethod_id'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans liste ") return False, " Impossible de créer el moyen de payement" number = "" if ("number" in diction.keys()): if diction['number']: number = diction['number'] exp_month = "" if ("exp_month" in diction.keys()): if diction['exp_month']: exp_month = diction['exp_month'] exp_year = "" if ("exp_year" in diction.keys()): if diction['exp_year']: exp_year = diction['exp_year'] cvc = "" if ("cvc" in diction.keys()): if diction['cvc']: cvc = diction['cvc'] customer_id = "" if ("customerid" in diction.keys()): if diction['customerid']: customer_id = diction['customerid'] stripe_paymentmethod_id = "" if ("stripe_paymentmethod_id" in diction.keys()): if diction['stripe_paymentmethod_id']: stripe_paymentmethod_id = diction['stripe_paymentmethod_id'] """ /!\ IMPORTANT : Si le champs "stripe_paymentmethod_id" est rempli alors c'est une mise à jour : a) on detache b) on créer la nouvelle carte c) on rattacjhe Si le champs "stripe_paymentmethod_id" est vide, alors c'est une creation + Attachement au client """ print(" ## stripe_paymentmethod_id existe. donc suppression de = "+str(stripe_paymentmethod_id)) if (len(stripe_paymentmethod_id) > 2): print(" ### UPDATE ==> detache") detache = stripe.PaymentMethod.detach( stripe_paymentmethod_id, ) print(" #### detache "+str(detache)) #print(" ### CREATE ") new_card = stripe.PaymentMethod.create( type="card", card={ "number": number, "exp_month": int(exp_month), "exp_year": int(exp_year), "cvc": cvc, }, ) #print(" new_card = "+str(new_card)) #print(" l'ID de la carte est : "+str(new_card.id)) attache = stripe.PaymentMethod.attach( str(new_card.id), customer=str(customer_id), ) #print(" attache = "+str(attache)) return True, new_card except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, False """ Cette fonction confirme un abonnement """ def confirm_suscription(diction): try: field_list_obligatoire = ['stripe_payment_id',] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans liste ") return False, " Impossible de créer el moyen de payement" stripe_payment_id = "" if ("stripe_payment_id" in diction.keys()): if diction['stripe_payment_id']: stripe_payment_id = diction['stripe_payment_id'] confirmation_payement = stripe.PaymentIntent.confirm( stripe_payment_id, payment_method="pm_card_visa", ) print("### confirmation_payement.data = "+str(confirmation_payement)) return True, confirmation_payement.id, confirmation_payement.status except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, False, False, def create_subscription(diction): # Simulating authenticated user. Lookup the logged in user in your # database, and set customer_id to the Stripe Customer ID of that user. customer_id = "" if ("customerid" in diction.keys()): if diction['customerid']: customer_id = diction['customerid'] # Extract the price ID from environment variables given the name # of the price passed from the front end. # # `price_id` is the an ID of a Price object on your account. # This was populated using Price's `lookup_key` in the /config endpoint price_id = "" if ("priceid" in diction.keys()): if diction['priceid']: price_id = diction['priceid'] try: # Create the subscription. Note we're using # expand here so that the API will return the Subscription's related # latest invoice, and that latest invoice's payment_intent # so we can collect payment information and confirm the payment on the front end. # Create the subscription subscription = stripe.Subscription.create( customer=customer_id, items=[{ 'price': price_id, }], payment_behavior='default_incomplete', expand=['latest_invoice.payment_intent'], ) print(" ### subscription = "+str(subscription)) return True, subscription.latest_invoice.payment_intent.id, subscription.latest_invoice.payment_intent.client_secret except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, False, False, @app.route('/cancel-subscription', methods=['POST']) def cancel_subscription(): data = json.loads(request.data) try: # Cancel the subscription by deleting it deletedSubscription = stripe.Subscription.delete(data['subscriptionId']) return jsonify(subscription=deletedSubscription) except Exception as e: return jsonify(error=str(e)), 403 @app.route('/subscriptions', methods=['GET']) def list_subscriptions(): # Simulating authenticated user. Lookup the logged in user in your # database, and set customer_id to the Stripe Customer ID of that user. customer_id = request.cookies.get('customer') try: # Cancel the subscription by deleting it subscriptions = stripe.Subscription.list( customer=customer_id, status='all', expand=['data.default_payment_method'] ) return jsonify(subscriptions=subscriptions) except Exception as e: return jsonify(error=str(e)), 403 @app.route('/invoice-preview', methods=['GET']) def preview_invoice(): # Simulating authenticated user. Lookup the logged in user in your # database, and set customer_id to the Stripe Customer ID of that user. customer_id = request.cookies.get('customer') subscription_id = request.args.get('subscriptionId') new_price_lookup_key = request.args.get('newPriceLookupKey') try: # Retrieve the subscription subscription = stripe.Subscription.retrieve(subscription_id) # Retrive the Invoice invoice = stripe.Invoice.upcoming( customer=customer_id, subscription=subscription_id, subscription_items=[{ 'id': subscription['items']['data'][0].id, 'price': os.getenv(new_price_lookup_key), }], ) return jsonify(invoice=invoice) except Exception as e: return jsonify(error=str(e)), 403 @app.route('/update-subscription', methods=['POST']) def update_subscription(): data = json.loads(request.data) try: subscription = stripe.Subscription.retrieve(data['subscriptionId']) update_subscription = stripe.Subscription.modify( data['subscriptionId'], items=[{ 'id': subscription['items']['data'][0].id, 'price': os.getenv(data['newPriceLookupKey'].upper()), }] ) return jsonify(update_subscription) except Exception as e: return jsonify(error=str(e)), 403 @app.route('/webhook', methods=['POST']) def webhook_received(): # You can use webhooks to receive information about asynchronous payment events. # For more about our webhook events check out https://stripe.com/docs/webhooks. webhook_secret = os.getenv('STRIPE_WEBHOOK_SECRET') request_data = json.loads(request.data) if webhook_secret: # Retrieve the event by verifying the signature using the raw body and secret if webhook signing is configured. signature = request.headers.get('stripe-signature') try: event = stripe.Webhook.construct_event( payload=request.data, sig_header=signature, secret=webhook_secret) data = event['data'] except Exception as e: return e event_type = event['type'] else: data = request_data['data'] event_type = request_data['type'] data_object = data['object'] if event_type == 'invoice.payment_succeeded': if data_object['billing_reason'] == 'subscription_create': # The subscription automatically activates after successful payment # Set the payment method used to pay the first invoice # as the default payment method for that subscription subscription_id = data_object['subscription'] payment_intent_id = data_object['payment_intent'] # Retrieve the payment intent used to pay the subscription payment_intent = stripe.PaymentIntent.retrieve(payment_intent_id) # Set the default payment method stripe.Subscription.modify( subscription_id, default_payment_method=payment_intent.payment_method ) print("Default payment method set for subscription:" + payment_intent.payment_method) elif event_type == 'invoice.payment_failed': # If the payment fails or the customer does not have a valid payment method, # an invoice.payment_failed event is sent, the subscription becomes past_due. # Use this webhook to notify your user that their payment has # failed and to retrieve new card details. # print(data) print('Invoice payment failed: %s', event.id) elif event_type == 'invoice.finalized': # If you want to manually send out invoices to your customers # or store them locally to reference to avoid hitting Stripe rate limits. # print(data) print('Invoice finalized: %s', event.id) elif event_type == 'customer.subscription.deleted': # handle subscription cancelled automatically based # upon your subscription settings. Or if the user cancels it. # print(data) print('Subscription canceled: %s', event.id) return jsonify({'status': 'success'})