#!/usr/bin/env python3
"""
Genesis Email Notification System
===================================
Sends email notifications via SMTP or SendGrid.
PM-048: Email Notification System
- Sends via SMTP/SendGrid
- Supports HTML templates
- Logs delivery status
"""
import os
import smtplib
import logging
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import Dict, Any, List, Optional
from datetime import datetime
from dataclasses import dataclass, asdict
logger = logging.getLogger("EmailNotifier")
logging.basicConfig(level=logging.INFO)
@dataclass
class EmailResult:
"""Result of email delivery."""
email_id: str
to: str
subject: str
status: str
timestamp: str
error: Optional[str]
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
class EmailNotifier:
"""
Email notification system for Genesis.
Supports SMTP and SendGrid for email delivery.
"""
def __init__(self, smtp_host: str = None, smtp_port: int = None,
smtp_user: str = None, smtp_password: str = None,
sendgrid_key: str = None, from_email: str = None):
"""
Initialize email notifier.
Args:
smtp_host: SMTP server host
smtp_port: SMTP server port
smtp_user: SMTP username
smtp_password: SMTP password
sendgrid_key: SendGrid API key (alternative to SMTP)
from_email: Default sender email
"""
# SMTP config
self.smtp_host = smtp_host or os.getenv("SMTP_HOST", "smtp.gmail.com")
self.smtp_port = smtp_port or int(os.getenv("SMTP_PORT", 587))
self.smtp_user = smtp_user or os.getenv("SMTP_USER")
self.smtp_password = smtp_password or os.getenv("SMTP_PASSWORD")
# SendGrid config
self.sendgrid_key = sendgrid_key or os.getenv("SENDGRID_API_KEY")
# Default sender
self.from_email = from_email or os.getenv("EMAIL_FROM", "genesis@example.com")
# Email log
self.email_log: List[EmailResult] = []
self.delivered_count = 0
self.failed_count = 0
# Determine provider
self.provider = "sendgrid" if self.sendgrid_key else "smtp"
logger.info(f"Email Notifier initialized (provider: {self.provider})")
def _generate_email_id(self) -> str:
"""Generate unique email ID."""
import uuid
return f"eml_{uuid.uuid4().hex[:12]}"
def send(self, to: str, subject: str, body: str,
html_body: str = None, from_email: str = None,
reply_to: str = None, cc: List[str] = None,
bcc: List[str] = None) -> EmailResult:
"""
Send an email notification.
Args:
to: Recipient email address
subject: Email subject
body: Plain text body
html_body: Optional HTML body
from_email: Optional sender override
reply_to: Optional reply-to address
cc: Optional CC recipients
bcc: Optional BCC recipients
Returns:
EmailResult
"""
email_id = self._generate_email_id()
sender = from_email or self.from_email
# Send via appropriate provider
if self.provider == "sendgrid":
result = self._send_sendgrid(
to=to, subject=subject, body=body, html_body=html_body,
sender=sender, reply_to=reply_to, cc=cc, bcc=bcc
)
else:
result = self._send_smtp(
to=to, subject=subject, body=body, html_body=html_body,
sender=sender, reply_to=reply_to, cc=cc, bcc=bcc
)
# Create result record
email_result = EmailResult(
email_id=email_id,
to=to,
subject=subject,
status=result.get("status", "unknown"),
timestamp=datetime.utcnow().isoformat(),
error=result.get("error")
)
# Update counters
if email_result.status == "delivered":
self.delivered_count += 1
else:
self.failed_count += 1
# Log
self.email_log.append(email_result)
logger.info(f"Email [{email_id}] to {to}: {email_result.status}")
return email_result
def _send_smtp(self, to: str, subject: str, body: str,
html_body: str = None, sender: str = None,
reply_to: str = None, cc: List[str] = None,
bcc: List[str] = None) -> Dict[str, str]:
"""Send email via SMTP."""
if not self.smtp_user or not self.smtp_password:
return {"status": "skipped", "error": "SMTP credentials not configured"}
try:
# Create message
msg = MIMEMultipart("alternative")
msg["From"] = sender
msg["To"] = to
msg["Subject"] = subject
if reply_to:
msg["Reply-To"] = reply_to
if cc:
msg["Cc"] = ", ".join(cc)
# Add plain text
msg.attach(MIMEText(body, "plain"))
# Add HTML if provided
if html_body:
msg.attach(MIMEText(html_body, "html"))
# Build recipient list
recipients = [to]
if cc:
recipients.extend(cc)
if bcc:
recipients.extend(bcc)
# Send
with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
server.starttls()
server.login(self.smtp_user, self.smtp_password)
server.sendmail(sender, recipients, msg.as_string())
return {"status": "delivered"}
except smtplib.SMTPException as e:
logger.error(f"SMTP error: {e}")
return {"status": "failed", "error": str(e)}
except Exception as e:
logger.error(f"Email send failed: {e}")
return {"status": "failed", "error": str(e)}
def _send_sendgrid(self, to: str, subject: str, body: str,
html_body: str = None, sender: str = None,
reply_to: str = None, cc: List[str] = None,
bcc: List[str] = None) -> Dict[str, str]:
"""Send email via SendGrid API."""
if not self.sendgrid_key:
return {"status": "skipped", "error": "SendGrid API key not configured"}
try:
import requests
# Build SendGrid payload
payload = {
"personalizations": [{
"to": [{"email": to}]
}],
"from": {"email": sender},
"subject": subject,
"content": [
{"type": "text/plain", "value": body}
]
}
if html_body:
payload["content"].append({"type": "text/html", "value": html_body})
if reply_to:
payload["reply_to"] = {"email": reply_to}
if cc:
payload["personalizations"][0]["cc"] = [{"email": e} for e in cc]
if bcc:
payload["personalizations"][0]["bcc"] = [{"email": e} for e in bcc]
response = requests.post(
"https://api.sendgrid.com/v3/mail/send",
headers={
"Authorization": f"Bearer {self.sendgrid_key}",
"Content-Type": "application/json"
},
json=payload,
timeout=30
)
if response.status_code in [200, 202]:
return {"status": "delivered"}
else:
return {"status": "failed", "error": response.text}
except ImportError:
return {"status": "failed", "error": "requests library not available"}
except Exception as e:
logger.error(f"SendGrid error: {e}")
return {"status": "failed", "error": str(e)}
# ===== TEMPLATE HELPERS =====
def create_html_template(self, title: str, content: str,
footer: str = None) -> str:
"""Create a simple HTML email template."""
footer = footer or "Genesis System"
return f"""
"""
# ===== CONVENIENCE METHODS =====
def send_alert(self, to: str, title: str, message: str,
severity: str = "info") -> EmailResult:
"""Send an alert email."""
severity_colors = {
"info": "#36a64f",
"warning": "#ff9800",
"error": "#f44336",
"critical": "#9c27b0"
}
color = severity_colors.get(severity, "#333")
html_content = f"""
Severity: {severity.upper()}
{message}
"""
html_body = self.create_html_template(title, html_content)
return self.send(
to=to,
subject=f"[Genesis Alert] {title}",
body=f"Severity: {severity}\n\n{message}",
html_body=html_body
)
def send_report(self, to: str, title: str,
sections: Dict[str, str]) -> EmailResult:
"""Send a report email with multiple sections."""
html_content = ""
text_content = ""
for section_title, section_content in sections.items():
html_content += f"{section_title}
{section_content}
"
text_content += f"\n{section_title}\n{'=' * len(section_title)}\n{section_content}\n"
html_body = self.create_html_template(title, html_content)
return self.send(
to=to,
subject=f"[Genesis Report] {title}",
body=text_content,
html_body=html_body
)
def send_welcome(self, to: str, name: str,
custom_message: str = None) -> EmailResult:
"""Send a welcome email."""
message = custom_message or f"Welcome to Genesis, {name}! We're excited to have you on board."
html_content = f"""
Hi {name},
{message}
If you have any questions, feel free to reply to this email.
Best,
The Genesis Team
"""
html_body = self.create_html_template("Welcome to Genesis", html_content)
return self.send(
to=to,
subject="Welcome to Genesis!",
body=f"Hi {name},\n\n{message}\n\nBest,\nThe Genesis Team",
html_body=html_body
)
def send_password_reset(self, to: str, reset_link: str) -> EmailResult:
"""Send a password reset email."""
html_content = f"""
You requested a password reset for your Genesis account.
Click the link below to reset your password:
Reset Password
If you didn't request this, please ignore this email.
"""
html_body = self.create_html_template("Password Reset", html_content)
return self.send(
to=to,
subject="[Genesis] Password Reset Request",
body=f"Password Reset Link: {reset_link}\n\nIf you didn't request this, ignore this email.",
html_body=html_body
)
def get_email_log(self, limit: int = 100,
status: str = None) -> List[Dict[str, Any]]:
"""Get email log, optionally filtered by status."""
logs = self.email_log[-limit:]
if status:
logs = [e for e in logs if e.status == status]
return [e.to_dict() for e in logs]
def get_metrics(self) -> Dict[str, Any]:
"""Get notifier metrics."""
return {
"total_emails": len(self.email_log),
"delivered": self.delivered_count,
"failed": self.failed_count,
"delivery_rate": round(
self.delivered_count / max(1, len(self.email_log)) * 100, 2
),
"provider": self.provider,
"from_email": self.from_email
}
# Global instance
_notifier: Optional[EmailNotifier] = None
def get_notifier() -> EmailNotifier:
"""Get or create global email notifier."""
global _notifier
if _notifier is None:
_notifier = EmailNotifier()
return _notifier
if __name__ == "__main__":
# Self-test
notifier = EmailNotifier()
print("\n=== Email Notifier Test ===")
# Test basic send
result = notifier.send(
to="test@example.com",
subject="Test Email from Genesis",
body="This is a test email from the Genesis system."
)
print(f"Basic Send: {result.to_dict()}")
# Test alert
result = notifier.send_alert(
to="admin@example.com",
title="System Alert",
message="CPU usage exceeded 90%",
severity="warning"
)
print(f"Alert: {result.to_dict()}")
# Test report
result = notifier.send_report(
to="reports@example.com",
title="Daily Summary",
sections={
"Revenue": "$5,250.00 generated today",
"Tasks": "147 tasks completed, 3 failed",
"Leads": "25 new leads captured"
}
)
print(f"Report: {result.to_dict()}")
# Metrics
print(f"\nMetrics: {notifier.get_metrics()}")