import aiosmtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart import logging from typing import Dict, Optional import os logger = logging.getLogger(__name__) class NotificationManager: """Manage notifications (Email, SMS, etc.)""" def __init__(self): self.smtp_host = os.getenv('SMTP_HOST', 'smtp.gmail.com') self.smtp_port = int(os.getenv('SMTP_PORT', 587)) self.smtp_user = os.getenv('SMTP_USER', '') self.smtp_password = os.getenv('SMTP_PASSWORD', '') self.from_email = os.getenv('FROM_EMAIL', self.smtp_user) async def send_email(self, to_email: str, subject: str, body: str) -> bool: """Send email notification""" if not self.smtp_user or not self.smtp_password: logger.warning("SMTP credentials not configured") return False try: message = MIMEMultipart() message['From'] = self.from_email message['To'] = to_email message['Subject'] = subject message.attach(MIMEText(body, 'html')) await aiosmtplib.send( message, hostname=self.smtp_host, port=self.smtp_port, username=self.smtp_user, password=self.smtp_password, start_tls=True ) logger.info(f"Email sent to {to_email}") return True except Exception as e: logger.error(f"Failed to send email: {e}") return False async def send_alert_email(self, alert: Dict, to_emails: list) -> bool: """Send alert notification email""" severity_colors = { 'critical': '#ef4444', 'warning': '#f59e0b', 'info': '#3b82f6' } color = severity_colors.get(alert.get('severity', 'info'), '#3b82f6') subject = f"[{alert['severity'].upper()}] {alert['message']}" body = f"""

Network Alert

Device: {alert.get('hostname', alert.get('ip', 'Unknown'))}

IP: {alert.get('ip', 'N/A')}

Type: {alert.get('type', 'N/A')}

Severity: {alert['severity'].upper()}

Message: {alert['message']}

Time: {alert.get('timestamp', 'N/A')}

This is an automated message from Network Management System.

""" success = True for email in to_emails: result = await self.send_email(email, subject, body) if not result: success = False return success async def send_telegram_notification(self, bot_token: str, chat_id: str, message: str) -> bool: """Send Telegram notification""" try: import aiohttp url = f"https://api.telegram.org/bot{bot_token}/sendMessage" data = { 'chat_id': chat_id, 'text': message, 'parse_mode': 'HTML' } async with aiohttp.ClientSession() as session: async with session.post(url, json=data) as response: if response.status == 200: logger.info("Telegram notification sent") return True else: logger.error(f"Telegram API error: {response.status}") return False except Exception as e: logger.error(f"Failed to send Telegram notification: {e}") return False notification_manager = NotificationManager()