""" Email notification service for critical errors and system failures. This module provides automatic email notifications when critical errors occur. """ import smtplib import ssl import logging from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.mime.base import MIMEBase from email import encoders from datetime import datetime from typing import Dict, Any, Optional, List import os import traceback import json from config.settings import ( EMAIL_SMTP_SERVER, EMAIL_SMTP_PORT, EMAIL_USERNAME, EMAIL_PASSWORD, EMAIL_RECIPIENTS, EMAIL_FROM_ADDRESS, EMAIL_USE_TLS, EMAIL_USE_SSL ) logger = logging.getLogger(__name__) class EmailNotifier: """Email notification service for critical errors and system alerts""" def __init__(self): """Initialize the email notifier with SMTP configuration""" self.smtp_server = EMAIL_SMTP_SERVER self.smtp_port = EMAIL_SMTP_PORT self.username = EMAIL_USERNAME self.password = EMAIL_PASSWORD self.recipients = EMAIL_RECIPIENTS self.from_address = EMAIL_FROM_ADDRESS self.use_tls = EMAIL_USE_TLS self.use_ssl = EMAIL_USE_SSL # Validate configuration self._validate_config() def _validate_config(self): """Validate email configuration""" required_configs = [ ('SMTP_SERVER', self.smtp_server), ('SMTP_PORT', self.smtp_port), ('USERNAME', self.username), ('PASSWORD', self.password), ('RECIPIENTS', self.recipients), ('FROM_ADDRESS', self.from_address) ] missing_configs = [] for name, value in required_configs: if not value: missing_configs.append(name) if missing_configs: logger.warning(f"Email notification disabled - missing configuration: {', '.join(missing_configs)}") self.enabled = False else: self.enabled = True logger.info("Email notification service initialized successfully") def send_critical_error_notification(self, error_type: str, error_message: str, context: Optional[Dict[str, Any]] = None, exception: Optional[Exception] = None, user_context: Optional[Dict[str, str]] = None, recipients: Optional[List[str]] = None) -> bool: """ Send critical error notification email Args: error_type: Type of error (API_FAILURE, SYSTEM_ERROR, etc.) error_message: Error message context: Additional context data exception: Exception object if available user_context: User information if available Returns: bool: True if email sent successfully, False otherwise """ if not self.enabled: logger.warning("Email notifications disabled - skipping critical error notification") return False try: subject = f"๐Ÿšจ CRITICAL ERROR ALERT - {error_type} - AI Test Case Generator" # Create email body body = self._create_error_email_body( error_type, error_message, context, exception, user_context ) # Send email return self._send_email(subject, body, is_html=True, recipients=recipients) except Exception as e: logger.error(f"Failed to send critical error notification: {str(e)}") return False def send_api_failure_notification(self, api_name: str, endpoint: str, error_message: str, status_code: Optional[int] = None, response_data: Optional[Dict] = None, request_data: Optional[Dict] = None) -> bool: """ Send API failure notification email Args: api_name: Name of the API (OpenAI, Jira, Azure, etc.) endpoint: API endpoint that failed error_message: Error message status_code: HTTP status code if available response_data: API response data if available request_data: Request data if available Returns: bool: True if email sent successfully, False otherwise """ if not self.enabled: logger.warning("Email notifications disabled - skipping API failure notification") return False try: subject = f"โš ๏ธ API FAILURE - {api_name} - AI Test Case Generator" # Create email body body = self._create_api_failure_email_body( api_name, endpoint, error_message, status_code, response_data, request_data ) # Send email return self._send_email(subject, body, is_html=True) except Exception as e: logger.error(f"Failed to send API failure notification: {str(e)}") return False def send_system_alert(self, alert_type: str, message: str, severity: str = "WARNING", context: Optional[Dict[str, Any]] = None) -> bool: """ Send system alert notification Args: alert_type: Type of alert (PERFORMANCE, SECURITY, etc.) message: Alert message severity: Alert severity (INFO, WARNING, CRITICAL) context: Additional context data Returns: bool: True if email sent successfully, False otherwise """ if not self.enabled: logger.warning("Email notifications disabled - skipping system alert") return False try: emoji = "๐Ÿ”ด" if severity == "CRITICAL" else "๐ŸŸก" if severity == "WARNING" else "๐Ÿ”ต" subject = f"{emoji} SYSTEM ALERT - {alert_type} - AI Test Case Generator" # Create email body body = self._create_system_alert_email_body(alert_type, message, severity, context) # Send email return self._send_email(subject, body, is_html=True) except Exception as e: logger.error(f"Failed to send system alert: {str(e)}") return False def _create_error_email_body(self, error_type: str, error_message: str, context: Optional[Dict[str, Any]] = None, exception: Optional[Exception] = None, user_context: Optional[Dict[str, str]] = None) -> str: """Create HTML email body for critical errors""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S UTC") # Get traceback if exception provided traceback_info = "" if exception: try: traceback_info = ''.join(traceback.format_exception( type(exception), exception, exception.__traceback__ )) except: traceback_info = "Unable to extract traceback" # Format context data context_html = "" if context: context_html = f"""

