from decimal import Decimal, ROUND_HALF_UP from django.contrib.auth.models import User from django.core.exceptions import ValidationError from django.db import models, transaction class BusinessProfile(models.Model): user = models.OneToOneField(User, on_delete=models.CASCADE, related_name='business_profile') business_name = models.CharField(max_length=120, blank=True) logo = models.FileField(upload_to='business_logos/', blank=True, null=True) opening_ecash = models.DecimalField(max_digits=12, decimal_places=2, default=Decimal('0.00')) opening_physical_cash = models.DecimalField(max_digits=12, decimal_places=2, default=Decimal('0.00')) current_ecash = models.DecimalField(max_digits=12, decimal_places=2, default=Decimal('0.00')) current_physical_cash = models.DecimalField(max_digits=12, decimal_places=2, default=Decimal('0.00')) mobile_token_version = models.PositiveIntegerField(default=1) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ['user__username'] def __str__(self): return self.business_name or f"{self.user.username}'s MoMo Business" @property def owner_label(self): return self.user.get_full_name() or self.user.username @property def total_cash(self): return (self.current_ecash or Decimal('0.00')) + (self.current_physical_cash or Decimal('0.00')) def sync_current_to_opening(self): self.current_ecash = self.opening_ecash self.current_physical_cash = self.opening_physical_cash class Transaction(models.Model): CASH_OUT = 'cash_out' CASH_IN = 'cash_in' SENDING = 'sending' AIRTIME = 'airtime' TRANSFER = 'transfer' DEBT = 'debt' EXPENDITURE = 'expenditure' CREDIT = 'credit' TYPE_CHOICES = [ (CASH_OUT, 'Cash-out'), (CASH_IN, 'Cash-In'), (SENDING, 'Sending'), (AIRTIME, 'Airtime'), (TRANSFER, 'Transfer'), (DEBT, 'Debt'), (EXPENDITURE, 'Expenditure'), (CREDIT, 'Credit'), ] business = models.ForeignKey(BusinessProfile, on_delete=models.CASCADE, related_name='transactions') created_by = models.ForeignKey(User, on_delete=models.CASCADE, related_name='momo_transactions') client_name = models.CharField(max_length=120) amount = models.DecimalField(max_digits=12, decimal_places=2) transaction_type = models.CharField(max_length=20, choices=TYPE_CHOICES) service_charge = models.DecimalField(max_digits=12, decimal_places=2, default=Decimal('0.00')) notes = models.CharField(max_length=255, blank=True) ecash_before = models.DecimalField(max_digits=12, decimal_places=2) ecash_after = models.DecimalField(max_digits=12, decimal_places=2) physical_before = models.DecimalField(max_digits=12, decimal_places=2) physical_after = models.DecimalField(max_digits=12, decimal_places=2) created_at = models.DateTimeField(auto_now_add=True) class Meta: ordering = ['-created_at', '-id'] def __str__(self): return f"{self.get_transaction_type_display()} · {self.client_name} · {self.amount}" @staticmethod def _round(value: Decimal) -> Decimal: return value.quantize(Decimal('0.01'), rounding=ROUND_HALF_UP) @classmethod def calculate_effect(cls, transaction_type: str, amount: Decimal) -> tuple[Decimal, Decimal, Decimal]: amount = cls._round(amount) fee = Decimal('0.00') ecash_delta = Decimal('0.00') physical_delta = Decimal('0.00') if transaction_type == cls.CASH_IN: ecash_delta = -amount physical_delta = amount elif transaction_type == cls.CASH_OUT: ecash_delta = amount physical_delta = -amount elif transaction_type in {cls.AIRTIME, cls.TRANSFER}: ecash_delta = -amount physical_delta = amount elif transaction_type == cls.SENDING: fee = cls._round(amount * Decimal('0.01')) ecash_delta = -amount physical_delta = amount + fee elif transaction_type in {cls.DEBT, cls.EXPENDITURE}: physical_delta = -amount elif transaction_type == cls.CREDIT: physical_delta = amount else: raise ValidationError('Unsupported transaction type.') return cls._round(ecash_delta), cls._round(physical_delta), cls._round(fee) @classmethod @transaction.atomic def create_logged_transaction( cls, *, business: BusinessProfile, user: User, client_name: str, amount: Decimal, transaction_type: str, notes: str = '', ): ecash_delta, physical_delta, fee = cls.calculate_effect(transaction_type, amount) ecash_before = cls._round(business.current_ecash) physical_before = cls._round(business.current_physical_cash) ecash_after = cls._round(ecash_before + ecash_delta) physical_after = cls._round(physical_before + physical_delta) if ecash_after < 0: raise ValidationError('This transaction would make e-cash go below zero.') if physical_after < 0: raise ValidationError('This transaction would make physical cash go below zero.') entry = cls.objects.create( business=business, created_by=user, client_name=client_name, amount=cls._round(amount), transaction_type=transaction_type, service_charge=fee, notes=notes, ecash_before=ecash_before, ecash_after=ecash_after, physical_before=physical_before, physical_after=physical_after, ) business.current_ecash = ecash_after business.current_physical_cash = physical_after business.save(update_fields=['current_ecash', 'current_physical_cash', 'updated_at']) return entry @classmethod def rebalance_business_ledger(cls, business: BusinessProfile): business = BusinessProfile.objects.select_for_update().get(pk=business.pk) ecash_balance = cls._round(business.opening_ecash) physical_balance = cls._round(business.opening_physical_cash) entries = list( cls.objects.select_for_update() .filter(business=business) .order_by('created_at', 'id') ) for entry in entries: ecash_delta, physical_delta, fee = cls.calculate_effect(entry.transaction_type, entry.amount) ecash_before = ecash_balance physical_before = physical_balance ecash_after = cls._round(ecash_before + ecash_delta) physical_after = cls._round(physical_before + physical_delta) if ecash_after < 0: raise ValidationError('This change would make e-cash go below zero in your transaction history.') if physical_after < 0: raise ValidationError('This change would make physical cash go below zero in your transaction history.') cls.objects.filter(pk=entry.pk).update( service_charge=fee, ecash_before=ecash_before, ecash_after=ecash_after, physical_before=physical_before, physical_after=physical_after, ) ecash_balance = ecash_after physical_balance = physical_after business.current_ecash = ecash_balance business.current_physical_cash = physical_balance business.save(update_fields=['current_ecash', 'current_physical_cash', 'updated_at']) return business @transaction.atomic def update_logged_transaction(self, *, client_name: str, amount: Decimal, transaction_type: str, notes: str = ''): business = BusinessProfile.objects.select_for_update().get(pk=self.business_id) self.client_name = client_name self.amount = self.__class__._round(amount) self.transaction_type = transaction_type self.notes = notes self.save(update_fields=['client_name', 'amount', 'transaction_type', 'notes']) business = self.__class__.rebalance_business_ledger(business) refreshed_entry = self.__class__.objects.select_related('created_by').get(pk=self.pk) return { 'transaction': refreshed_entry, 'business': business, } @transaction.atomic def delete_logged_transaction(self): business = BusinessProfile.objects.select_for_update().get(pk=self.business_id) transaction_id = self.pk self.delete() business = self.__class__.rebalance_business_ledger(business) return { 'transaction_id': transaction_id, 'business': business, }