"""邮件发送服务 - SMTP email sending for alarm notifications and report delivery.""" import logging import smtplib import ssl from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.base import MIMEBase from email import encoders from pathlib import Path from typing import Optional from app.core.config import get_settings logger = logging.getLogger("email_service") async def send_email( to: list[str], subject: str, body_html: str, attachments: Optional[list[str]] = None, ) -> bool: """ Send an email via SMTP. Args: to: List of recipient email addresses. subject: Email subject line. body_html: HTML body content. attachments: Optional list of file paths to attach. Returns: True if sent successfully, False otherwise. """ settings = get_settings() if not settings.SMTP_ENABLED: logger.warning("SMTP is not enabled (SMTP_ENABLED=False). Skipping email send.") return False if not settings.SMTP_HOST: logger.warning("SMTP_HOST is not configured. Skipping email send.") return False if not to: logger.warning("No recipients specified. Skipping email send.") return False try: msg = MIMEMultipart("mixed") msg["From"] = settings.SMTP_FROM msg["To"] = ", ".join(to) msg["Subject"] = subject # HTML body html_part = MIMEText(body_html, "html", "utf-8") msg.attach(html_part) # Attachments if attachments: for filepath in attachments: path = Path(filepath) if not path.exists(): logger.warning(f"Attachment not found, skipping: {filepath}") continue with open(path, "rb") as f: part = MIMEBase("application", "octet-stream") part.set_payload(f.read()) encoders.encode_base64(part) part.add_header( "Content-Disposition", f'attachment; filename="{path.name}"', ) msg.attach(part) # Send via SMTP context = ssl.create_default_context() if settings.SMTP_PORT == 465: # SSL connection with smtplib.SMTP_SSL(settings.SMTP_HOST, settings.SMTP_PORT, context=context) as server: if settings.SMTP_USER and settings.SMTP_PASSWORD: server.login(settings.SMTP_USER, settings.SMTP_PASSWORD) server.sendmail(settings.SMTP_FROM, to, msg.as_string()) else: # STARTTLS connection (port 587 or 25) with smtplib.SMTP(settings.SMTP_HOST, settings.SMTP_PORT) as server: server.ehlo() if settings.SMTP_PORT == 587: server.starttls(context=context) server.ehlo() if settings.SMTP_USER and settings.SMTP_PASSWORD: server.login(settings.SMTP_USER, settings.SMTP_PASSWORD) server.sendmail(settings.SMTP_FROM, to, msg.as_string()) logger.info(f"Email sent successfully to {to}, subject: {subject}") return True except smtplib.SMTPException as e: logger.error(f"SMTP error sending email to {to}: {e}") return False except Exception as e: logger.error(f"Unexpected error sending email to {to}: {e}") return False