from decimal import Decimal from typing import Literal, Optional from dataclasses import dataclass from django.utils.translation import gettext_lazy as _ from rest_framework.request import Request from labs.billing.providers.stripe import StripeProvider from labs.billing.exceptions import ( InvalidAmountError, CreateOrderError, CheckoutOrderError, ) from labs.billing.consts import ( BUY_CREDITS_MIN_QUANTITY, BUY_CREDITS_MAX_QUANTITY, BILLING_ALLOWED_COUNTRIES, ) from labs.models import BillingCreditOrder, User, BillingUserProfile from labs.user_events import UserEvent, log_user_event NOT_ALLOWED_COUNTRIES_ERROR_MESSAGE = _( "User is not allowed to buy credits from this country, " "allowed countries: %(allowed_countries)s" ) % { "allowed_countries": ", ".join(BILLING_ALLOWED_COUNTRIES), } @dataclass class CalculatePriceWhenBuyCreditsResult: currency: Literal["usd"] discount: Decimal amount_total: Decimal product_amount: Decimal tax_amount: Decimal def to_frontend(self) -> dict: return { "currency": self.currency, "discount": self.discount, "amount_total": str(self.amount_total), "product_amount": str(self.product_amount), "tax_amount": str(self.tax_amount) if self.tax_amount else None, } def get_buy_credits_discount(quantity: int) -> Decimal: if quantity >= 40: return Decimal("0.1") return Decimal("0") def calculate_price_when_buy_credits( quantity: int, include_tax: bool, country: Optional[str] = None ) -> CalculatePriceWhenBuyCreditsResult: """ Calculate the price when buying credits. Args: quantity: int, the amount of credits user want to buy include_tax: bool, whether to include tax in the price country: str, the country of the user, used for tax calculation Returns: CalculatePriceWhenBuyCreditsResult """ if not quantity or not isinstance(quantity, int): raise InvalidAmountError(f"Invalid amount ({quantity}), must be an integer") if quantity < BUY_CREDITS_MIN_QUANTITY or quantity > BUY_CREDITS_MAX_QUANTITY: raise InvalidAmountError( f"Invalid amount ({quantity}), must be between " f"{BUY_CREDITS_MIN_QUANTITY} and {BUY_CREDITS_MAX_QUANTITY}" ) tax_amount = Decimal("0") discount = get_buy_credits_discount(quantity) # 1 credits = 1 usd product_amount = (Decimal(quantity) * (Decimal("1") - discount)).quantize( Decimal("0.01") ) if include_tax: assert country in BILLING_ALLOWED_COUNTRIES, NOT_ALLOWED_COUNTRIES_ERROR_MESSAGE if country == "JP": tax_amount = product_amount * Decimal("0.1") elif country == "HK": tax_amount = product_amount * Decimal("0") elif country == "SG": tax_amount = product_amount * Decimal("0.09") elif country == "AE": tax_amount = product_amount * Decimal("0.05") elif country == "US": # TODO: currently we only support California for US users # We need pass full billing address to this method tax_amount = product_amount * Decimal("0") amount_total = product_amount + tax_amount return CalculatePriceWhenBuyCreditsResult( currency="usd", amount_total=amount_total, product_amount=product_amount, tax_amount=tax_amount, discount=discount, ) def create_credit_order(user: User, quantity: int) -> BillingCreditOrder: try: profile = BillingUserProfile.objects.get(user=user) except BillingUserProfile.DoesNotExist: raise CreateOrderError( _("User has no billing profile, please update billing details first") ) if profile.country not in BILLING_ALLOWED_COUNTRIES: raise CreateOrderError(NOT_ALLOWED_COUNTRIES_ERROR_MESSAGE) price_result = calculate_price_when_buy_credits( quantity, include_tax=True, country=profile.country ) order = BillingCreditOrder.objects.create( user=user, quantity=quantity, amount_total=price_result.amount_total, product_amount=price_result.product_amount, tax_amount=price_result.tax_amount, currency=price_result.currency, status=BillingCreditOrder.Status.CREATED, ) log_user_event( user=user, event_name=UserEvent.EventName.BILLING_CREATE_ORDER, event_body={ "order_id": order.id, "quantity": quantity, "amount_total": str(price_result.amount_total), "product_amount": str(price_result.product_amount), "tax_amount": str(price_result.tax_amount), "currency": price_result.currency, }, ) return order def checkout_order( request: Request, order: BillingCreditOrder, payment_method_id: str ) -> tuple[str, dict]: log_user_event( user=request.user, event_name=UserEvent.EventName.BILLING_CHECKOUT_ORDER, event_body={ "order_id": order.id, "payment_method_id": payment_method_id, }, ) return StripeProvider(request.user, request).checkout_order( order, payment_method_id )