from typing import Type from constance import config from django.conf import settings from django.contrib.postgres.indexes import GinIndex from django.db.models import ( CASCADE, CharField, FloatField, ForeignKey, JSONField, OneToOneField, PositiveIntegerField, QuerySet, Sum, ) from django.utils.timezone import now from django.utils.translation import gettext_lazy as _ from core.abstract import NiceModel from evibes.utils.misc import create_object from payments.gateways import AbstractGateway from payments.managers import GatewayManager class Transaction(NiceModel): amount = FloatField(null=False, blank=False) balance = ForeignKey("payments.Balance", on_delete=CASCADE, blank=True, null=True, related_name="transactions") currency = CharField(max_length=3, null=False, blank=False) payment_method = CharField(max_length=20, null=True, blank=True) order = ForeignKey( "core.Order", on_delete=CASCADE, blank=True, null=True, help_text=_("order to process after paid"), related_name="payments_transactions", ) process = JSONField(verbose_name=_("processing details"), default=dict) gateway = ForeignKey("payments.Gateway", on_delete=CASCADE, blank=True, null=True, related_name="transactions") def __str__(self): return ( f"{self.balance.user.email} | {self.amount}" if self.balance else f"{self.order.attributes.get('customer_email')} | {self.amount}" ) def save(self, **kwargs): if self.amount != 0.0 and ( (config.PAYMENT_GATEWAY_MINIMUM <= self.amount <= config.PAYMENT_GATEWAY_MAXIMUM) or (config.PAYMENT_GATEWAY_MINIMUM == 0 and config.PAYMENT_GATEWAY_MAXIMUM == 0) ): if len(str(self.amount).split(".")[1]) > 2: self.amount = round(self.amount, 2) super().save(**kwargs) return self raise ValueError( _(f"transaction amount must fit into {config.PAYMENT_GATEWAY_MINIMUM}-{config.PAYMENT_GATEWAY_MAXIMUM}") ) class Meta: verbose_name = _("transaction") verbose_name_plural = _("transactions") indexes = [ GinIndex(fields=["process"]), ] class Balance(NiceModel): amount = FloatField(null=False, blank=False, default=0) user = OneToOneField( to="vibes_auth.User", on_delete=CASCADE, blank=True, null=True, related_name="payments_balance" ) transactions: QuerySet["Transaction"] def __str__(self): return f"{self.user.email} | {self.amount}" class Meta: verbose_name = _("balance") verbose_name_plural = _("balances") def save(self, **kwargs): if self.amount != 0.0 and len(str(self.amount).split(".")[1]) > 2: self.amount = round(self.amount, 2) super().save(**kwargs) class Gateway(NiceModel): objects = GatewayManager() name = CharField(max_length=20, null=False, blank=False, verbose_name=_("name")) default_currency = CharField( max_length=4, null=False, blank=False, verbose_name=_("default currency"), choices=settings.CURRENCIES_WITH_SYMBOLS, ) currencies = CharField( max_length=255, null=False, blank=False, verbose_name=_("currencies"), help_text=_( f"comma separated list of currencies supported by this gateway, " f"choose from {', '.join([code for code, _ in settings.CURRENCIES_WITH_SYMBOLS])}" ), ) integration_path = CharField(max_length=255, null=True, blank=True) minimum_transaction_amount = FloatField( null=False, blank=False, default=0, verbose_name=_("minimum transaction amount") ) maximum_transaction_amount = FloatField( null=False, blank=False, default=0, verbose_name=_("maximum transaction amount") ) daily_limit = PositiveIntegerField( null=False, blank=False, default=0, verbose_name=_("daily limit"), help_text=_("daily sum limit of transactions' amounts. 0 means no limit"), ) monthly_limit = PositiveIntegerField( null=False, blank=False, default=0, verbose_name=_("monthly limit"), help_text=_("monthly sum limit of transactions' amounts. 0 means no limit"), ) priority = PositiveIntegerField(null=False, blank=False, default=10, verbose_name=_("priority"), unique=True) integration_variables = JSONField(null=False, blank=False, default=dict, verbose_name=_("integration variables")) def __str__(self): return self.name class Meta: verbose_name = _("payment gateway") verbose_name_plural = _("payment gateways") @property def can_be_used(self) -> bool: if not self.is_active: return False today = now().date() current_month_start = today.replace(day=1) daily_sum = self.transactions.filter(created__date=today).aggregate(total=Sum("amount"))["total"] or 0 monthly_sum = ( self.transactions.filter(created__gte=current_month_start).aggregate(total=Sum("amount"))["total"] or 0 ) daily_ok = self.daily_limit == 0 or daily_sum < self.daily_limit monthly_ok = self.monthly_limit == 0 or monthly_sum < self.monthly_limit return daily_ok and monthly_ok @can_be_used.setter def can_be_used(self, value: bool): self.__dict__["can_be_used"] = value def get_integration_class_object(self, raise_exc: bool = True) -> Type[AbstractGateway] | None: if not self.integration_path: if raise_exc: raise ValueError(_("gateway integration path is not set")) return None try: module_name, class_name = self.integration_path.rsplit(".", 1) except ValueError as exc: raise ValueError(_("invalid integration path: %(path)s") % {"path": self.integration_path}) from exc return create_object(module_name, class_name)