360 lines
11 KiB
Python
360 lines
11 KiB
Python
"""Email service for sending alerts and reports.
|
|
|
|
This module provides email functionality for the AlpineBits application,
|
|
including error alerts and daily reports.
|
|
"""
|
|
|
|
import asyncio
|
|
import smtplib
|
|
import ssl
|
|
from datetime import datetime
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.mime.text import MIMEText
|
|
from typing import Any
|
|
|
|
from pydantic import EmailStr, Field, field_validator
|
|
|
|
from .logging_config import get_logger
|
|
|
|
_LOGGER = get_logger(__name__)
|
|
|
|
|
|
class EmailConfig:
|
|
"""Configuration for email service.
|
|
|
|
Attributes:
|
|
smtp_host: SMTP server hostname
|
|
smtp_port: SMTP server port
|
|
smtp_username: SMTP authentication username
|
|
smtp_password: SMTP authentication password
|
|
use_tls: Use STARTTLS for encryption
|
|
use_ssl: Use SSL/TLS from the start
|
|
from_address: Sender email address
|
|
from_name: Sender display name
|
|
timeout: Connection timeout in seconds
|
|
|
|
"""
|
|
|
|
def __init__(self, config: dict[str, Any]):
|
|
"""Initialize email configuration from config dict.
|
|
|
|
Args:
|
|
config: Email configuration dictionary
|
|
|
|
"""
|
|
smtp_config = config.get("smtp", {})
|
|
self.smtp_host: str = smtp_config.get("host", "localhost")
|
|
self.smtp_port: int = smtp_config.get("port", 587)
|
|
self.smtp_username: str | None = smtp_config.get("username")
|
|
self.smtp_password: str | None = smtp_config.get("password")
|
|
self.use_tls: bool = smtp_config.get("use_tls", True)
|
|
self.use_ssl: bool = smtp_config.get("use_ssl", False)
|
|
self.from_address: str = config.get("from_address", "noreply@example.com")
|
|
self.from_name: str = config.get("from_name", "AlpineBits Server")
|
|
self.timeout: int = config.get("timeout", 10)
|
|
|
|
# Validate configuration
|
|
if self.use_tls and self.use_ssl:
|
|
msg = "Cannot use both TLS and SSL"
|
|
raise ValueError(msg)
|
|
|
|
|
|
class EmailService:
|
|
"""Service for sending emails via SMTP.
|
|
|
|
This service handles sending both plain text and HTML emails,
|
|
with support for TLS/SSL encryption and authentication.
|
|
"""
|
|
|
|
def __init__(self, config: EmailConfig):
|
|
"""Initialize email service.
|
|
|
|
Args:
|
|
config: Email configuration
|
|
|
|
"""
|
|
self.config = config
|
|
self._executor = None # Lazy-initialized thread pool for blocking SMTP
|
|
|
|
async def send_email(
|
|
self,
|
|
recipients: list[str],
|
|
subject: str,
|
|
body: str,
|
|
html_body: str | None = None,
|
|
) -> bool:
|
|
"""Send an email to recipients.
|
|
|
|
Args:
|
|
recipients: List of recipient email addresses
|
|
subject: Email subject line
|
|
body: Plain text email body
|
|
html_body: Optional HTML email body
|
|
|
|
Returns:
|
|
True if email was sent successfully, False otherwise
|
|
|
|
"""
|
|
if not recipients:
|
|
_LOGGER.warning("No recipients specified for email: %s", subject)
|
|
return False
|
|
|
|
try:
|
|
# Build message
|
|
msg = MIMEMultipart("alternative")
|
|
msg["Subject"] = subject
|
|
msg["From"] = f"{self.config.from_name} <{self.config.from_address}>"
|
|
msg["To"] = ", ".join(recipients)
|
|
msg["Date"] = datetime.now().strftime("%a, %d %b %Y %H:%M:%S %z")
|
|
|
|
# Attach plain text body
|
|
msg.attach(MIMEText(body, "plain"))
|
|
|
|
# Attach HTML body if provided
|
|
if html_body:
|
|
msg.attach(MIMEText(html_body, "html"))
|
|
|
|
# Send email in thread pool (SMTP is blocking)
|
|
loop = asyncio.get_event_loop()
|
|
await loop.run_in_executor(None, self._send_smtp, msg, recipients)
|
|
|
|
_LOGGER.info("Email sent successfully to %s: %s", recipients, subject)
|
|
return True
|
|
|
|
except Exception:
|
|
_LOGGER.exception("Failed to send email to %s: %s", recipients, subject)
|
|
return False
|
|
|
|
def _send_smtp(self, msg: MIMEMultipart, recipients: list[str]) -> None:
|
|
"""Send email via SMTP (blocking operation).
|
|
|
|
Args:
|
|
msg: Email message to send
|
|
recipients: List of recipient addresses
|
|
|
|
Raises:
|
|
Exception: If email sending fails
|
|
|
|
"""
|
|
if self.config.use_ssl:
|
|
# Connect with SSL from the start
|
|
context = ssl.create_default_context()
|
|
with smtplib.SMTP_SSL(
|
|
self.config.smtp_host,
|
|
self.config.smtp_port,
|
|
timeout=self.config.timeout,
|
|
context=context,
|
|
) as server:
|
|
if self.config.smtp_username and self.config.smtp_password:
|
|
server.login(self.config.smtp_username, self.config.smtp_password)
|
|
server.send_message(msg, self.config.from_address, recipients)
|
|
else:
|
|
# Connect and optionally upgrade to TLS
|
|
with smtplib.SMTP(
|
|
self.config.smtp_host,
|
|
self.config.smtp_port,
|
|
timeout=self.config.timeout,
|
|
) as server:
|
|
if self.config.use_tls:
|
|
context = ssl.create_default_context()
|
|
server.starttls(context=context)
|
|
|
|
if self.config.smtp_username and self.config.smtp_password:
|
|
server.login(self.config.smtp_username, self.config.smtp_password)
|
|
|
|
server.send_message(msg, self.config.from_address, recipients)
|
|
|
|
async def send_alert(
|
|
self,
|
|
recipients: list[str],
|
|
subject: str,
|
|
body: str,
|
|
) -> bool:
|
|
"""Send an alert email (convenience method).
|
|
|
|
Args:
|
|
recipients: List of recipient email addresses
|
|
subject: Email subject line
|
|
body: Email body text
|
|
|
|
Returns:
|
|
True if email was sent successfully, False otherwise
|
|
|
|
"""
|
|
return await self.send_email(recipients, subject, body)
|
|
|
|
async def send_daily_report(
|
|
self,
|
|
recipients: list[str],
|
|
stats: dict[str, Any],
|
|
errors: list[dict[str, Any]] | None = None,
|
|
) -> bool:
|
|
"""Send a daily report email.
|
|
|
|
Args:
|
|
recipients: List of recipient email addresses
|
|
stats: Dictionary containing statistics to include in report
|
|
errors: Optional list of errors to include
|
|
|
|
Returns:
|
|
True if email was sent successfully, False otherwise
|
|
|
|
"""
|
|
date_str = datetime.now().strftime("%Y-%m-%d")
|
|
subject = f"AlpineBits Daily Report - {date_str}"
|
|
|
|
# Build plain text body
|
|
body = f"AlpineBits Daily Report for {date_str}\n"
|
|
body += "=" * 60 + "\n\n"
|
|
|
|
# Add statistics
|
|
if stats:
|
|
body += "Statistics:\n"
|
|
body += "-" * 60 + "\n"
|
|
for key, value in stats.items():
|
|
body += f" {key}: {value}\n"
|
|
body += "\n"
|
|
|
|
# Add errors if present
|
|
if errors:
|
|
body += f"Errors ({len(errors)}):\n"
|
|
body += "-" * 60 + "\n"
|
|
for error in errors[:20]: # Limit to 20 most recent errors
|
|
timestamp = error.get("timestamp", "Unknown")
|
|
level = error.get("level", "ERROR")
|
|
message = error.get("message", "No message")
|
|
body += f" [{timestamp}] {level}: {message}\n"
|
|
if len(errors) > 20:
|
|
body += f" ... and {len(errors) - 20} more errors\n"
|
|
body += "\n"
|
|
|
|
body += "-" * 60 + "\n"
|
|
body += "Generated by AlpineBits Server\n"
|
|
|
|
# Build HTML body for better formatting
|
|
html_body = self._build_daily_report_html(date_str, stats, errors)
|
|
|
|
return await self.send_email(recipients, subject, body, html_body)
|
|
|
|
def _build_daily_report_html(
|
|
self,
|
|
date_str: str,
|
|
stats: dict[str, Any],
|
|
errors: list[dict[str, Any]] | None,
|
|
) -> str:
|
|
"""Build HTML version of daily report.
|
|
|
|
Args:
|
|
date_str: Date string for the report
|
|
stats: Statistics dictionary
|
|
errors: Optional list of errors
|
|
|
|
Returns:
|
|
HTML string for the email body
|
|
|
|
"""
|
|
html = f"""
|
|
<html>
|
|
<head>
|
|
<style>
|
|
body {{ font-family: Arial, sans-serif; }}
|
|
h1 {{ color: #333; }}
|
|
h2 {{ color: #666; margin-top: 20px; }}
|
|
table {{ border-collapse: collapse; width: 100%; }}
|
|
th, td {{ text-align: left; padding: 8px; border-bottom: 1px solid #ddd; }}
|
|
th {{ background-color: #f2f2f2; }}
|
|
.error {{ color: #d32f2f; }}
|
|
.warning {{ color: #f57c00; }}
|
|
.footer {{ margin-top: 30px; color: #999; font-size: 12px; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1>AlpineBits Daily Report</h1>
|
|
<p><strong>Date:</strong> {date_str}</p>
|
|
"""
|
|
|
|
# Add statistics table
|
|
if stats:
|
|
html += """
|
|
<h2>Statistics</h2>
|
|
<table>
|
|
<tr>
|
|
<th>Metric</th>
|
|
<th>Value</th>
|
|
</tr>
|
|
"""
|
|
for key, value in stats.items():
|
|
html += f"""
|
|
<tr>
|
|
<td>{key}</td>
|
|
<td>{value}</td>
|
|
</tr>
|
|
"""
|
|
html += "</table>"
|
|
|
|
# Add errors table
|
|
if errors:
|
|
html += f"""
|
|
<h2>Errors ({len(errors)})</h2>
|
|
<table>
|
|
<tr>
|
|
<th>Time</th>
|
|
<th>Level</th>
|
|
<th>Message</th>
|
|
</tr>
|
|
"""
|
|
for error in errors[:20]: # Limit to 20 most recent
|
|
timestamp = error.get("timestamp", "Unknown")
|
|
level = error.get("level", "ERROR")
|
|
message = error.get("message", "No message")
|
|
css_class = "error" if level == "ERROR" or level == "CRITICAL" else "warning"
|
|
html += f"""
|
|
<tr>
|
|
<td>{timestamp}</td>
|
|
<td class="{css_class}">{level}</td>
|
|
<td>{message}</td>
|
|
</tr>
|
|
"""
|
|
if len(errors) > 20:
|
|
html += f"""
|
|
<tr>
|
|
<td colspan="3"><em>... and {len(errors) - 20} more errors</em></td>
|
|
</tr>
|
|
"""
|
|
html += "</table>"
|
|
|
|
html += """
|
|
<div class="footer">
|
|
<p>Generated by AlpineBits Server</p>
|
|
</div>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
return html
|
|
|
|
|
|
def create_email_service(config: dict[str, Any]) -> EmailService | None:
|
|
"""Create an email service from configuration.
|
|
|
|
Args:
|
|
config: Full application configuration dictionary
|
|
|
|
Returns:
|
|
EmailService instance if email is configured, None otherwise
|
|
|
|
"""
|
|
email_config = config.get("email")
|
|
if not email_config:
|
|
_LOGGER.info("Email not configured, email service disabled")
|
|
return None
|
|
|
|
try:
|
|
email_cfg = EmailConfig(email_config)
|
|
service = EmailService(email_cfg)
|
|
_LOGGER.info("Email service initialized: %s:%s", email_cfg.smtp_host, email_cfg.smtp_port)
|
|
return service
|
|
except Exception:
|
|
_LOGGER.exception("Failed to initialize email service")
|
|
return None
|