schon/engine/payments/models.py

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)