""" Email sender module for ScrAIbe. Sends transcription outputs (TXT, JSON, etc.) via SMTP. All credentials are configured via environment variables. Supports both plain text and HTML email bodies. """ import os import smtplib import logging from email import encoders from email.mime.base import MIMEBase from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from typing import List, Optional logger = logging.getLogger("scraibe.email_sender") class EmailError(Exception): pass def get_email_config(): """ Read email configuration from environment variables. Raises EmailError if required fields are missing. """ smtp_host = os.getenv("EMAIL_SMTP_HOST") smtp_port = os.getenv("EMAIL_SMTP_PORT") smtp_user = os.getenv("EMAIL_SMTP_USER") smtp_password = os.getenv("EMAIL_SMTP_PASSWORD") from_address = os.getenv("EMAIL_FROM_ADDRESS") use_tls_str = os.getenv("EMAIL_SMTP_USE_TLS", "true").strip().lower() use_tls = use_tls_str not in ("false", "0", "no") if not all([smtp_host, smtp_port, smtp_user, smtp_password, from_address]): raise EmailError( "Email configuration incomplete. " "Ensure EMAIL_SMTP_HOST, EMAIL_SMTP_PORT, EMAIL_SMTP_USER, " "EMAIL_SMTP_PASSWORD, and EMAIL_FROM_ADDRESS are set." ) return { "smtp_host": smtp_host, "smtp_port": int(smtp_port), "smtp_user": smtp_user, "smtp_password": smtp_password, "from_address": from_address, "use_tls": use_tls, } def load_template(template_name: str, **kwargs) -> str: """ Load an HTML email template from misc/ and render placeholders. Expects files like: /app/src/misc/upload_notification_template.html /app/src/misc/success_template.html /app/src/misc/error_notification_template.html """ base = os.getenv("SCRAIBE_TEMPLATES_DIR", "/app/src/misc") path = os.path.join(base, template_name) if not os.path.exists(path): raise EmailError(f"Email template not found: {path}") with open(path, "r", encoding="utf-8") as f: template = f.read() # Replace {placeholder} style variables safely try: return template.format(**kwargs) except KeyError as e: raise EmailError(f"Missing template variable: {e}") def send_email( to: str, subject: str, body: str, html: Optional[str], attachments: List[str], cc: Optional[str] = None, ) -> bool: """ Send an email with optional HTML body and file attachments. Args: to: Comma-separated list of recipient email addresses. subject: Email subject. body: Email body (plain text). html: Email body (HTML), or None. attachments: List of file paths to attach. cc: Comma-separated list of CC email addresses (optional). Returns: True if sent successfully. Raises: EmailError if sending fails. """ try: cfg = get_email_config() except EmailError as e: logger.error("Email configuration error: %s", e) raise # Parse recipients to_list = [addr.strip() for addr in to.split(",") if addr.strip()] cc_list = [addr.strip() for addr in cc.split(",") if addr.strip()] if cc else [] if not to_list: raise EmailError("No valid 'To' email addresses provided.") # Build message msg = MIMEMultipart("alternative") msg["From"] = cfg["from_address"] msg["To"] = ", ".join(to_list) if cc_list: msg["Cc"] = ", ".join(cc_list) msg["Subject"] = subject # Attach plain text msg.attach(MIMEText(body, "plain")) # Attach HTML if provided if html: msg.attach(MIMEText(html, "html")) # Attach files in a separate multipart/mixed part if attachments: mixed = MIMEMultipart("mixed") mixed.attach(msg) msg = mixed for file_path in attachments: if not os.path.isfile(file_path): logger.warning("Attachment file not found, skipping: %s", file_path) continue try: with open(file_path, "rb") as f: part = MIMEBase("application", "octet-stream") part.set_payload(f.read()) encoders.encode_base64(part) part.add_header( "Content-Disposition", "attachment", filename=os.path.basename(file_path), ) msg.attach(part) except Exception as e: logger.warning("Failed to attach file %s: %s", file_path, e) # Connect and send try: if cfg["use_tls"]: server = smtplib.SMTP(cfg["smtp_host"], cfg["smtp_port"], timeout=30) server.ehlo() server.starttls() server.ehlo() else: server = smtplib.SMTP(cfg["smtp_host"], cfg["smtp_port"], timeout=30) server.ehlo() server.login(cfg["smtp_user"], cfg["smtp_password"]) server.sendmail( cfg["from_address"], to_list + cc_list, msg.as_string(), ) server.quit() logger.info( "Email sent to %s (CC: %s)", to_list, cc_list or "None", ) return True except Exception as e: logger.error("Failed to send email: %s", e) raise EmailError(f"Failed to send email: {e}")