336 lines
11 KiB
Python
336 lines
11 KiB
Python
"""
|
|
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)
|