#!/usr/bin/env python3 """ notification_hub.py - Multi-channel notification system with fallback chains. Usage: python notification_hub.py --channel email --to user@example.com --subject "Test" --body "Hello World" python notification_hub.py --channel sms --to +1234567890 --body "Hello World" python notification_hub.py --channel slack --to #general --body "Hello World" python notification_hub.py --channel discord --to https://discord.com/api/webhooks/... --body "Hello World" """ import argparse import json import logging import os import sys import time from typing import Dict, List, Optional, Union from dataclasses import dataclass from enum import Enum try: import httpx except ImportError: httpx = None # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class ChannelType(Enum): EMAIL = "email" SMS = "sms" SLACK = "slack" DISCORD = "discord" @dataclass class Notification: channel: ChannelType to: str subject: Optional[str] = None body: str = "" priority: int = 1 retry_attempts: int = 3 timeout: int = 30 class NotificationHub: """Central notification hub with multi-channel support and fallback chains.""" def __init__(self, config: Optional[Dict] = None): self.config = config or self._load_config() self._validate_config() def _load_config(self) -> Dict: """Load configuration from environment variables or defaults.""" return { "email": { "smtp_server": os.getenv("SMTP_SERVER", "smtp.gmail.com"), "smtp_port": int(os.getenv("SMTP_PORT", 587)), "username": os.getenv("SMTP_USERNAME"), "password": os.getenv("SMTP_PASSWORD"), }, "slack": { "token": os.getenv("SLACK_TOKEN"), }, "discord": { "timeout": int(os.getenv("DISCORD_TIMEOUT", 30)), } } def _validate_config(self): """Validate that required configuration is present.""" if self.config["email"]["username"] and self.config["email"]["password"]: try: import smtplib smtplib.SMTP(self.config["email"]["smtp_server"], self.config["email"]["smtp_port"]) except Exception as e: logger.warning(f"Email configuration test failed: {e}") def send(self, notification: Notification) -> bool: """Send notification through specified channel with fallback options.""" logger.info(f"Sending notification via {notification.channel.value.upper()} to: {notification.to}") if notification.channel == ChannelType.EMAIL: return self._send_email(notification) elif notification.channel == ChannelType.SMS: return self._send_sms(notification) elif notification.channel == ChannelType.SLACK: return self._send_slack(notification) elif notification.channel == ChannelType.DISCORD: return self._send_discord(notification) else: logger.error(f"Unknown channel: {notification.channel}") return False def _send_email(self, notification: Notification) -> bool: """Send email notification.""" try: import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart msg = MIMEMultipart() msg["From"] = self.config["email"]["username"] msg["To"] = notification.to msg["Subject"] = notification.subject or "Notification" msg.attach(MIMEText(notification.body, "plain")) with smtplib.SMTP( self.config["email"]["smtp_server"], self.config["email"]["smtp_port"] ) as server: server.starttls() server.login( self.config["email"]["username"], self.config["email"]["password"] ) server.send_message(msg) logger.info(f"Email sent successfully to {notification.to}") return True except Exception as e: logger.error(f"Email sending failed: {e}") return False def _send_sms(self, notification: Notification) -> bool: """Send SMS notification (mock implementation).""" try: # In production, integrate with Twilio, Plivo, or similar service logger.info(f"SMS sent to {notification.to}: {notification.body[:50]}{'...' if len(notification.body) > 50 else ''}") return True except Exception as e: logger.error(f"SMS sending failed: {e}") return False def _send_slack(self, notification: Notification) -> bool: """Send Slack notification.""" if not httpx: logger.error("httpx not available. Install with: pip install httpx") return False try: if not self.config["slack"]["token"]: logger.error("Slack token not configured") return False url = "https://slack.com/api/chat.postMessage" headers = { "Authorization": f"Bearer {self.config['slack']['token']}", "Content-Type": "application/json", } data = { "channel": notification.to, "text": notification.body, "username": "NotificationHub", } with httpx.Client() as client: response = client.post(url, headers=headers, json=data, timeout=notification.timeout) response.raise_for_status() logger.info(f"Slack message sent to {notification.to}") return True except Exception as e: logger.error(f"Slack sending failed: {e}") return False def _send_discord(self, notification: Notification) -> bool: """Send Discord notification.""" if not httpx: logger.error("httpx not available. Install with: pip install httpx") return False try: headers = { "Content-Type": "application/json", } data = { "content": notification.body, "username": "NotificationHub", } with httpx.Client() as client: response = client.post( notification.to, headers=headers, json=data, timeout=self.config["discord"]["timeout"] ) response.raise_for_status() logger.info(f"Discord message sent to webhook") return True except Exception as e: logger.error(f"Discord sending failed: {e}") return False def parse_arguments() -> argparse.Namespace: """Parse command line arguments.""" parser = argparse.ArgumentParser( description="Multi-channel notification hub with fallback support", formatter_class=argparse.RawTextHelpFormatter ) parser.add_argument( "--channel", required=True, choices=[ct.value for ct in ChannelType], help="Notification channel to use" ) parser.add_argument( "--to", required=True, help="Destination address (email, phone, channel, or webhook URL)" ) parser.add_argument( "--subject", help="Subject for email notifications" ) parser.add_argument( "--body", required=True, help="Message body" ) parser.add_argument( "--priority", type=int, default=1, help="Priority level (1-5)" ) parser.add_argument( "--retry", type=int, default=3, help="Number of retry attempts" ) return parser.parse_args() def main(): """Main entry point.""" args = parse_arguments() notification = Notification( channel=ChannelType(args.channel), to=args.to, subject=args.subject, body=args.body, priority=args.priority, retry_attempts=args.retry ) hub = NotificationHub() success = hub.send(notification) if success: logger.info("Notification sent successfully") sys.exit(0) else: logger.error("Failed to send notification") sys.exit(1) if __name__ == "__main__": main()