57 lines
2 KiB
Python
57 lines
2 KiB
Python
"""Email notification provider."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import smtplib
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.mime.text import MIMEText
|
|
|
|
from notifications.base import LoggingNotificationService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class EmailNotificationService(LoggingNotificationService):
|
|
"""
|
|
Email notification service.
|
|
|
|
Uses SMTP credentials from environment variables when configured.
|
|
Falls back to logging when SMTP is not set up.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.smtp_host = os.environ.get("SMTP_HOST", "")
|
|
self.smtp_port = int(os.environ.get("SMTP_PORT", "587"))
|
|
self.smtp_user = os.environ.get("SMTP_USER", "")
|
|
self.smtp_password = os.environ.get("SMTP_PASSWORD", "")
|
|
self.email_from = os.environ.get("EMAIL_FROM", "")
|
|
self.email_to = os.environ.get("EMAIL_TO", "")
|
|
self.use_tls = os.environ.get("SMTP_USE_TLS", "true").lower() == "true"
|
|
|
|
def _smtp_configured(self) -> bool:
|
|
return all([self.smtp_host, self.smtp_user, self.smtp_password, self.email_from, self.email_to])
|
|
|
|
def send_email(self, subject: str, body: str) -> bool:
|
|
if not self._smtp_configured():
|
|
return super().send_email(subject, body)
|
|
|
|
try:
|
|
msg = MIMEMultipart()
|
|
msg["From"] = self.email_from
|
|
msg["To"] = self.email_to
|
|
msg["Subject"] = subject
|
|
msg.attach(MIMEText(body, "plain"))
|
|
|
|
with smtplib.SMTP(self.smtp_host, self.smtp_port, timeout=30) as server:
|
|
if self.use_tls:
|
|
server.starttls()
|
|
server.login(self.smtp_user, self.smtp_password)
|
|
server.sendmail(self.email_from, [self.email_to], msg.as_string())
|
|
|
|
logger.info("Email sent to %s", self.email_to)
|
|
return True
|
|
except Exception as exc:
|
|
logger.error("Failed to send email: %s", exc)
|
|
return False
|