import logging from datetime import timedelta from typing import Optional from django.db import transaction, models from django.utils import timezone from rest_framework.request import Request from labs.models import ( User, UserCreditSummary, UserCreditDetail, CreditTransaction, BillingUserProfile, BillingStripePaymentMethod, ) from labs.billing.providers.stripe import StripeProvider from labs.billing.utils import round_to_end_of_day from labs.billing.consts import BILLING_ALLOWED_COUNTRIES logger = logging.getLogger(__name__) INITIAL_FREE_CREDITS_AMOUNT = 5 INITIAL_FREE_CREDITS_EXPIRES_IN_DAYS = 366 @transaction.atomic def init_user_credit_record(user: User) -> None: """ Initialize the credit record for a user. Should be called when a user is created(Sign up). Args: user: The user to initialize the credit record for Returns: None """ credit_summary, created = ( UserCreditSummary.objects.select_for_update().get_or_create( user=user, ) ) if not created: logger.info( f"User {user.id} already has a credit record. Skipping initialization." ) return UserCreditDetail.objects.create( user=user, credit_type=UserCreditDetail.CreditType.PAID, amount=0, ) UserCreditDetail.objects.create( user=user, credit_type=UserCreditDetail.CreditType.FREE, initial_amount=INITIAL_FREE_CREDITS_AMOUNT, amount=INITIAL_FREE_CREDITS_AMOUNT, remark="Sign up bonus", expires_at=round_to_end_of_day( timezone.now() + timedelta(days=INITIAL_FREE_CREDITS_EXPIRES_IN_DAYS) ), ) credit_summary.free_credits = INITIAL_FREE_CREDITS_AMOUNT credit_summary.save() CreditTransaction.objects.create( user=user, transaction_type=CreditTransaction.TransactionType.BONUS, paid_credits=0, free_credits=INITIAL_FREE_CREDITS_AMOUNT, remark="Sign up bonus", ) logger.info(f"User {user.id} credit record initialized.") def get_balance(user: User) -> dict: """ Get the balance of a user. Returns: dict: The balance of the user. """ now = timezone.now() data = { "paid_credits": user.credit_summary.paid_credits, } free_credits_details = ( UserCreditDetail.objects.filter( user=user, credit_type=UserCreditDetail.CreditType.FREE, expires_at__gte=now, ) .order_by("expires_at", "created_at") .values( "id", "credit_type", "amount", "initial_amount", "expires_at", "created_at" ) ) data["free_credits"] = sum(detail["amount"] for detail in free_credits_details) data["free_credits_details"] = free_credits_details return data @transaction.atomic def update_user_credit_summary(user: User) -> None: """ Update the credit summary for a user. Since the free credits may expire, we need to update the credit summary to ensure the credits summary is correct. """ try: credit_summary = UserCreditSummary.objects.select_for_update().get(user=user) except UserCreditSummary.DoesNotExist: init_user_credit_record(user) credit_summary = UserCreditSummary.objects.select_for_update().get(user=user) now = timezone.now() free_credits_amount = ( UserCreditDetail.objects.filter( user=user, credit_type=UserCreditDetail.CreditType.FREE, amount__gt=0, expires_at__gte=now, ).aggregate(total_amount=models.Sum("amount"))["total_amount"] or 0 ) credit_summary.free_credits = free_credits_amount credit_summary.save() logger.info(f"User {user.id} credit summary updated.") def get_user_billing_profile(user: User) -> tuple[Optional[dict], list]: """ Get the user billing profile for a user. Args: user: The user to get the profile for Returns: tuple[Optional[dict], list]: A tuple containing: - billing_details: User's billing address information or None if not set - stripe_payment_methods: List of user's stripe payment methods """ billing_details = None try: profile = BillingUserProfile.objects.get(user=user) if profile.country: billing_details = { "country": profile.country, "city": profile.city, "state": profile.state, "postal_code": profile.postal_code, "line1": profile.line1, "line2": profile.line2, } except BillingUserProfile.DoesNotExist: logger.debug(f"No billing profile found for user {user.id}") stripe_payment_methods = BillingStripePaymentMethod.objects.filter( user=user ).values( "stripe_pm_id", "card_brand", "last4", "exp_month", "exp_year", ) return billing_details, stripe_payment_methods