from django.http import JsonResponse, HttpResponse from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods from django.utils.decorators import method_decorator from django.views import View from django.contrib.auth.decorators import login_required from django.contrib.admin.views.decorators import staff_member_required from django.shortcuts import render, get_object_or_404, redirect from django.contrib import messages from django.core.paginator import Paginator from django.db.models import Q, Sum, Count from django.utils import timezone import json import logging from .models import MpesaCallback, MpesaConfiguration from loans.models import MpesaTransaction from .services import PaymentProcessor, MpesaService from .forms import MpesaConfigurationForm, PaymentSimulationForm from users.models import CustomUser logger = logging.getLogger(__name__) @method_decorator(csrf_exempt, name='dispatch') class MpesaValidationView(View): """ Handle M-Pesa validation callbacks """ def post(self, request): try: # Get client IP for logging client_ip = request.META.get('HTTP_X_FORWARDED_FOR') if client_ip: client_ip = client_ip.split(',')[0].strip() else: client_ip = request.META.get('REMOTE_ADDR') # Parse JSON data callback_data = json.loads(request.body.decode('utf-8')) # Log the callback logger.info(f"M-Pesa validation callback from {client_ip}: {callback_data}") # Process the validation response = PaymentProcessor.process_validation_callback(callback_data) return JsonResponse(response) except json.JSONDecodeError: logger.error("Invalid JSON in M-Pesa validation callback") return JsonResponse({ 'ResultCode': 'C2B00016', 'ResultDesc': 'Invalid JSON format' }) except Exception as e: logger.error(f"Error processing M-Pesa validation: {str(e)}") return JsonResponse({ 'ResultCode': 'C2B00016', 'ResultDesc': 'System error' }) @method_decorator(csrf_exempt, name='dispatch') class MpesaConfirmationView(View): """ Handle M-Pesa confirmation callbacks """ def post(self, request): try: # Get client IP for logging client_ip = request.META.get('HTTP_X_FORWARDED_FOR') if client_ip: client_ip = client_ip.split(',')[0].strip() else: client_ip = request.META.get('REMOTE_ADDR') # Parse JSON data callback_data = json.loads(request.body.decode('utf-8')) # Log the callback logger.info(f"M-Pesa confirmation callback from {client_ip}: {callback_data}") # Process the confirmation response = PaymentProcessor.process_confirmation_callback(callback_data) return JsonResponse(response) except json.JSONDecodeError: logger.error("Invalid JSON in M-Pesa confirmation callback") return JsonResponse({ 'ResultCode': '0', 'ResultDesc': 'Invalid JSON but acknowledged' }) except Exception as e: logger.error(f"Error processing M-Pesa confirmation: {str(e)}") return JsonResponse({ 'ResultCode': '0', 'ResultDesc': 'Error but acknowledged' }) @login_required @staff_member_required def mpesa_dashboard(request): """ M-Pesa dashboard for admin users """ # Get statistics today = timezone.now().date() # Branch filtering - get the base queryset filtered by branch from users.models import Branch selected_branch_id = request.session.get('selected_branch_id') base_queryset = MpesaTransaction.objects.all() if selected_branch_id: try: branch = Branch.objects.get(id=selected_branch_id) if branch.mpesa_shortcode: # Filter transactions by the branch's paybill number base_queryset = base_queryset.filter(business_short_code=branch.mpesa_shortcode) except Branch.DoesNotExist: pass stats = { 'total_transactions': base_queryset.count(), 'today_transactions': base_queryset.filter(created_at__date=today).count(), 'pending_transactions': base_queryset.filter(status='pending').count(), 'failed_transactions': base_queryset.filter(status='failed').count(), 'total_amount_today': base_queryset.filter( created_at__date=today, status='processed' ).aggregate(total=Sum('amount'))['total'] or 0, 'success_rate': 0 } # Calculate success rate total = base_queryset.count() if total > 0: processed = base_queryset.filter(status='processed').count() stats['success_rate'] = round((processed / total) * 100, 1) # Recent transactions recent_transactions = base_queryset.select_related( 'borrower', 'loan' ).order_by('-created_at')[:10] # Failed transactions needing attention failed_transactions = base_queryset.filter( status__in=['failed', 'pending'] ).select_related('borrower').order_by('-created_at')[:5] context = { 'stats': stats, 'recent_transactions': recent_transactions, 'failed_transactions': failed_transactions, } return render(request, 'payments/dashboard.html', context) @login_required @staff_member_required def mpesa_transactions(request): """ List all M-Pesa transactions with filtering """ transactions = MpesaTransaction.objects.select_related( 'borrower', 'loan', 'repayment' ).order_by('-created_at') # Branch filtering - filter by selected branch's paybill number from users.models import Branch selected_branch_id = request.session.get('selected_branch_id') if selected_branch_id: try: branch = Branch.objects.get(id=selected_branch_id) if branch.mpesa_shortcode: # Filter transactions by the branch's paybill number transactions = transactions.filter(business_short_code=branch.mpesa_shortcode) except Branch.DoesNotExist: pass # Filtering status_filter = request.GET.get('status') if status_filter: transactions = transactions.filter(status=status_filter) phone_filter = request.GET.get('phone') if phone_filter: transactions = transactions.filter(msisdn__icontains=phone_filter) amount_min = request.GET.get('amount_min') if amount_min: try: transactions = transactions.filter(trans_amount__gte=float(amount_min)) except ValueError: pass amount_max = request.GET.get('amount_max') if amount_max: try: transactions = transactions.filter(trans_amount__lte=float(amount_max)) except ValueError: pass date_from = request.GET.get('date_from') if date_from: transactions = transactions.filter(created_at__date__gte=date_from) date_to = request.GET.get('date_to') if date_to: transactions = transactions.filter(created_at__date__lte=date_to) # Pagination paginator = Paginator(transactions, 25) page_number = request.GET.get('page') page_obj = paginator.get_page(page_number) context = { 'page_obj': page_obj, 'status_choices': MpesaTransaction.STATUS_CHOICES, 'filters': { 'status': status_filter, 'phone': phone_filter, 'amount_min': amount_min, 'amount_max': amount_max, 'date_from': date_from, 'date_to': date_to, } } return render(request, 'payments/transactions.html', context) @login_required @staff_member_required def mpesa_transaction_detail(request, transaction_id): """ View details of a specific M-Pesa transaction """ transaction = get_object_or_404(MpesaTransaction, id=transaction_id) # Get related callbacks callbacks = MpesaCallback.objects.filter(transaction=transaction).order_by('-created_at') context = { 'transaction': transaction, 'callbacks': callbacks, } return render(request, 'payments/transaction_detail.html', context) @login_required @staff_member_required def reprocess_transaction(request, transaction_id): """ Manually reprocess a failed M-Pesa transaction This allows matching payments even if the ID was added after payment was made """ transaction = get_object_or_404(MpesaTransaction, id=transaction_id) if request.method == 'POST': try: # Force re-matching by clearing borrower/loan # This allows matching even if ID was added after payment force_rematch = request.POST.get('force_rematch', 'false') == 'true' success = transaction.process_payment(force_rematch=True) transaction.refresh_from_db() if success and transaction.repayment: messages.success(request, f'Transaction {transaction.trans_id} processed successfully! Repayment: {transaction.repayment.receipt_number}') elif success: messages.warning(request, f'Transaction processed but no repayment created. Status: {transaction.status}') else: error_msg = transaction.processing_notes or 'Unknown error' messages.error(request, f'Failed to process transaction: {error_msg}') except Exception as e: messages.error(request, f'Error processing transaction: {str(e)}') # Redirect back to the page that called this (could be callbacks or transaction detail) redirect_to = request.GET.get('redirect_to', 'payments:transaction_detail') if redirect_to == 'payments:callbacks': return redirect('payments:callbacks') return redirect('payments:transaction_detail', transaction_id=transaction_id) return render(request, 'payments/reprocess_transaction.html', {'transaction': transaction}) @login_required @staff_member_required def mpesa_configuration(request): """ Manage M-Pesa configuration """ config = MpesaConfiguration.get_active_config() if request.method == 'POST': form = MpesaConfigurationForm(request.POST, instance=config) if form.is_valid(): config = form.save() messages.success(request, 'M-Pesa configuration updated successfully') # Try to register URLs try: service = MpesaService(config) result = service.register_urls() messages.success(request, f'URLs registered successfully: {result}') except Exception as e: messages.warning(request, f'Configuration saved but URL registration failed: {str(e)}') return redirect('payments:configuration') else: form = MpesaConfigurationForm(instance=config) context = { 'form': form, 'config': config, } return render(request, 'payments/configuration.html', context) @login_required @staff_member_required def test_mpesa_payment(request): """ Test M-Pesa payment simulation (sandbox only) """ config = MpesaConfiguration.get_active_config() if not config or config.environment != 'sandbox': messages.error(request, 'Payment simulation is only available in sandbox environment') return redirect('payments:dashboard') if request.method == 'POST': form = PaymentSimulationForm(request.POST) if form.is_valid(): try: service = MpesaService(config) result = service.simulate_c2b_payment( phone_number=form.cleaned_data['phone_number'], amount=form.cleaned_data['amount'], bill_ref_number=form.cleaned_data['bill_ref_number'] ) messages.success(request, f'Payment simulation successful: {result}') except Exception as e: messages.error(request, f'Payment simulation failed: {str(e)}') return redirect('payments:test_payment') else: form = PaymentSimulationForm() context = { 'form': form, 'config': config, } return render(request, 'payments/test_payment.html', context) @login_required @staff_member_required def mpesa_callbacks(request): """ View M-Pesa callback logs with borrower and repayment details """ # Use prefetch_related to avoid collation issues and get related data callbacks = MpesaCallback.objects.prefetch_related( 'transaction__borrower', 'transaction__repayment', 'transaction__loan' ).order_by('-created_at') # Branch filtering - filter by selected branch's paybill number from users.models import Branch selected_branch_id = request.session.get('selected_branch_id') if selected_branch_id: try: branch = Branch.objects.get(id=selected_branch_id) if branch.mpesa_shortcode: # Filter callbacks by the branch's paybill number callbacks = callbacks.filter(transaction__business_short_code=branch.mpesa_shortcode) except Branch.DoesNotExist: pass # Filtering callback_type = request.GET.get('type') if callback_type: callbacks = callbacks.filter(callback_type=callback_type) processed = request.GET.get('processed') if processed: callbacks = callbacks.filter(processed=processed == 'true') # Pagination paginator = Paginator(callbacks, 25) page_number = request.GET.get('page') page_obj = paginator.get_page(page_number) # Get all active loans for the manual matching dropdown from loans.models import Loan active_loans = Loan.active_objects.filter( status='active' ).select_related('borrower', 'application__loan_product').order_by('due_date', 'borrower__first_name') context = { 'page_obj': page_obj, 'callback_types': MpesaCallback.CALLBACK_TYPES, 'active_loans': active_loans, 'filters': { 'type': callback_type, 'processed': processed, } } return render(request, 'payments/callbacks.html', context) @login_required @staff_member_required def manual_match_payment(request, transaction_id): """ Manually match a payment to a specific loan """ transaction = get_object_or_404(MpesaTransaction, id=transaction_id) if request.method == 'POST': loan_id = request.POST.get('loan_id') if not loan_id: messages.error(request, 'Please select a loan') return redirect('payments:callbacks') try: from loans.models import Loan loan = Loan.active_objects.get(id=loan_id, status='active') borrower = loan.borrower # Update the transaction with the selected loan and borrower transaction.borrower = borrower transaction.loan = loan transaction.bill_ref_number = borrower.id_number # Update the bill ref to match transaction.processing_notes = f"Manually matched to loan {loan.loan_number} ({borrower.get_full_name()}) by {request.user.get_full_name()}" transaction.processed_by = request.user transaction.save() # Try to process the payment success = transaction.process_payment(force_rematch=True) transaction.refresh_from_db() if success and transaction.repayment: messages.success( request, f'✓ Payment successfully matched to loan {loan.loan_number} and processed! ' f'Receipt: {transaction.repayment.receipt_number}' ) elif success: messages.warning( request, f'Payment matched to loan {loan.loan_number} but no repayment created. ' f'Status: {transaction.status}' ) else: error_msg = transaction.processing_notes or 'Unknown error' messages.error(request, f'Payment matched but processing failed: {error_msg}') except Loan.DoesNotExist: messages.error(request, 'Invalid loan selected or loan is not active') except Exception as e: logger.error(f"Error manually matching payment: {str(e)}") messages.error(request, f'Error matching payment: {str(e)}') return redirect('payments:callbacks') @login_required def my_payments(request): """ Show borrower's own M-Pesa payments """ if request.user.role != 'borrower': messages.error(request, 'Access denied') return redirect('dashboard') transactions = MpesaTransaction.objects.filter( borrower=request.user ).select_related('loan', 'repayment').order_by('-created_at') # Pagination paginator = Paginator(transactions, 10) page_number = request.GET.get('page') page_obj = paginator.get_page(page_number) context = { 'page_obj': page_obj, } return render(request, 'payments/my_payments.html', context) @login_required @staff_member_required def unconfirmed_payments(request): """ List all unconfirmed payments that need admin approval """ from .models import UnconfirmedPayment # Fetch without select_related to avoid collation conflicts # Django will fetch related objects separately unconfirmed = UnconfirmedPayment.objects.all().order_by('-created_at') # Branch filtering - filter by selected branch's paybill number from users.models import Branch selected_branch_id = request.session.get('selected_branch_id') if selected_branch_id: try: branch = Branch.objects.get(id=selected_branch_id) if branch.mpesa_shortcode: # Filter unconfirmed payments by the branch's paybill number # Use extra() with COLLATE to avoid collation mismatch unconfirmed = unconfirmed.extra( where=["mpesa_transactions.business_short_code COLLATE utf8mb4_unicode_ci = %s"], params=[branch.mpesa_shortcode], tables=['mpesa_transactions'], ).filter(mpesa_transaction__isnull=False) except Branch.DoesNotExist: pass # Filtering status_filter = request.GET.get('status', 'pending') if status_filter: unconfirmed = unconfirmed.filter(status=status_filter) match_type_filter = request.GET.get('match_type') if match_type_filter: unconfirmed = unconfirmed.filter(match_type=match_type_filter) # Pagination paginator = Paginator(unconfirmed, 25) page_number = request.GET.get('page') page_obj = paginator.get_page(page_number) # Statistics - use separate queries to avoid collation issues stats = { 'pending': UnconfirmedPayment.objects.filter(status='pending').count(), 'approved': UnconfirmedPayment.objects.filter(status='approved').count(), 'rejected': UnconfirmedPayment.objects.filter(status='rejected').count(), 'processed': UnconfirmedPayment.objects.filter(status='processed').count(), } context = { 'page_obj': page_obj, 'status_choices': UnconfirmedPayment.STATUS_CHOICES, 'match_type_choices': UnconfirmedPayment.MATCH_TYPE_CHOICES, 'stats': stats, 'filters': { 'status': status_filter, 'match_type': match_type_filter, } } return render(request, 'payments/unconfirmed_payments.html', context) @login_required @staff_member_required def unconfirmed_payment_detail(request, payment_id): """ View details of an unconfirmed payment """ from .models import UnconfirmedPayment # Fetch without select_related to avoid collation conflicts unconfirmed = get_object_or_404( UnconfirmedPayment.objects.all(), id=payment_id ) context = { 'unconfirmed': unconfirmed, 'transaction': unconfirmed.mpesa_transaction, } return render(request, 'payments/unconfirmed_payment_detail.html', context) @login_required @staff_member_required def approve_unconfirmed_payment(request, payment_id): """ Approve an unconfirmed payment """ from .models import UnconfirmedPayment unconfirmed = get_object_or_404(UnconfirmedPayment, id=payment_id) if request.method == 'POST': admin_notes = request.POST.get('admin_notes', '').strip() success = unconfirmed.approve(request.user, admin_notes) if success: messages.success(request, f'Payment of KES {unconfirmed.mpesa_transaction.amount:,.2f} approved and processed successfully.') else: messages.error(request, 'Failed to process payment. Please check the suggested borrower.') return redirect('payments:unconfirmed_payment_detail', payment_id=payment_id) return redirect('payments:unconfirmed_payments') @login_required @staff_member_required def reject_unconfirmed_payment(request, payment_id): """ Reject an unconfirmed payment """ from .models import UnconfirmedPayment unconfirmed = get_object_or_404(UnconfirmedPayment, id=payment_id) if request.method == 'POST': reason = request.POST.get('rejection_reason', '').strip() if not reason: messages.error(request, 'Please provide a reason for rejection.') return redirect('payments:unconfirmed_payment_detail', payment_id=payment_id) unconfirmed.reject(request.user, reason) messages.success(request, 'Payment rejected successfully.') return redirect('payments:unconfirmed_payments') return redirect('payments:unconfirmed_payments')