import json
import logging
from datetime import UTC, datetime
import stripe
from django.conf import settings
from django.contrib import messages
from django.contrib.auth.decorators import login_required
from django.contrib.auth.mixins import LoginRequiredMixin
from django.core.mail import mail_admins
from django.http import (
HttpResponse,
HttpResponseBadRequest,
HttpResponseRedirect,
)
from django.shortcuts import redirect, render
from django.urls import reverse_lazy
from django.views import View
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from django.views.generic.edit import FormView
from main import forms, models, scheme
logger = logging.getLogger(__name__)
@login_required
def billing_overview(request):
"""
Renders the billing index page which includes a summary of subscription and
payment methods.
"""
# respond for grandfathered users first
if request.user.is_grandfathered:
return render(
request,
"main/billing_overview.html",
{
"is_grandfathered": True,
},
)
# respond for monero case
if request.user.monero_address:
return render(request, "main/billing_overview.html")
stripe.api_key = settings.STRIPE_API_KEY
# create stripe customer for user if it does not exist
if not request.user.stripe_customer_id:
try:
stripe_response = stripe.Customer.create()
except stripe.StripeError as ex:
logger.error(str(ex))
raise Exception("Failed to create customer on Stripe.") from ex
request.user.stripe_customer_id = stripe_response["id"]
request.user.save()
# get subscription if exists
current_period_start = None
current_period_end = None
subscription_status = None
if request.user.stripe_subscription_id:
subscription = _get_stripe_subscription(request.user.stripe_subscription_id)
if subscription:
subscription_status = subscription.get("status")
if subscription.get("cancel_at_period_end"):
subscription_status = "canceling"
# parse period fields when present (even if not active),
# so "Last payment" can be shown after scheduling cancellation
latest_invoice = subscription.get("latest_invoice") if subscription else None
if isinstance(latest_invoice, dict) and latest_invoice.get("status") == "paid":
items = (subscription or {}).get("items") or {}
item_data = items.get("data") or []
first_item = item_data[0] if item_data else {}
if current_period_start := first_item.get("current_period_start"):
current_period_start = datetime.fromtimestamp(
current_period_start, tz=UTC
)
if current_period_end := first_item.get("current_period_end"):
current_period_end = datetime.fromtimestamp(current_period_end, tz=UTC)
# transform into list of values
payment_methods = _get_payment_methods(request.user.stripe_customer_id).values()
return render(
request,
"main/billing_overview.html",
{
"stripe_customer_id": request.user.stripe_customer_id,
"stripe_public_key": settings.STRIPE_PUBLIC_KEY,
"stripe_price_id": settings.STRIPE_PRICE_ID,
"current_period_end": current_period_end,
"current_period_start": current_period_start,
"subscription_status": subscription_status,
"payment_methods": payment_methods,
"invoice_list": _get_invoices(request.user.stripe_customer_id),
},
)
def _get_stripe_subscription(stripe_subscription_id):
stripe.api_key = settings.STRIPE_API_KEY
try:
stripe_subscription = stripe.Subscription.retrieve(
stripe_subscription_id,
expand=["latest_invoice", "latest_invoice.payment_intent"],
)
except stripe.InvalidRequestError as ex:
logger.warning("Subscription %s not found: %s", stripe_subscription_id, str(ex))
return None
except stripe.StripeError as ex:
logger.error(
"Failed to get subscription %s from Stripe: %s",
stripe_subscription_id,
str(ex),
)
raise Exception("Failed to get subscription from Stripe.") from ex
return stripe_subscription
def _get_payment_methods(stripe_customer_id):
"""Get user's payment methods and transform them into a dictionary."""
stripe.api_key = settings.STRIPE_API_KEY
# get default payment method id
try:
default_pm_id = stripe.Customer.retrieve(
stripe_customer_id
).invoice_settings.default_payment_method
except stripe.StripeError as ex:
logger.error(str(ex))
raise Exception("Failed to retrieve customer data from Stripe.") from ex
# get payment methods
try:
stripe_payment_methods = stripe.PaymentMethod.list(
customer=stripe_customer_id,
type="card",
)
except stripe.StripeError as ex:
logger.error(str(ex))
raise Exception("Failed to retrieve payment methods from Stripe.") from ex
# normalise payment methods
payment_methods = {}
for stripe_pm in stripe_payment_methods.data:
payment_methods[stripe_pm.id] = {
"id": stripe_pm.id,
"brand": stripe_pm.card.brand,
"last4": stripe_pm.card.last4,
"exp_month": stripe_pm.card.exp_month,
"exp_year": stripe_pm.card.exp_year,
"is_default": False,
}
if stripe_pm.id == default_pm_id:
payment_methods[stripe_pm.id]["is_default"] = True
return payment_methods
def _get_invoices(stripe_customer_id):
"""Get user's invoices and transform them into a dictionary."""
stripe.api_key = settings.STRIPE_API_KEY
# get user invoices
try:
stripe_invoices = stripe.Invoice.list(customer=stripe_customer_id)
except stripe.StripeError as ex:
logger.error(str(ex))
raise Exception("Failed to retrieve invoices data from Stripe.") from ex
# normalize invoice objects
invoice_list = []
for stripe_inv in stripe_invoices.data:
invoice_list.append(
{
"id": stripe_inv.id,
"url": stripe_inv.hosted_invoice_url,
"pdf": stripe_inv.invoice_pdf,
"period_start": datetime.fromtimestamp(stripe_inv.period_start, tz=UTC),
"period_end": datetime.fromtimestamp(stripe_inv.period_end, tz=UTC),
"created": datetime.fromtimestamp(stripe_inv.created, tz=UTC),
}
)
return invoice_list
class BillingSubscribe(LoginRequiredMixin, FormView):
form_class = forms.StripeForm
template_name = "main/billing_subscribe.html"
success_url = reverse_lazy("billing_overview")
success_message = (
"payment is processing; premium will be enabled once the charge succeeds"
)
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context["stripe_public_key"] = settings.STRIPE_PUBLIC_KEY
return context
def get(self, request, *args, **kwargs):
stripe.api_key = settings.STRIPE_API_KEY
# ensure customer exists
if not request.user.stripe_customer_id:
try:
created = stripe.Customer.create()
request.user.stripe_customer_id = created.get("id")
request.user.save()
except stripe.StripeError as ex:
logger.error("Failed creating customer before subscribe: %s", str(ex))
messages.error(
request, "payment processor unavailable; please try again later"
)
return redirect("billing_overview")
url = f"{scheme.get_protocol()}//{settings.CANONICAL_HOST}"
url += reverse_lazy("billing_welcome")
if request.user.stripe_subscription_id:
stripe_subscription = _get_stripe_subscription(
request.user.stripe_subscription_id
)
# create new subscription if:
# * subscription is canceled but webhook was not received (yet)
# * stripe fails or returns None
if (
stripe_subscription.get("status") == "canceled"
or stripe_subscription is None
):
stripe_subscription = _create_stripe_subscription(
request.user.stripe_customer_id
)
else:
stripe_subscription = _create_stripe_subscription(
request.user.stripe_customer_id
)
request.user.stripe_subscription_id = stripe_subscription.get("id")
request.user.save()
payment_intents = stripe.PaymentIntent.list(
customer=request.user.stripe_customer_id, limit=1
)
client_secret = None
if payment_intents.data:
client_secret = payment_intents.data[0].client_secret
context = self.get_context_data()
context["stripe_client_secret"] = client_secret
context["stripe_return_url"] = url
return self.render_to_response(context)
def post(self, request, *args, **kwargs):
form_class = self.get_form_class()
form = self.get_form(form_class)
if form.is_valid():
messages.success(request, self.success_message)
return self.form_valid(form)
else:
return self.form_invalid(form)
def _create_stripe_subscription(customer_id):
stripe.api_key = settings.STRIPE_API_KEY
# expand subscription's latest invoice and invoice's payment_intent
# so we can pass it to the front end to confirm the payment
try:
stripe_subscription = stripe.Subscription.create(
customer=customer_id,
items=[
{
"price": settings.STRIPE_PRICE_ID,
}
],
payment_behavior="default_incomplete",
payment_settings={"save_default_payment_method": "on_subscription"},
)
logger.info(f"Created subscription: {stripe_subscription.get('id')}")
except stripe.StripeError as ex:
logger.error(str(ex))
raise Exception("Failed to create subscription on Stripe.") from ex
return stripe_subscription
class BillingCard(LoginRequiredMixin, FormView):
form_class = forms.StripeForm
template_name = "main/billing_card.html"
success_url = reverse_lazy("billing_overview")
success_message = "new card added"
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context["stripe_public_key"] = settings.STRIPE_PUBLIC_KEY
return context
def get(self, request, *args, **kwargs):
stripe.api_key = settings.STRIPE_API_KEY
context = self.get_context_data()
data = _create_setup_intent(request.user.stripe_customer_id)
context["stripe_client_secret"] = data["stripe_client_secret"]
url = f"{scheme.get_protocol()}//{settings.CANONICAL_HOST}"
url += reverse_lazy("billing_card_confirm")
context["stripe_return_url"] = url
return self.render_to_response(context)
def post(self, request, *args, **kwargs):
form_class = self.get_form_class()
form = self.get_form(form_class)
if form.is_valid():
messages.success(request, self.success_message)
return self.form_valid(form)
else:
return self.form_invalid(form)
def _create_setup_intent(customer_id):
stripe.api_key = settings.STRIPE_API_KEY
try:
stripe_setup_intent = stripe.SetupIntent.create(
automatic_payment_methods={"enabled": True},
customer=customer_id,
)
except stripe.StripeError as ex:
logger.error(str(ex))
raise Exception("Failed to create setup intent on Stripe.") from ex
return {
"stripe_client_secret": stripe_setup_intent["client_secret"],
}
class BillingCardDelete(LoginRequiredMixin, View):
"""View that deletes a card from a user on Stripe."""
template_name = "main/billing_card_confirm_delete.html"
success_url = reverse_lazy("billing_overview")
success_message = "card deleted"
slug_url_kwarg = "stripe_payment_method_id"
# dict of valid payment methods with id as key and an obj as val
stripe_payment_methods = {}
def get_context_data(self, **kwargs):
card_id = self.kwargs.get(self.slug_url_kwarg)
context = {
"card": self.stripe_payment_methods[card_id],
}
return context
def post(self, request, *args, **kwargs):
card_id = self.kwargs.get(self.slug_url_kwarg)
try:
stripe.PaymentMethod.detach(card_id)
except stripe.StripeError as ex:
logger.error(str(ex))
messages.error(request, "payment processor unresponsive; please try again")
return redirect(reverse_lazy("billing_overview"))
messages.success(request, self.success_message)
return HttpResponseRedirect(self.success_url)
def get(self, request, *args, **kwargs):
context = self.get_context_data(**kwargs)
return render(request, self.template_name, context)
def dispatch(self, request, *args, **kwargs):
if not request.user.stripe_customer_id:
mail_admins(
"User tried to delete card without stripe_customer_id",
f"user.id={request.user.id}\nuser.username={request.user.username}",
)
messages.error(
request,
"something has gone terribly wrong but we were just notified about it",
)
return redirect("dashboard")
self.stripe_payment_methods = _get_payment_methods(
request.user.stripe_customer_id
)
# check if card id is valid for user
card_id = self.kwargs.get(self.slug_url_kwarg)
if card_id not in self.stripe_payment_methods:
mail_admins(
"User tried to delete card with invalid Stripe card ID",
f"user.id={request.user.id}\nuser.username={request.user.username}",
)
messages.error(
request,
"this is not a valid card ID",
)
return redirect("dashboard")
return super().dispatch(request, *args, **kwargs)
@require_POST
@login_required
def billing_card_default(request, stripe_payment_method_id):
"""View method that changes the default card of a user on Stripe."""
stripe_payment_methods = _get_payment_methods(request.user.stripe_customer_id)
if stripe_payment_method_id not in stripe_payment_methods:
return HttpResponseBadRequest("Invalid Card ID.")
stripe.api_key = settings.STRIPE_API_KEY
try:
stripe.Customer.modify(
request.user.stripe_customer_id,
invoice_settings={
"default_payment_method": stripe_payment_method_id,
},
)
except stripe.StripeError as ex:
logger.error(str(ex))
return HttpResponse("Could not change default card.", status=503)
messages.success(request, "default card updated")
return redirect("billing_overview")
class BillingCancel(LoginRequiredMixin, View):
"""View that cancels a user subscription on Stripe."""
template_name = "main/billing_subscription_cancel.html"
success_url = reverse_lazy("billing_overview")
success_message = "subscription will be canceled at period end"
def post(self, request):
subscription = _get_stripe_subscription(request.user.stripe_subscription_id)
try:
# cancel at period end to keep access for the remainder of the paid period
stripe.Subscription.modify(subscription["id"], cancel_at_period_end=True)
except stripe.StripeError as ex:
logger.error(str(ex))
return HttpResponse("Subscription could not be canceled.", status=503)
mail_admins(
f"Cancellation premium subscriber: {request.user.username}",
f"{request.user.blog_absolute_url}\n",
)
messages.success(request, self.success_message)
return HttpResponseRedirect(self.success_url)
def get(self, request):
return render(request, self.template_name)
def dispatch(self, request, *args, **kwargs):
# redirect grandfathered users to dashboard
if request.user.is_grandfathered:
return redirect("dashboard")
# if user has no customer id, redirect to billing_overview to have it generated
if not request.user.stripe_customer_id:
return redirect("billing_overview")
# if user is not premium, redirect
if not request.user.is_premium:
return redirect("billing_overview")
subscription = _get_stripe_subscription(request.user.stripe_subscription_id)
if not subscription:
return redirect("billing_overview")
return super().dispatch(request, *args, **kwargs)
class BillingResume(LoginRequiredMixin, View):
"""View that resumes a canceled user subscription on Stripe."""
template_name = "main/billing_subscription_resume.html"
success_url = reverse_lazy("billing_overview")
success_message = "subscription resumed"
def post(self, request, *args, **kwargs):
subscription = _get_stripe_subscription(request.user.stripe_subscription_id)
try:
stripe.Subscription.modify(subscription["id"], cancel_at_period_end=False)
except stripe.StripeError as ex:
logger.error(str(ex))
return HttpResponse("Subscription could not be resumed.", status=503)
messages.success(request, self.success_message)
return HttpResponseRedirect(self.success_url)
def get(self, request, *args, **kwargs):
return render(request, self.template_name)
def dispatch(self, request, *args, **kwargs):
# redirect grandfathered users to dashboard
if request.user.is_grandfathered:
return redirect("dashboard")
# if user has no customer id, redirect to billing_overview
if not request.user.stripe_customer_id:
return redirect("billing_overview")
# if user is not premium, redirect
if not request.user.is_premium:
return redirect("billing_overview")
subscription = _get_stripe_subscription(request.user.stripe_subscription_id)
if not subscription:
return redirect("billing_overview")
# only allow resuming if subscription is set to cancel
if not subscription.get("cancel_at_period_end"):
return redirect("billing_overview")
return super().dispatch(request, *args, **kwargs)
class BillingResubscribe(LoginRequiredMixin, View):
"""
View that creates a new subscription for returning users with saved payment methods.
Charges immediately using the default payment method.
"""
template_name = "main/billing_resubscribe.html"
success_url = reverse_lazy("billing_overview")
success_message = "premium subscription enabled"
def get(self, request):
stripe.api_key = settings.STRIPE_API_KEY
payment_methods = _get_payment_methods(request.user.stripe_customer_id)
# make a card default if not already
has_default = any(pm.get("is_default") for pm in payment_methods.values())
if payment_methods and not has_default:
first_pm = next(iter(payment_methods.values()))
try:
stripe.Customer.modify(
request.user.stripe_customer_id,
invoice_settings={"default_payment_method": first_pm["id"]},
)
payment_methods[first_pm["id"]]["is_default"] = True
except Exception as e:
logger.error(f"Unable to set default payment method: {e}")
default_card = None
for pm in payment_methods.values():
if pm["is_default"]:
default_card = pm
break
context = {
"default_card": default_card,
}
return render(request, self.template_name, context)
def post(self, request):
stripe.api_key = settings.STRIPE_API_KEY
try:
# create subscription with immediate charge using saved payment method
stripe_subscription = stripe.Subscription.create(
customer=request.user.stripe_customer_id,
items=[
{
"price": settings.STRIPE_PRICE_ID,
}
],
payment_behavior="error_if_incomplete",
payment_settings={"save_default_payment_method": "on_subscription"},
expand=["latest_invoice.payment_intent"],
)
request.user.stripe_subscription_id = stripe_subscription.get("id")
request.user.save()
# check if payment succeeded immediately
latest_invoice = stripe_subscription.get("latest_invoice")
if latest_invoice:
payment_intent = latest_invoice.get("payment_intent")
if payment_intent and payment_intent.get("status") == "succeeded":
if not request.user.is_premium:
request.user.is_premium = True
request.user.is_approved = True
request.user.save()
if request.user.blog_absolute_url == request.user.blog_url:
blog_info = request.user.blog_absolute_url
else:
blog_info = f"{request.user.blog_absolute_url}\n\n{request.user.blog_url}"
mail_admins(
f"New premium resubscriber: {request.user.username}",
blog_info,
)
messages.success(request, self.success_message)
else:
messages.info(request, "payment is processing")
else:
messages.info(request, "payment processing")
except stripe.StripeError as ex:
logger.error("Failed to create resubscription: %s", str(ex))
messages.error(
request,
"failed to create subscription; please try again or contact support",
)
return redirect(self.success_url)
def dispatch(self, request, *args, **kwargs):
# redirect grandfathered users
if request.user.is_grandfathered:
return redirect("billing_overview")
# redirect if already premium
if request.user.is_premium:
return redirect("billing_overview")
# ensure customer exists
if not request.user.stripe_customer_id:
return redirect("billing_overview")
stripe.api_key = settings.STRIPE_API_KEY
# check if user has payment methods
payment_methods = _get_payment_methods(request.user.stripe_customer_id)
if not payment_methods:
# no saved payment methods, redirect to regular subscribe flow
return redirect("billing_subscribe")
return super().dispatch(request, *args, **kwargs)
@login_required
def billing_welcome(request):
"""
View that Stripe returns to if subscription initialisation is successful.
Adds a message alert and redirects to billing_overview.
"""
payment_intent = request.GET.get("payment_intent")
if not payment_intent:
return redirect("billing_overview")
stripe.api_key = settings.STRIPE_API_KEY
stripe_intent = stripe.PaymentIntent.retrieve(payment_intent)
if stripe_intent["status"] == "succeeded":
# charge succeeded during client-side confirmation flow
# enable premium if not already enabled via webhook
if not request.user.is_premium:
request.user.is_premium = True
request.user.is_approved = True
request.user.save()
if request.user.blog_absolute_url == request.user.blog_url:
blog_info = request.user.blog_absolute_url
else:
blog_info = (
f"{request.user.blog_absolute_url}\n\n{request.user.blog_url}"
)
mail_admins(
f"New premium subscriber from welcome page: {request.user.username}",
blog_info,
)
messages.success(request, "premium subscription enabled")
elif stripe_intent["status"] == "processing":
messages.info(request, "payment is currently processing")
else:
messages.error(
request,
"something is wrong. don't sweat it, worst case you get premium for free",
)
return redirect("billing_overview")
@login_required
def billing_card_confirm(request):
setup_intent = request.GET.get("setup_intent")
if not setup_intent:
return redirect("billing_overview")
stripe.api_key = settings.STRIPE_API_KEY
stripe_intent = stripe.SetupIntent.retrieve(setup_intent)
if stripe_intent["status"] == "succeeded":
messages.success(request, "payment method added")
elif stripe_intent["status"] == "processing":
messages.info(request, "payment method addition processing")
elif stripe_intent["status"] == "requires_payment_method":
messages.info(request, "error setting up payment method :(")
else:
messages.error(
request,
"something is wrong. don't sweat it, worst case you get premium for free",
)
return redirect("billing_overview")
@csrf_exempt
def billing_stripe_webhook(request):
"""
Handle Stripe webhooks.
See: https://stripe.com/docs/webhooks
"""
# ensure only POST requests are allowed
if request.method != "POST":
return HttpResponse(status=405)
# get Stripe settings
stripe.api_key = settings.STRIPE_API_KEY
webhook_secret = getattr(settings, "STRIPE_WEBHOOK_SECRET", "")
try:
# parse event from Stripe
if webhook_secret:
# verify webhook signature
sig_header = request.headers.get("Stripe-Signature", "")
event = stripe.Webhook.construct_event(
payload=request.body,
sig_header=sig_header,
secret=webhook_secret,
)
else:
# Development mode: skip signature verification
data = json.loads(request.body.decode("utf-8"))
event = stripe.Event.construct_from(data, stripe.api_key)
except (ValueError, stripe.SignatureVerificationError) as ex:
# invalid payload or signature
logger.error(f"Webhook validation failed: {type(ex).__name__}: {str(ex)}")
return HttpResponse(status=400)
except Exception as ex:
logger.error(f"Webhook processing error: {str(ex)}")
return HttpResponse(status=500)
# process webhook event types
try:
if event.type == "invoice.payment_succeeded":
invoice = event.data.object
customer_id = getattr(invoice, "customer", None)
if customer_id:
customer_id_str = (
customer_id.id if hasattr(customer_id, "id") else str(customer_id)
)
try:
user = models.User.objects.get(stripe_customer_id=customer_id_str)
if not user.is_premium:
user.is_premium = True
user.is_approved = True
user.save()
if user.blog_absolute_url == user.blog_url:
blog_info = user.blog_absolute_url
else:
blog_info = f"{user.blog_absolute_url}\n\n{user.blog_url}"
mail_admins(
f"New premium subscriber from webhook: {user.username}",
blog_info,
)
except models.User.DoesNotExist:
logger.warning(
f"Webhook: user not found for customer_id={customer_id_str}"
)
elif event.type == "customer.subscription.deleted":
subscription = event.data.object
customer_id = getattr(subscription, "customer", None)
if customer_id:
customer_id_str = (
customer_id.id if hasattr(customer_id, "id") else str(customer_id)
)
try:
user = models.User.objects.get(stripe_customer_id=customer_id_str)
if user.is_premium:
user.is_premium = False
user.stripe_subscription_id = None
user.save()
except models.User.DoesNotExist:
logger.warning(
f"Webhook: user not found for customer_id={customer_id_str}"
)
elif event.type == "payment_method.attached":
payment_method = event.data.object
customer_id = getattr(payment_method, "customer", None)
if customer_id:
customer_id_str = (
customer_id.id if hasattr(customer_id, "id") else str(customer_id)
)
try:
# set payment method as default if customer has no default one yet
customer = stripe.Customer.retrieve(customer_id_str)
if not customer.invoice_settings.default_payment_method:
stripe.Customer.modify(
customer_id_str,
invoice_settings={
"default_payment_method": payment_method.id,
},
)
logger.info(
f"Set payment method {payment_method.id} as default for customer {customer_id_str}"
)
except stripe.StripeError as ex:
logger.error(
f"Failed to set default payment method for customer {customer_id_str}: {str(ex)}"
)
except Exception as ex:
logger.error(f"Webhook event processing error: {str(ex)}")
return HttpResponse(status=200)