๐Ÿ“‹ Additional Context:

{json.dumps(context, indent=2)}
""" # Format user context user_html = "" if user_context: user_html = f"""

๐Ÿ‘ค User Information:

""" # Format traceback traceback_html = "" if traceback_info: traceback_html = f"""

๐Ÿ” Stack Trace:

{traceback_info}
""" html_body = f""" Critical Error Alert

๐Ÿšจ CRITICAL ERROR ALERT

AI Test Case Generator System

Error Details

โฐ Timestamp: {timestamp}

๐Ÿ”ง Error Type: {error_type}

๐Ÿ“ Error Message:

{error_message}
{user_html} {context_html} {traceback_html}

๐Ÿ”ง Recommended Actions:

This is an automated alert from the AI Test Case Generator monitoring system.

""" return html_body def _create_api_failure_email_body(self, api_name: str, endpoint: str, error_message: str, status_code: Optional[int] = None, response_data: Optional[Dict] = None, request_data: Optional[Dict] = None) -> str: """Create HTML email body for API failures""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S UTC") # Format response data response_html = "" if response_data: response_html = f"""

๐Ÿ“ค API Response:

{json.dumps(response_data, indent=2)}
""" # Format request data request_html = "" if request_data: request_html = f"""

๐Ÿ“ฅ Request Data:

{json.dumps(request_data, indent=2)}
""" html_body = f""" API Failure Alert

โš ๏ธ API FAILURE ALERT

AI Test Case Generator System

API Failure Details

โฐ Timestamp: {timestamp}

๐Ÿ”Œ API Name: {api_name}

๐ŸŒ Endpoint: {endpoint}

