from django.shortcuts import render, redirect, get_object_or_404 from django.contrib import messages from django.contrib.auth.decorators import login_required, user_passes_test from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin from django.views.generic import ListView, UpdateView from django.urls import reverse_lazy from django.core.mail import send_mail from django.conf import settings from django.http import JsonResponse from django.db.models import Q import json import logging from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods from django.core.paginator import Paginator from django.utils import timezone from .models import SystemSettings, CompanyProfile from .forms import CompanyProfileForm, TestEmailForm, SystemSettingsForm from notifications.models import EmailLog from notifications.email_service import email_service from accounts.permissions import ( AdminRequiredMixin, has_admin_access, permission_required, has_permission, get_user_permissions ) logger = logging.getLogger(__name__) def is_admin(user): """Check if user has admin access""" return has_admin_access(user) class AdminRequiredMixin(UserPassesTestMixin): def test_func(self): return self.request.user.is_superuser @login_required @user_passes_test(is_admin) def settings_dashboard(request): """Main settings dashboard with all configurations and POST handling""" if request.method == 'POST': action = request.POST.get('action') if action == 'save_email_settings': # Save email settings email_fields = ['HOST', 'PORT', 'HOST_USER', 'HOST_PASSWORD', 'USE_TLS', 'USE_SSL', 'DEFAULT_FROM_EMAIL', 'TIMEOUT'] for field in email_fields: value = request.POST.get(field, '').strip() if value and not (field == 'HOST_PASSWORD' and value == '••••••••'): SystemSettings.objects.update_or_create( category='email', key=field, defaults={'value': value, 'is_sensitive': field in ['HOST_PASSWORD']} ) messages.success(request, '✅ Email settings saved successfully!') elif action == 'save_mpesa_settings': # Save M-Pesa settings mpesa_fields = ['CONSUMER_KEY', 'CONSUMER_SECRET', 'BUSINESS_SHORT_CODE', 'ENVIRONMENT', 'PASSKEY', 'CALLBACK_BASE_URL'] for field in mpesa_fields: value = request.POST.get(field, '').strip() if value and not (field in ['CONSUMER_SECRET', 'PASSKEY'] and value == '••••••••'): SystemSettings.objects.update_or_create( category='mpesa', key=field, defaults={'value': value, 'is_sensitive': field in ['CONSUMER_SECRET', 'PASSKEY']} ) messages.success(request, '✅ M-Pesa settings saved successfully!') elif action == 'save_sms_settings': # Save SMS settings sms_fields = ['API_KEY', 'API_SECRET', 'SENDER_ID', 'API_URL'] for field in sms_fields: value = request.POST.get(field, '').strip() if value and not (field == 'API_SECRET' and value == '••••••••'): SystemSettings.objects.update_or_create( category='sms', key=field, defaults={'value': value, 'is_sensitive': field in ['API_SECRET']} ) messages.success(request, '✅ SMS settings saved successfully!') elif action == 'save_maps_settings': # Save Google Maps settings value = request.POST.get('API_KEY', '').strip() if value and value != '••••••••': SystemSettings.objects.update_or_create( category='maps', key='API_KEY', defaults={'value': value, 'is_sensitive': True} ) messages.success(request, '✅ Google Maps settings saved successfully!') elif action == 'save_company_settings': # Save company profile settings profile, created = CompanyProfile.objects.get_or_create(id=1) # Update basic fields profile.company_name = request.POST.get('company_name', '').strip() profile.email = request.POST.get('email', '').strip() profile.phone = request.POST.get('phone', '').strip() profile.address = request.POST.get('address', '').strip() profile.website = request.POST.get('website', '').strip() profile.tax_id = request.POST.get('tax_id', '').strip() profile.business_registration = request.POST.get('business_registration', '').strip() # Handle file uploads if 'company_logo' in request.FILES: profile.company_logo = request.FILES['company_logo'] if 'favicon' in request.FILES: profile.favicon = request.FILES['favicon'] profile.save() messages.success(request, '✅ Company profile updated successfully!') return redirect('settings:dashboard') def get_api_settings(category): """Helper function to get settings for a specific API category""" settings = {} category_settings = SystemSettings.objects.filter(category=category) for setting in category_settings: settings[setting.key] = setting.value return settings # Get settings for each API mpesa_settings = get_api_settings('mpesa') email_settings = get_api_settings('email') sms_settings = get_api_settings('sms') mikrotik_settings = get_api_settings('mikrotik') maps_settings = get_api_settings('maps') general_settings = get_api_settings('general') # Get company profile - ensure we always have one try: company_profile = CompanyProfile.objects.first() if not company_profile: # Create default company profile if none exists company_profile = CompanyProfile.objects.create( company_name="OptiNet Solutions" ) except Exception as e: # If there's any error, create a basic profile company_profile = CompanyProfile.objects.create( company_name="OptiNet Solutions" ) # Get email stats try: from notifications.models import EmailLog email_stats = { 'total_sent': EmailLog.objects.filter(status='sent').count(), 'total_failed': EmailLog.objects.filter(status='failed').count(), 'total_pending': EmailLog.objects.filter(status='pending').count(), 'today_sent': EmailLog.objects.filter( status='sent', sent_at__date=timezone.now().date() ).count(), } except: email_stats = None context = { 'mpesa_settings': mpesa_settings, 'email_settings': email_settings, 'sms_settings': sms_settings, 'mikrotik_settings': mikrotik_settings, 'maps_settings': maps_settings, 'general_settings': general_settings, 'company_profile': company_profile, 'email_stats': email_stats, } return render(request, 'settings/dashboard.html', context) @login_required @user_passes_test(is_admin) def update_api_settings(request): """Update API settings via modal forms""" if request.method == 'POST': api_type = request.POST.get('api_type') if not api_type: messages.error(request, 'Invalid API type specified.') return redirect('settings:dashboard') # Define the settings structure for each API api_settings_map = { 'mpesa': { 'consumer_key': {'description': 'M-Pesa Consumer Key', 'is_sensitive': True}, 'consumer_secret': {'description': 'M-Pesa Consumer Secret', 'is_sensitive': True}, 'business_short_code': {'description': 'M-Pesa Business Short Code', 'is_sensitive': False}, 'passkey': {'description': 'M-Pesa Passkey', 'is_sensitive': True}, 'initiator_name': {'description': 'M-Pesa Initiator Name', 'is_sensitive': False}, 'initiator_password': {'description': 'M-Pesa Initiator Password (encrypted)', 'is_sensitive': True}, 'environment': {'description': 'M-Pesa Environment (sandbox/production)', 'is_sensitive': False}, 'stk_callback_url': {'description': 'STK Push Callback URL', 'is_sensitive': False}, 'c2b_validation_url': {'description': 'C2B Validation URL', 'is_sensitive': False}, 'c2b_confirmation_url': {'description': 'C2B Confirmation URL', 'is_sensitive': False}, }, 'email': { 'HOST': {'description': 'SMTP Host', 'is_sensitive': False}, 'PORT': {'description': 'SMTP Port', 'is_sensitive': False}, 'HOST_USER': {'description': 'SMTP Username', 'is_sensitive': False}, 'HOST_PASSWORD': {'description': 'SMTP Password', 'is_sensitive': True}, 'USE_TLS': {'description': 'Use TLS Encryption', 'is_sensitive': False}, }, 'sms': { 'API_KEY': {'description': 'SMS API Key', 'is_sensitive': True}, 'PARTNER_ID': {'description': 'SMS Partner ID', 'is_sensitive': False}, 'SHORT_CODE': {'description': 'SMS Short Code', 'is_sensitive': False}, 'API_URL': {'description': 'SMS API Base URL', 'is_sensitive': False}, 'METHOD': {'description': 'SMS API Method', 'is_sensitive': False}, 'PHONE_PARAMETER': {'description': 'SMS Phone Parameter Name', 'is_sensitive': False}, 'MESSAGE_PARAMETER': {'description': 'SMS Message Parameter Name', 'is_sensitive': False}, }, 'mikrotik': { 'HOST': {'description': 'MikroTik Router IP', 'is_sensitive': False}, 'PORT': {'description': 'MikroTik API Port', 'is_sensitive': False}, 'USERNAME': {'description': 'MikroTik Username', 'is_sensitive': False}, 'PASSWORD': {'description': 'MikroTik Password', 'is_sensitive': True}, }, 'maps': { 'API_KEY': {'description': 'Google Maps API Key', 'is_sensitive': True}, }, 'general': { 'TIMEZONE': {'description': 'System Timezone', 'is_sensitive': False}, 'CURRENCY': {'description': 'Default Currency', 'is_sensitive': False}, } } if api_type not in api_settings_map: messages.error(request, f'Unknown API type: {api_type}') return redirect('settings:dashboard') # Update or create settings for this API settings_updated = 0 # Handle M-Pesa field mapping field_mapping = {} if api_type == 'mpesa': field_mapping = { 'CONSUMER_KEY': 'consumer_key', 'CONSUMER_SECRET': 'consumer_secret', 'BUSINESS_SHORTCODE': 'business_short_code', 'PASSKEY': 'passkey', 'INITIATOR_NAME': 'initiator_name', 'INITIATOR_PASSWORD': 'initiator_password', 'ENVIRONMENT': 'environment', 'STK_CALLBACK_URL': 'stk_callback_url', 'C2B_VALIDATION_URL': 'c2b_validation_url', 'C2B_CONFIRMATION_URL': 'c2b_confirmation_url', } for key, config in api_settings_map[api_type].items(): # Get the form field name (might be different from DB key) form_field = key if api_type == 'mpesa': # Find the form field that maps to this DB key for form_key, db_key in field_mapping.items(): if db_key == key: form_field = form_key break value = request.POST.get(form_field, '').strip() # Skip if value is empty (for optional fields) if not value: continue # Skip if the value is just asterisks (password field placeholder) if value == '••••••••': continue setting, created = SystemSettings.objects.get_or_create( category=api_type, key=key, defaults={ 'value': value, 'description': config['description'], 'is_sensitive': config['is_sensitive'], 'is_required': key in ['CONSUMER_KEY', 'HOST', 'API_KEY'] if api_type != 'general' else False } ) if not created: setting.value = value setting.description = config['description'] setting.is_sensitive = config['is_sensitive'] setting.save() settings_updated += 1 messages.success(request, f'{api_type.upper()} API settings updated successfully! ({settings_updated} settings saved)') return redirect('settings:dashboard') return redirect('settings:dashboard') @login_required @user_passes_test(is_admin) def company_profile(request): """Manage company profile""" profile, created = CompanyProfile.objects.get_or_create(id=1) if request.method == 'POST': form = CompanyProfileForm(request.POST, request.FILES, instance=profile) if form.is_valid(): form.save() messages.success(request, '✅ Company profile updated successfully!') return redirect('settings:dashboard') else: form = CompanyProfileForm(instance=profile) return render(request, 'settings/company_profile.html', {'form': form}) @csrf_exempt @require_http_methods(["POST"]) @login_required @user_passes_test(is_admin) def delete_image(request): """Delete company profile images""" try: import json data = json.loads(request.body) image_type = data.get('image_type') profile = CompanyProfile.objects.first() if not profile: return JsonResponse({'success': False, 'error': 'Company profile not found'}) if image_type == 'logo' and profile.company_logo: # Delete the file if profile.company_logo.path: import os if os.path.exists(profile.company_logo.path): os.remove(profile.company_logo.path) profile.company_logo = None profile.save() return JsonResponse({'success': True, 'message': 'Logo deleted successfully'}) elif image_type == 'favicon' and profile.favicon: # Delete the file if profile.favicon.path: import os if os.path.exists(profile.favicon.path): os.remove(profile.favicon.path) profile.favicon = None profile.save() return JsonResponse({'success': True, 'message': 'Favicon deleted successfully'}) return JsonResponse({'success': False, 'error': f'No {image_type} to delete'}) except Exception as e: return JsonResponse({'success': False, 'error': str(e)}) class EmailLogListView(LoginRequiredMixin, UserPassesTestMixin, ListView): model = EmailLog template_name = 'settings/email_logs.html' context_object_name = 'email_logs' paginate_by = 25 def test_func(self): return has_admin_access(self.request.user) def get_queryset(self): queryset = EmailLog.objects.all().order_by('-sent_at') # Filter by email type email_type = self.request.GET.get('type') if email_type: queryset = queryset.filter(email_type=email_type) # Filter by status status = self.request.GET.get('status') if status: queryset = queryset.filter(status=status) # Search by recipient email or subject search = self.request.GET.get('search') if search: queryset = queryset.filter( Q(recipient_email__icontains=search) | Q(subject__icontains=search) | Q(recipient_name__icontains=search) ) return queryset def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) context['email_types'] = EmailLog.EMAIL_TYPE_CHOICES context['status_choices'] = EmailLog.STATUS_CHOICES context['selected_type'] = self.request.GET.get('type', '') context['selected_status'] = self.request.GET.get('status', '') context['search_query'] = self.request.GET.get('search', '') context['email_stats'] = email_service.get_email_stats() return context @login_required @user_passes_test(is_admin) def email_log_detail(request, log_id): """View detailed email log""" email_log = get_object_or_404(EmailLog, id=log_id) return render(request, 'settings/email_log_detail.html', {'email_log': email_log}) # URL endpoint for the "Test Email" button in base.html @csrf_exempt @require_http_methods(["GET", "POST"]) def test_email_endpoint(request): """Simple test email endpoint for the navbar button""" if not request.user.is_authenticated or not has_admin_access(request.user): return JsonResponse({'success': False, 'error': 'Access denied'}) try: # Use the user's email as default recipient_email = request.user.email or 'admin@optinet.co.ke' success, message = email_service.send_test_email( recipient_email=recipient_email, sent_by=request.user ) return JsonResponse({ 'success': success, 'message': message if success else f'Error: {message}' }) except Exception as e: logger.error(f"Error in test email endpoint: {e}") return JsonResponse({ 'success': False, 'error': f'Unexpected error: {str(e)}' }) # The following functions are kept from the original file as they are not directly replaced or modified by the snippet. # They handle specific functionalities related to M-Pesa, SMS, and other settings. @login_required @user_passes_test(is_admin) def test_email_settings(request): """Test email configuration by sending a test email using the new email service""" if request.method == 'POST': form = TestEmailForm(request.POST) if form.is_valid(): test_email = form.cleaned_data['test_email'] try: # Use the new email service email_service_instance = email_service # Use the imported instance success, error_msg = email_service_instance.send_test_email( recipient_email=test_email, sent_by=request.user ) if success: messages.success( request, f'✅ Test email sent successfully to {test_email}! Please check your inbox (including spam folder).' ) else: messages.error(request, f'❌ {error_msg}') except Exception as e: messages.error(request, f'❌ Error testing email configuration: {str(e)}') else: form = TestEmailForm() # Get current email settings for display email_settings_display = {} try: host_setting = SystemSettings.objects.get(category='email', key='HOST') email_settings_display['host'] = host_setting.value except SystemSettings.DoesNotExist: email_settings_display['host'] = getattr(settings, 'EMAIL_HOST', 'Not configured') try: user_setting = SystemSettings.objects.get(category='email', key='HOST_USER') email_settings_display['user'] = user_setting.value except SystemSettings.DoesNotExist: email_settings_display['user'] = getattr(settings, 'EMAIL_HOST_USER', 'Not configured') context = { 'form': form, 'email_settings': email_settings_display, } return render(request, 'settings/test_email.html', context) @login_required @user_passes_test(is_admin) def test_sms(request): """Test SMS configuration by sending a test SMS""" if request.method == 'POST': test_phone = request.POST.get('test_phone', '').strip() test_message = request.POST.get('test_message', '').strip() if not test_phone or not test_message: messages.error(request, '❌ Please provide both phone number and message.') return redirect('settings:dashboard') try: # Get SMS settings directly for validation from settings.utils import get_sms_settings sms_settings = get_sms_settings() # Check if SMS is configured if not sms_settings.get('api_key') or not sms_settings.get('api_url'): messages.error(request, '❌ SMS API is not properly configured. Please check your SMS settings.') return redirect('settings:dashboard') # Use the SMS service from notifications from notifications.services import NotificationService # Initialize notification service notification_service = NotificationService() # Format phone number properly if test_phone.startswith('0'): formatted_phone = '254' + test_phone[1:] elif test_phone.startswith('+254'): formatted_phone = test_phone[1:] elif test_phone.startswith('254'): formatted_phone = test_phone else: # If no country code, assume Kenya formatted_phone = '254' + test_phone.lstrip('+') # Send test SMS success = notification_service.send_sms( phone_number=formatted_phone, message=test_message, sent_by=request.user, recipient_name="Test User" ) if success: messages.success( request, f'✅ Test SMS sent successfully to {formatted_phone}!' ) else: # Get the last SMS log to show detailed error from notifications.models import SMSLog last_sms_log = SMSLog.objects.filter( recipient_phone=formatted_phone, sent_by=request.user ).order_by('-sent_at').first() error_detail = "" if last_sms_log and last_sms_log.error_message: error_detail = f" Error: {last_sms_log.error_message}" messages.error( request, f'❌ Failed to send test SMS to {formatted_phone}.{error_detail} Please check your SMS configuration.' ) except Exception as e: messages.error(request, f'❌ Error testing SMS configuration: {str(e)}') return redirect('settings:dashboard') @login_required @user_passes_test(is_admin) def update_organization(request): """Update organization/company profile settings""" if request.method == 'POST': try: # Get or create company profile company_profile, created = CompanyProfile.objects.get_or_create( defaults={'company_name': 'OptiNet Solutions'} ) # Update fields from form data company_profile.company_name = request.POST.get('company_name', '').strip() company_profile.business_registration = request.POST.get('business_registration', '').strip() company_profile.tax_id = request.POST.get('tax_id', '').strip() company_profile.phone = request.POST.get('phone', '').strip() company_profile.email = request.POST.get('email', '').strip() company_profile.website = request.POST.get('website', '').strip() company_profile.address = request.POST.get('address', '').strip() company_profile.default_currency = request.POST.get('default_currency', 'KES') company_profile.timezone = request.POST.get('timezone', 'Africa/Nairobi') # Handle file uploads if 'company_logo' in request.FILES: company_profile.company_logo = request.FILES['company_logo'] if 'favicon' in request.FILES: company_profile.favicon = request.FILES['favicon'] # Validate required fields if not company_profile.company_name: messages.error(request, 'Company name is required.') return redirect('settings:dashboard') company_profile.save() messages.success(request, 'Organization settings updated successfully!') except Exception as e: messages.error(request, f'Error updating organization settings: {str(e)}') return redirect('settings:dashboard') class CompanyProfileView(AdminRequiredMixin, LoginRequiredMixin, UpdateView): model = CompanyProfile form_class = CompanyProfileForm template_name = 'settings/company_profile.html' success_url = reverse_lazy('settings:dashboard') def get_object(self, queryset=None): # Get or create the company profile (there should only be one) profile, created = CompanyProfile.objects.get_or_create( defaults={'company_name': 'OptiNet Solutions'} ) return profile def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) context['profile'] = self.get_object() return context def form_valid(self, form): messages.success(self.request, 'Company profile updated successfully!') return super().form_valid(form) @login_required @user_passes_test(is_admin) def register_paybill_urls(request): """Register M-Pesa paybill URLs""" if request.method == 'POST': try: import json data = json.loads(request.body) base_url = data.get('base_url', '').strip() validation_url = data.get('validation_url', '').strip() confirmation_url = data.get('confirmation_url', '').strip() if not base_url: return JsonResponse({ 'success': False, 'message': 'Please provide a base URL.' }) # Build URLs if not provided by frontend if not validation_url: validation_url = f"{base_url}/payments/paybill-validation/" if not confirmation_url: confirmation_url = f"{base_url}/payments/paybill-confirmation/" print(f"Registering M-Pesa URLs - Validation: {validation_url}, Confirmation: {confirmation_url}") # Check if M-Pesa is configured mpesa_settings = SystemSettings.get_mpesa_settings() if not mpesa_settings.get('consumer_key') or not mpesa_settings.get('business_short_code'): return JsonResponse({ 'success': False, 'message': 'M-Pesa API is not properly configured. Please check your M-Pesa settings.' }) # Use the M-Pesa service from payments.services import MpesaService mpesa_service = MpesaService() # Register paybill URLs with separate validation and confirmation endpoints response = mpesa_service.register_c2b_urls(validation_url, confirmation_url) if response.get('ResponseCode') == '0': messages.success( request, f'✅ Paybill URLs registered successfully. Customers can now pay using paybill number {mpesa_settings.get("business_short_code")} with their customer ID as account number.' ) return JsonResponse({ 'success': True, 'message': f'Paybill URLs registered successfully. Paybill Number: {mpesa_settings.get("business_short_code")}', 'paybill_number': mpesa_settings.get("business_short_code"), 'validation_url': validation_url, 'confirmation_url': confirmation_url }) else: error_message = response.get('errorMessage', response.get('ResponseDescription', 'Unknown error')) # Provide specific guidance for common errors if 'Invalid Access Token' in error_message: detailed_message = 'Invalid Access Token. This usually means:\n' + \ '1. Your M-Pesa app is not approved for production (if using production environment)\n' + \ '2. Your Consumer Key/Secret are incorrect\n' + \ '3. Your app credentials don\'t match the selected environment\n\n' + \ 'Please verify your credentials and environment settings.' else: detailed_message = error_message messages.error( request, f'❌ Failed to register paybill URLs: {detailed_message}' ) return JsonResponse({ 'success': False, 'message': f'Failed to register paybill URLs: {detailed_message}' }) except Exception as e: print(f"Error registering paybill URLs: {str(e)}") return JsonResponse({ 'success': False, 'message': f'Error registering URLs: {str(e)}' }) @login_required @user_passes_test(is_admin) def mpesa_balance_history(request): """Get M-Pesa balance inquiry history""" try: # Check if the model exists try: from payments.models import MpesaBalanceInquiry except ImportError: return JsonResponse({ 'success': False, 'message': 'MpesaBalanceInquiry model not found. Please ensure payments app is properly configured.' }) # Get recent balance inquiries balance_inquiries = MpesaBalanceInquiry.objects.all().order_by('-created_at')[:10] balance_data = [] for inquiry in balance_inquiries: try: balance_data.append({ 'id': inquiry.id, 'conversation_id': getattr(inquiry, 'conversation_id', 'N/A'), 'created_at': inquiry.created_at.strftime('%Y-%m-%d %H:%M:%S') if inquiry.created_at else 'N/A', 'completed_at': inquiry.completed_at.strftime('%Y-%m-%d %H:%M:%S') if inquiry.completed_at else None, 'result_code': getattr(inquiry, 'result_code', None), 'result_desc': getattr(inquiry, 'result_desc', None), 'working_account_funds': f"KSh {inquiry.working_account_funds:,.2f}" if inquiry.working_account_funds else None, 'utility_account_funds': f"KSh {inquiry.utility_account_funds:,.2f}" if inquiry.utility_account_funds else None, 'charges_paid_funds': f"KSh {inquiry.charges_paid_funds:,.2f}" if inquiry.charges_paid_funds else None, 'status': 'completed' if inquiry.completed_at else 'pending' }) except Exception as item_error: print(f"Error processing balance inquiry {inquiry.id}: {str(item_error)}") continue return JsonResponse({ 'success': True, 'balance_history': balance_data, 'count': len(balance_data) }) except Exception as e: import traceback error_details = traceback.format_exc() print(f"Error getting balance history: {str(e)}") print(f"Full traceback: {error_details}") return JsonResponse({ 'success': False, 'message': f'Error getting balance history: {str(e)}' }, status=500) @login_required @user_passes_test(is_admin) def test_mpesa_credentials(request): """Test M-Pesa credentials by attempting to get access token""" if request.method == 'POST': try: # Check if M-Pesa is configured from settings.models import SystemSettings mpesa_settings = SystemSettings.get_mpesa_settings() if not mpesa_settings.get('consumer_key') or not mpesa_settings.get('consumer_secret'): return JsonResponse({ 'success': False, 'message': 'M-Pesa Consumer Key and Secret are required. Please configure them first.' }) # Validate credential format consumer_key = mpesa_settings.get('consumer_key', '').strip() consumer_secret = mpesa_settings.get('consumer_secret', '').strip() if len(consumer_key) < 10 or len(consumer_secret) < 10: return JsonResponse({ 'success': False, 'message': 'Consumer Key and Secret appear to be too short. Please check your credentials.' }) # Test the credentials from payments.services import MpesaService mpesa_service = MpesaService() print(f"Testing M-Pesa credentials for environment: {mpesa_settings.get('environment', 'sandbox')}") access_token = mpesa_service.get_access_token() if access_token: return JsonResponse({ 'success': True, 'message': 'M-Pesa credentials are valid! Connection successful.', 'environment': mpesa_settings.get('environment', 'sandbox') }) else: return JsonResponse({ 'success': False, 'message': 'Failed to authenticate with M-Pesa. Please check your credentials and try again.' }) except Exception as e: print(f"Error testing M-Pesa credentials: {str(e)}") return JsonResponse({ 'success': False, 'message': f'Error testing credentials: {str(e)}' }) return JsonResponse({ 'success': False, 'message': 'Invalid request method' }) @login_required @user_passes_test(is_admin) def test_stk_push(request): """Test M-Pesa STK push functionality""" if request.method == 'POST': try: import json data = json.loads(request.body) phone_number = data.get('phone_number', '').strip() amount = data.get('amount', '').strip() if not phone_number or not amount: return JsonResponse({ 'success': False, 'message': 'Please provide both phone number and amount.' }) # Validate amount try: amount_float = float(amount) if amount_float < 1 or amount_float > 70000: return JsonResponse({ 'success': False, 'message': 'Amount must be between KES 1 and KES 70,000.' }) except ValueError: return JsonResponse({ 'success': False, 'message': 'Invalid amount format.' }) # Check if M-Pesa is configured from settings.models import SystemSettings mpesa_settings = SystemSettings.get_mpesa_settings() if not mpesa_settings.get('consumer_key') or not mpesa_settings.get('business_short_code'): return JsonResponse({ 'success': False, 'message': 'M-Pesa API is not properly configured. Please check your M-Pesa settings.' }) # Ensure callback URL is set to production domain callback_url = 'https://system.optinet.co.ke/payments/stk-callback/' SystemSettings.objects.update_or_create( category='mpesa', key='callback_url', defaults={'value': callback_url, 'description': 'STK Push callback URL'} ) # Use the M-Pesa service from payments.services import MpesaService mpesa_service = MpesaService() # Initiate test STK push response = mpesa_service.stk_push( phone_number=phone_number, amount=int(float(amount)), account_reference=f"TEST-{timezone.now().strftime('%Y%m%d%H%M%S')}", transaction_desc="Test STK Push from Optinet Global Links" ) if response.get('ResponseCode') == '0': # Log successful test messages.success( request, f'✅ Test STK Push sent successfully to {phone_number} for KES {amount}. Customer should receive M-Pesa prompt.' ) return JsonResponse({ 'success': True, 'message': f'Test STK Push sent successfully to {phone_number}. Check your phone for the M-Pesa prompt.', 'checkout_request_id': response.get('CheckoutRequestID') }) else: error_message = response.get('errorMessage', response.get('ResponseDescription', 'Unknown error')) messages.error( request, f'❌ Failed to send test STK Push to {phone_number}. Error: {error_message}' ) return JsonResponse({ 'success': False, 'message': f'Failed to send STK Push. Error: {error_message}' }) except Exception as e: messages.error(request, f'❌ Error testing STK Push: {str(e)}') return JsonResponse({ 'success': False, 'message': f'Error testing STK Push: {str(e)}' }) return JsonResponse({'success': False, 'message': 'Invalid request method'}) @login_required @user_passes_test(is_admin) def mpesa_health_check(request): """Health check endpoint for M-Pesa configuration""" from settings.health_checks import MpesaHealthCheck health_check = MpesaHealthCheck.check_mpesa_settings() config = MpesaHealthCheck.get_effective_mpesa_config() # Mask sensitive data safe_config = config.copy() for key in ['consumer_secret', 'passkey']: if safe_config.get(key): safe_config[key] = '*' * len(safe_config[key]) return JsonResponse({ 'health_check': health_check, 'effective_config': safe_config, 'environment': config.get('environment', 'sandbox'), 'callback_urls': { 'stk_callback': config.get('stk_callback_url', ''), 'c2b_validation': config.get('c2b_validation_url', ''), 'c2b_confirmation': config.get('c2b_confirmation_url', ''), } }) @login_required @user_passes_test(is_admin) def check_mpesa_balance(request): """Check M-Pesa account balance""" if request.method == 'POST': try: # Check if M-Pesa is configured from settings.models import SystemSettings mpesa_settings = SystemSettings.get_mpesa_settings() if not mpesa_settings.get('consumer_key') or not mpesa_settings.get('consumer_secret'): return JsonResponse({ 'success': False, 'message': 'M-Pesa is not configured. Please configure your M-Pesa settings first.' }) if not mpesa_settings.get('business_short_code'): return JsonResponse({ 'success': False, 'message': 'Business Short Code is required for balance checking.' }) # Check balance from payments.services import MpesaService mpesa_service = MpesaService() response = mpesa_service.check_account_balance() print(f"M-Pesa Balance Response: {response}") if response.get('ResponseCode') == '0': # Create pending record conversation_id = response.get('ConversationID') originator_conversation_id = response.get('OriginatorConversationID') from payments.models import MpesaBalanceInquiry balance_inquiry = MpesaBalanceInquiry.objects.create( conversation_id=conversation_id, originator_conversation_id=originator_conversation_id ) return JsonResponse({ 'success': True, 'message': 'Balance inquiry initiated successfully. You will receive the results via callback.', 'conversation_id': conversation_id, 'status': 'pending' }) else: error_msg = response.get('errorMessage', response.get('ResponseDescription', 'Unknown error')) return JsonResponse({ 'success': False, 'message': f'Balance inquiry failed: {error_msg}', 'response': response }) except Exception as e: print(f"Error checking M-Pesa balance: {str(e)}") return JsonResponse({ 'success': False, 'message': f'Error checking balance: {str(e)}' }) return JsonResponse({ 'success': False, 'message': 'Invalid request method' }) @login_required @user_passes_test(is_admin) def email_logs(request): """View email logs""" email_logs = EmailLog.objects.all().order_by('-sent_at') # Filter by email type email_type = request.GET.get('type') if email_type: email_logs = email_logs.filter(email_type=email_type) # Filter by status status = request.GET.get('status') if status: email_logs = email_logs.filter(status=status) # Search by recipient email or subject search = request.GET.get('search') if search: email_logs = email_logs.filter( Q(recipient_email__icontains=search) | Q(subject__icontains=search) | Q(recipient_name__icontains=search) ) context = { 'email_logs': email_logs, 'email_types': EmailLog.EMAIL_TYPE_CHOICES, 'status_choices': EmailLog.STATUS_CHOICES, 'selected_type': request.GET.get('type', ''), 'selected_status': request.GET.get('status', ''), 'search_query': request.GET.get('search', ''), } return render(request, 'settings/email_logs.html', context) @login_required @user_passes_test(is_admin) def password_reset_logs(request): """View password reset email logs""" password_reset_logs = EmailLog.objects.filter(email_type='password_reset').order_by('-sent_at') # Filter by status status = request.GET.get('status') if status: password_reset_logs = password_reset_logs.filter(status=status) # Search by recipient email search = request.GET.get('search') if search: password_reset_logs = password_reset_logs.filter( Q(recipient_email__icontains=search) | Q(recipient_name__icontains=search) ) context = { 'password_reset_logs': password_reset_logs, 'status_choices': EmailLog.STATUS_CHOICES, 'selected_status': request.GET.get('status', ''), 'search_query': request.GET.get('search', ''), } return render(request, 'settings/password_reset_logs.html', context)