from datetime import datetime, time from django.conf import settings from django.contrib.postgres.indexes import GinIndex from django.db.models import ( CASCADE, CharField, FloatField, ForeignKey, Index, JSONField, OneToOneField, PositiveIntegerField, QuerySet, Sum, ) from django.utils import timezone from django.utils.translation import gettext_lazy as _ from engine.core.abstract import NiceModel from engine.payments.gateways import AbstractGateway from engine.payments.managers import GatewayManager from evibes.utils.misc import create_object 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 len(str(self.amount).split(".")[1]) > 2: self.amount = round(self.amount, 2) super().save(**kwargs) return self class Meta: verbose_name = _("transaction") verbose_name_plural = _("transactions") indexes = [ GinIndex(fields=["process"]), Index(fields=["created"]), ] class Balance(NiceModel): amount = FloatField(null=False, blank=False, default=0) user = OneToOneField( to=settings.AUTH_USER_MODEL, 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 = timezone.localdate() tz = timezone.get_current_timezone() month_start = timezone.make_aware( datetime.combine(today.replace(day=1), time.min), tz ) if today.month == 12: next_month_date = today.replace(year=today.year + 1, month=1, day=1) else: next_month_date = today.replace(month=today.month + 1, day=1) month_end = timezone.make_aware(datetime.combine(next_month_date, time.min), tz) daily_sum = ( self.transactions.filter(created__date=today).aggregate( total=Sum("amount") )["total"] or 0 ) monthly_sum = ( self.transactions.filter( created__gte=month_start, created__lt=month_end ).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 ) -> 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)