from django.db import models from django.contrib.auth import get_user_model from django.utils import timezone from decimal import Decimal import uuid import json User = get_user_model() class MpesaConfiguration(models.Model): """ M-Pesa API configuration settings """ ENVIRONMENT_CHOICES = [ ('sandbox', 'Sandbox'), ('production', 'Production'), ] id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) environment = models.CharField(max_length=20, choices=ENVIRONMENT_CHOICES, default='sandbox') consumer_key = models.CharField(max_length=100) consumer_secret = models.CharField(max_length=100) business_short_code = models.CharField(max_length=10) passkey = models.CharField(max_length=100, blank=True, null=True) # Callback URLs validation_url = models.URLField(help_text="URL for validating payments") confirmation_url = models.URLField(help_text="URL for confirming payments") # Response type for failed validations response_type = models.CharField( max_length=20, choices=[('Completed', 'Completed'), ('Cancelled', 'Cancelled')], default='Completed', help_text="What to do if validation fails" ) is_active = models.BooleanField(default=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: db_table = 'mpesa_configurations' ordering = ['-created_at'] def __str__(self): return f"M-Pesa Config ({self.environment}) - {self.business_short_code}" @classmethod def get_active_config(cls): """Get the active M-Pesa configuration""" return cls.objects.filter(is_active=True).first() # We'll enhance the existing MpesaTransaction model in loans app instead class MpesaCallback(models.Model): """ Raw M-Pesa callback logs for debugging and audit """ CALLBACK_TYPES = [ ('validation', 'Validation'), ('confirmation', 'Confirmation'), ] id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) callback_type = models.CharField(max_length=20, choices=CALLBACK_TYPES) raw_data = models.JSONField() ip_address = models.GenericIPAddressField(blank=True, null=True) user_agent = models.CharField(max_length=500, blank=True, null=True) # Processing transaction = models.ForeignKey('loans.MpesaTransaction', on_delete=models.SET_NULL, null=True, blank=True) processed = models.BooleanField(default=False) response_sent = models.JSONField(blank=True, null=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: db_table = 'mpesa_callbacks' ordering = ['-created_at'] def __str__(self): return f"M-Pesa {self.callback_type} callback - {self.created_at}" class MpesaAccessToken(models.Model): """ Store M-Pesa access tokens for API calls """ id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) configuration = models.ForeignKey(MpesaConfiguration, on_delete=models.CASCADE) access_token = models.TextField() expires_at = models.DateTimeField() created_at = models.DateTimeField(auto_now_add=True) class Meta: db_table = 'mpesa_access_tokens' ordering = ['-created_at'] def __str__(self): return f"Access Token for {self.configuration.business_short_code}" def is_expired(self): return timezone.now() >= self.expires_at @classmethod def get_valid_token(cls, configuration): """Get a valid access token, creating one if needed""" token = cls.objects.filter( configuration=configuration, expires_at__gt=timezone.now() ).first() if token: return token.access_token # Generate new token from .services import MpesaService service = MpesaService(configuration) return service.get_access_token() class PaymentAllocation(models.Model): """ Track how payments are allocated across multiple loans """ id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) mpesa_transaction = models.ForeignKey('loans.MpesaTransaction', on_delete=models.CASCADE, related_name='allocations') loan = models.ForeignKey('loans.Loan', on_delete=models.CASCADE) allocated_amount = models.DecimalField(max_digits=12, decimal_places=2) repayment = models.OneToOneField('loans.Repayment', on_delete=models.CASCADE) created_at = models.DateTimeField(auto_now_add=True) class Meta: db_table = 'payment_allocations' ordering = ['-created_at'] def __str__(self): return f"Allocation: KES {self.allocated_amount} to {self.loan.loan_number}" class UnconfirmedPayment(models.Model): """ Track M-Pesa payments that need admin approval when ID number or phone number don't match """ MATCH_TYPE_CHOICES = [ ('id_only', 'ID Number Matches (Phone Doesn\'t)'), ('phone_only', 'Phone Matches (ID Doesn\'t)'), ('none', 'No Match Found'), ] STATUS_CHOICES = [ ('pending', 'Pending Approval'), ('approved', 'Approved'), ('rejected', 'Rejected'), ('processed', 'Processed'), ] id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) mpesa_transaction = models.OneToOneField('loans.MpesaTransaction', on_delete=models.CASCADE, related_name='unconfirmed_payment') # Payment details payment_phone = models.CharField(max_length=17, blank=True, null=True, help_text="Phone number from payment") payment_id_number = models.CharField(max_length=50, blank=True, null=True, help_text="ID number from payment") # Matching info suggested_borrower = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='unconfirmed_payments', help_text="Borrower suggested by system") match_type = models.CharField(max_length=20, choices=MATCH_TYPE_CHOICES, default='none') # Admin review status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending') approved_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='approved_unconfirmed_payments') approved_at = models.DateTimeField(null=True, blank=True) rejection_reason = models.TextField(blank=True, null=True) admin_notes = models.TextField(blank=True, null=True) # Processing notes = models.TextField(blank=True, null=True, help_text="System notes about the mismatch") created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: db_table = 'unconfirmed_payments' ordering = ['-created_at'] verbose_name = 'Unconfirmed Payment' verbose_name_plural = 'Unconfirmed Payments' def __str__(self): return f"Unconfirmed Payment: KES {self.mpesa_transaction.amount} - {self.get_match_type_display()}" def approve(self, approved_by, admin_notes=None): """Approve the unconfirmed payment and process it""" from loans.models import MpesaTransaction self.status = 'approved' self.approved_by = approved_by self.approved_at = timezone.now() if admin_notes: self.admin_notes = admin_notes self.save() # Link transaction to suggested borrower and process transaction = self.mpesa_transaction if self.suggested_borrower: transaction.borrower = self.suggested_borrower transaction.save() # Process the payment success = transaction.process_payment() if success: self.status = 'processed' self.save() return True return False def reject(self, rejected_by, reason): """Reject the unconfirmed payment""" self.status = 'rejected' self.approved_by = rejected_by self.approved_at = timezone.now() self.rejection_reason = reason self.save() # Mark transaction as rejected self.mpesa_transaction.status = 'rejected' self.mpesa_transaction.processing_notes = f"Rejected by admin: {reason}" self.mpesa_transaction.save()