import requests from django.core.management.base import BaseCommand from django.utils import timezone from settings.models import SystemSettings from payments.services import MpesaService class Command(BaseCommand): help = 'Comprehensive M-Pesa error checking and diagnostics' def handle(self, *args, **options): self.stdout.write("=" * 60) self.stdout.write("M-PESA COMPREHENSIVE ERROR DIAGNOSTICS") self.stdout.write("=" * 60) # 1. Check Configuration self.check_configuration() # 2. Check Network Connectivity self.check_connectivity() # 3. Check API Access self.check_api_access() # 4. Check C2B Registration Capability self.check_c2b_capability() # 5. Check URL Configuration self.check_url_configuration() # 6. Summary and Recommendations self.provide_recommendations() def check_configuration(self): self.stdout.write("\n" + "1. CONFIGURATION CHECK") self.stdout.write("-" * 30) mpesa_settings = SystemSettings.get_mpesa_settings() # Check required settings required_settings = [ 'consumer_key', 'consumer_secret', 'business_short_code', 'passkey', 'environment' ] missing_settings = [] for setting in required_settings: value = mpesa_settings.get(setting, '') if not value or not value.strip(): missing_settings.append(setting) else: # Mask sensitive data if setting in ['consumer_secret', 'passkey']: display_value = '*' * min(len(value), 10) else: display_value = value self.stdout.write(f" ✅ {setting}: {display_value}") if missing_settings: self.stdout.write( self.style.ERROR(f" ❌ Missing: {', '.join(missing_settings)}") ) return False else: self.stdout.write(self.style.SUCCESS(" ✅ All required settings configured")) return True def check_connectivity(self): self.stdout.write("\n" + "2. NETWORK CONNECTIVITY CHECK") self.stdout.write("-" * 30) test_urls = [ 'https://www.google.com', 'https://api.safaricom.co.ke', 'https://sandbox.safaricom.co.ke' ] for url in test_urls: try: response = requests.get(url, timeout=10) if response.status_code == 200: self.stdout.write(f" ✅ {url}: OK") else: self.stdout.write(f" ⚠️ {url}: {response.status_code}") except Exception as e: self.stdout.write(f" ❌ {url}: {str(e)}") def check_api_access(self): self.stdout.write("\n" + "3. API ACCESS CHECK") self.stdout.write("-" * 30) mpesa_service = MpesaService() self.stdout.write(f" Environment: {mpesa_service.environment}") self.stdout.write(f" Base URL: {mpesa_service.base_url}") # Test access token access_token = mpesa_service.get_access_token() if access_token: self.stdout.write(self.style.SUCCESS(" ✅ Access token obtained successfully")) self.stdout.write(f" Token preview: {access_token[:15]}...") # Test token validity by making a test API call test_url = f"{mpesa_service.base_url}/mpesa/accountbalance/v1/query" headers = {'Authorization': f'Bearer {access_token}'} try: test_response = requests.post(test_url, headers=headers, json={}, timeout=15) self.stdout.write(f" API Test Response: {test_response.status_code}") if test_response.status_code == 401: response_text = test_response.text if "Invalid API call as no apiproduct match found" in response_text: self.stdout.write( self.style.ERROR(" ❌ CRITICAL: Your M-Pesa app lacks required API products") ) self.stdout.write(" This means your app is not properly configured on Safaricom Developer Portal") else: self.stdout.write( self.style.ERROR(f" ❌ Authentication issue: {response_text[:100]}") ) elif test_response.status_code == 400: self.stdout.write( self.style.SUCCESS(" ✅ API is accessible (400 expected with empty payload)") ) else: self.stdout.write(f" Response: {test_response.text[:200]}") except Exception as e: self.stdout.write(f" ⚠️ API test failed: {str(e)}") else: self.stdout.write(self.style.ERROR(" ❌ Failed to get access token")) def check_c2b_capability(self): self.stdout.write("\n" + "4. C2B REGISTRATION CAPABILITY CHECK") self.stdout.write("-" * 30) mpesa_service = MpesaService() # Test C2B URL registration with dummy URLs test_validation_url = "https://test.example.com/validation/" test_confirmation_url = "https://test.example.com/confirmation/" try: response = mpesa_service.register_c2b_urls( test_validation_url, test_confirmation_url ) self.stdout.write(f" Registration Response: {response}") if response.get('ResponseCode') == '0': self.stdout.write(self.style.SUCCESS(" ✅ C2B registration capability confirmed")) else: error_msg = response.get('errorMessage', response.get('ResponseDescription', 'Unknown')) if 'Invalid Access Token' in error_msg or 'Invalid API call' in error_msg: self.stdout.write( self.style.ERROR(" ❌ CRITICAL: C2B API not available for your app") ) self.stdout.write(" Your M-Pesa app needs C2B API product enabled") else: self.stdout.write(f" ⚠️ C2B registration issue: {error_msg}") except Exception as e: self.stdout.write(f" ❌ C2B test failed: {str(e)}") def check_url_configuration(self): self.stdout.write("\n" + "5. URL CONFIGURATION CHECK") self.stdout.write("-" * 30) import os # Check Replit domain domains = os.environ.get("REPLIT_DOMAINS", "").split(',') if domains and domains[0]: base_url = f'https://{domains[0]}' self.stdout.write(f" Base URL: {base_url}") # Check if URLs are accessible validation_url = f"{base_url}/payments/mpesa-paybill-validation/" confirmation_url = f"{base_url}/payments/mpesa-paybill-confirmation/" for url_name, url in [("Validation", validation_url), ("Confirmation", confirmation_url)]: try: response = requests.get(url, timeout=10) if response.status_code == 405: # Method not allowed is expected for POST endpoints self.stdout.write(f" ✅ {url_name} URL accessible: {url}") else: self.stdout.write(f" ⚠️ {url_name} URL status {response.status_code}: {url}") except Exception as e: self.stdout.write(f" ❌ {url_name} URL not accessible: {str(e)}") else: self.stdout.write(" ⚠️ REPLIT_DOMAINS not found") def provide_recommendations(self): self.stdout.write("\n" + "6. RECOMMENDATIONS") self.stdout.write("-" * 30) mpesa_settings = SystemSettings.get_mpesa_settings() self.stdout.write("\n📋 NEXT STEPS TO RESOLVE M-PESA ISSUES:") self.stdout.write("\n1. SAFARICOM DEVELOPER PORTAL:") self.stdout.write(" • Log into https://developer.safaricom.co.ke") self.stdout.write(" • Go to your production app (Prod-optinet-global-links-1703160740)") self.stdout.write(" • Check 'APIs' section and ensure these are enabled:") self.stdout.write(" - M-Pesa Express (STK Push)") self.stdout.write(" - C2B API") self.stdout.write(" - Transaction Status API") self.stdout.write("\n2. API PRODUCTS:") self.stdout.write(" • If C2B API is missing, click 'Add API' and select 'C2B API'") self.stdout.write(" • Wait for approval (can take 24-48 hours)") self.stdout.write(" • Regenerate credentials if needed") self.stdout.write("\n3. CREDENTIALS VERIFICATION:") consumer_key = mpesa_settings.get('consumer_key', '') if consumer_key: self.stdout.write(f" • Your Consumer Key: {consumer_key}") self.stdout.write(f" • Environment: {mpesa_settings.get('environment', 'sandbox')}") self.stdout.write(" • Ensure credentials match the selected environment") self.stdout.write("\n4. TESTING SEQUENCE:") self.stdout.write(" • python manage.py test_mpesa_config") self.stdout.write(" • python manage.py register_paybill_urls --base-url [YOUR_URL]") self.stdout.write(" • Test STK Push functionality") self.stdout.write("\n" + "=" * 60) self.stdout.write("DIAGNOSIS COMPLETE") self.stdout.write("=" * 60)