{f'

๐Ÿ“Š Status Code: {status_code}

' if status_code else ''}

๐Ÿ“ Error Message:

{error_message}
{request_html} {response_html}

๐Ÿ”ง Recommended Actions:

This is an automated alert from the AI Test Case Generator monitoring system.

""" return html_body def _create_system_alert_email_body(self, alert_type: str, message: str, severity: str, context: Optional[Dict[str, Any]] = None) -> str: """Create HTML email body for system alerts""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S UTC") # Set color based on severity color = "#dc3545" if severity == "CRITICAL" else "#fd7e14" if severity == "WARNING" else "#17a2b8" emoji = "๐Ÿ”ด" if severity == "CRITICAL" else "๐ŸŸก" if severity == "WARNING" else "๐Ÿ”ต" # Format context data context_html = "" if context: context_html = f"""

๐Ÿ“‹ Additional Context:

{json.dumps(context, indent=2)}
""" html_body = f""" System Alert

{emoji} SYSTEM ALERT

AI Test Case Generator System

Alert Details

โฐ Timestamp: {timestamp}

๐Ÿšจ Alert Type: {alert_type}

โšก Severity: {severity}

๐Ÿ“ Message:

{message}

{context_html}

๐Ÿ”ง Recommended Actions:

This is an automated alert from the AI Test Case Generator monitoring system.

""" return html_body def _send_email(self, subject: str, body: str, is_html: bool = True, recipients: List[str] = None) -> bool: """ Send email using SMTP Args: subject: Email subject body: Email body is_html: Whether the body is HTML format recipients: List of email recipients (defaults to configured recipients) Returns: bool: True if email sent successfully, False otherwise """ try: # Use provided recipients or default to configured recipients email_recipients = recipients if recipients else self.recipients if not email_recipients: logger.error("No email recipients specified") return False # Create message msg = MIMEMultipart('alternative') msg['Subject'] = subject msg['From'] = self.from_address msg['To'] = ', '.join(email_recipients) # Add body if is_html: msg.attach(MIMEText(body, 'html')) else: msg.attach(MIMEText(body, 'plain')) # Create SMTP session if self.use_ssl: context = ssl.create_default_context() server = smtplib.SMTP_SSL(self.smtp_server, self.smtp_port, context=context) else: server = smtplib.SMTP(self.smtp_server, self.smtp_port) if self.use_tls: server.starttls() # Login and send email server.login(self.username, self.password) server.send_message(msg) server.quit() logger.info(f"Email sent successfully to {', '.join(email_recipients)}") return True except Exception as e: logger.error(f"Failed to send email: {str(e)}") return False def send_password_reset_email(self, email: str, reset_token: str, expires_at: datetime) -> bool: """ Send password reset email Args: email: User's email address reset_token: Password reset token expires_at: Token expiration time Returns: bool: True if email sent successfully, False otherwise """ if not self.enabled: logger.warning("Email notifications disabled - cannot send password reset email") return False try: from config.settings import BASE_URL subject = "๐Ÿ” Password Reset Request - AI Test Case Generator" # Create reset link reset_link = f"{BASE_URL}/reset-password-confirm?token={reset_token}" # Format expiration time expires_str = expires_at.strftime("%B %d, %Y at %I:%M %p UTC") # Create email body body = self._create_password_reset_email_body(email, reset_link, expires_str) # Send email to the specific user return self._send_email(subject, body, is_html=True, recipients=[email]) except Exception as e: logger.error(f"Failed to send password reset email: {str(e)}") return False def _create_password_reset_email_body(self, email: str, reset_link: str, expires_str: str) -> str: """Create HTML email body for password reset""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S UTC") html_body = f""" Password Reset Request

๐Ÿ” Password Reset Request

AI Test Case Generator

Reset Your Password

Hello,

We received a request to reset your password for your AI Test Case Generator account associated with {email}.

Click the button below to reset your password:

Reset My Password

Or copy and paste this link into your browser:

{reset_link}

โš ๏ธ Important Security Information

๐Ÿ”’ Security Tips

This email was sent on {timestamp}
If you have any questions, please contact our support team.

""" return html_body def test_email_configuration(self) -> bool: """ Test email configuration by sending a test email Returns: bool: True if test email sent successfully, False otherwise """ if not self.enabled: logger.warning("Email notifications disabled - cannot test configuration") return False try: subject = "๐Ÿงช Email Configuration Test - AI Test Case Generator" body = f""" Email Test

โœ… Email Configuration Test Successful

Timestamp: {datetime.now().strftime("%Y-%m-%d %H:%M:%S UTC")}

Service: AI Test Case Generator

Status: Email notifications are working correctly

This is a test email to verify email notification configuration.

""" return self._send_email(subject, body, is_html=True) except Exception as e: logger.error(f"Failed to send test email: {str(e)}") return False # Global email notifier instance email_notifier = EmailNotifier() # Convenience functions def send_critical_error_notification(error_type: str, error_message: str, context: Optional[Dict[str, Any]] = None, exception: Optional[Exception] = None, user_context: Optional[Dict[str, str]] = None, recipients: Optional[List[str]] = None) -> bool: """Send critical error notification email""" return email_notifier.send_critical_error_notification( error_type, error_message, context, exception, user_context, recipients ) def send_api_failure_notification(api_name: str, endpoint: str, error_message: str, status_code: Optional[int] = None, response_data: Optional[Dict] = None, request_data: Optional[Dict] = None) -> bool: """Send API failure notification email""" return email_notifier.send_api_failure_notification( api_name, endpoint, error_message, status_code, response_data, request_data ) def send_system_alert(alert_type: str, message: str, severity: str = "WARNING", context: Optional[Dict[str, Any]] = None) -> bool: """Send system alert notification email""" return email_notifier.send_system_alert(alert_type, message, severity, context) def send_password_reset_email(email: str, reset_token: str, expires_at: datetime) -> bool: """Send password reset email""" return email_notifier.send_password_reset_email(email, reset_token, expires_at) def test_email_configuration() -> bool: """Test email configuration""" return email_notifier.test_email_configuration()