""" Simulator views for testing external notification APIs. This module provides API endpoints that simulate external email and SMS services: - Email simulator: Sends real emails via Django SMTP - SMS simulator: Prints messages to terminal with formatted output """ import logging from datetime import datetime from django.conf import settings from django.core.mail import send_mail from django.http import JsonResponse from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods import json logger = logging.getLogger(__name__) # Request counter for tracking request_counter = {'email': 0, 'sms': 0} # Request history (last 10 requests) request_history = [] def log_simulator_request(channel, payload, status): """Log simulator request to history and file.""" request_id = len(request_history) + 1 entry = { 'id': request_id, 'channel': channel, 'timestamp': datetime.now().isoformat(), 'status': status, 'payload': payload } request_history.append(entry) # Keep only last 10 requests if len(request_history) > 10: request_history.pop(0) # Log to file logger.info(f"[Simulator] {channel.upper()} Request #{request_id}: {status}") @csrf_exempt @require_http_methods(["POST"]) def email_simulator(request): """ Simulate external email API endpoint. Accepts POST request with JSON payload: { "to": "recipient@example.com", "subject": "Email subject", "message": "Plain text message", "html_message": "Optional HTML content" } Sends real email via Django SMTP and returns 200 OK. """ request_counter['email'] += 1 try: # Parse request body data = json.loads(request.body) # Validate required fields required_fields = ['to', 'subject', 'message'] missing_fields = [field for field in required_fields if field not in data] if missing_fields: response = { 'success': False, 'error': f'Missing required fields: {", ".join(missing_fields)}' } log_simulator_request('email', data, 'failed') return JsonResponse(response, status=400) # Extract fields to_email = data['to'] subject = data['subject'] message = data['message'] html_message = data.get('html_message', None) # Log the request logger.info(f"[Email Simulator] Sending email to {to_email}: {subject}") # Print formatted email to terminal print(f"\n{'╔' + '═'*68 + '╗'}") print(f"║{' ' * 15}📧 EMAIL SIMULATOR{' ' * 34}║") print(f"╠{'═'*68}╣") print(f"║ Request #: {request_counter['email']:<52} ║") print(f"╠{'═'*68}╣") print(f"║ To: {to_email:<52} ║") print(f"║ Subject: {subject[:52]:<52} ║") print(f"╠{'═'*68}╣") print(f"║ Message:{' '*55} ║") # Word wrap message words = message.split() line = "║ " for word in words: if len(line) + len(word) + 1 > 68: print(f"{line:<68} ║") line = "║ " + word else: if line == "║ ": line += word else: line += " " + word # Print last line if not empty if line != "║ ": print(f"{line:<68} ║") # Print HTML section if present if html_message: print(f"╠{'═'*68}╣") print(f"║ HTML Message:{' '*48} ║") html_preview = html_message[:200].replace('\n', ' ') print(f"║ {html_preview:<66} ║") if len(html_message) > 200: print(f"║ ... ({len(html_message)} total characters){' '*30} ║") print(f"╚{'═'*68}╝\n") # Send real email via Django SMTP try: send_mail( subject=subject, message=message, from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[to_email], html_message=html_message, fail_silently=False ) logger.info(f"[Email Simulator] Email sent via SMTP to {to_email}") email_status = 'sent' except Exception as email_error: logger.error(f"[Email Simulator] SMTP Error: {str(email_error)}") # Log as 'partial' since we displayed it but failed to send email_status = 'partial' # Return error response response = { 'success': False, 'error': f'Email displayed but failed to send via SMTP: {str(email_error)}', 'data': { 'to': to_email, 'subject': subject, 'message_length': len(message), 'has_html': html_message is not None, 'displayed': True, 'sent': False } } log_simulator_request('email', data, email_status) return JsonResponse(response, status=500) # Log success logger.info(f"[Email Simulator] Email sent successfully to {to_email}") log_simulator_request('email', data, email_status) response = { 'success': True, 'message': 'Email sent successfully', 'data': { 'to': to_email, 'subject': subject, 'message_length': len(message), 'has_html': html_message is not None } } return JsonResponse(response, status=200) except json.JSONDecodeError: response = { 'success': False, 'error': 'Invalid JSON format' } log_simulator_request('email', {}, 'failed') return JsonResponse(response, status=400) except Exception as e: logger.error(f"[Email Simulator] Error: {str(e)}") response = { 'success': False, 'error': str(e) } log_simulator_request('email', data if 'data' in locals() else {}, 'failed') return JsonResponse(response, status=500) @csrf_exempt @require_http_methods(["POST"]) def sms_simulator(request): """ Simulate external SMS API endpoint. Accepts POST request with JSON payload: { "to": "+966501234567", "message": "SMS message text" } Prints SMS to terminal with formatted output and returns 200 OK. """ request_counter['sms'] += 1 try: # Parse request body data = json.loads(request.body) # Validate required fields required_fields = ['to', 'message'] missing_fields = [field for field in required_fields if field not in data] if missing_fields: response = { 'success': False, 'error': f'Missing required fields: {", ".join(missing_fields)}' } log_simulator_request('sms', data, 'failed') return JsonResponse(response, status=400) # Extract fields to_phone = data['to'] message = data['message'] timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') # Log the request logger.info(f"[SMS Simulator] Sending SMS to {to_phone}") # Print formatted SMS to terminal print(f"\n{'╔' + '═'*68 + '╗'}") print(f"║{' ' * 15}📱 SMS SIMULATOR{' ' * 35}║") print(f"╠{'═'*68}╣") print(f"║ Request #: {request_counter['sms']:<52} ║") print(f"╠{'═'*68}╣") print(f"║ To: {to_phone:<52} ║") print(f"║ Time: {timestamp:<52} ║") print(f"╠{'═'*68}╣") print(f"║ Message:{' '*55} ║") # Word wrap message words = message.split() line = "║ " for word in words: if len(line) + len(word) + 1 > 68: print(f"{line:<68} ║") line = "║ " + word else: if line == "║ ": line += word else: line += " " + word # Print last line if not empty if line != "║ ": print(f"{line:<68} ║") print(f"╚{'═'*68}╝\n") # Log success logger.info(f"[SMS Simulator] SMS sent to {to_phone}: {message[:50]}...") log_simulator_request('sms', data, 'sent') response = { 'success': True, 'message': 'SMS sent successfully', 'data': { 'to': to_phone, 'message_length': len(message) } } return JsonResponse(response, status=200) except json.JSONDecodeError: response = { 'success': False, 'error': 'Invalid JSON format' } log_simulator_request('sms', {}, 'failed') return JsonResponse(response, status=400) except Exception as e: logger.error(f"[SMS Simulator] Error: {str(e)}") response = { 'success': False, 'error': str(e) } log_simulator_request('sms', data if 'data' in locals() else {}, 'failed') return JsonResponse(response, status=500) @csrf_exempt @require_http_methods(["GET"]) def health_check(request): """ Health check endpoint for simulator. Returns simulator status and statistics. """ return JsonResponse({ 'status': 'healthy', 'timestamp': datetime.now().isoformat(), 'statistics': { 'total_requests': request_counter['email'] + request_counter['sms'], 'email_requests': request_counter['email'], 'sms_requests': request_counter['sms'] }, 'recent_requests': request_history[-5:] # Last 5 requests }, status=200) @csrf_exempt @require_http_methods(["GET"]) def reset_simulator(request): """ Reset simulator statistics and history. Clears request counter and history. """ global request_counter, request_history request_counter = {'email': 0, 'sms': 0} request_history = [] logger.info("[Simulator] Reset statistics and history") return JsonResponse({ 'success': True, 'message': 'Simulator reset successfully' }, status=200)