"""
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.
Template placeholders are primarily filled via environment variables.
"""
import base64
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, Dict, Any
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_css(path: str) -> str:
"""
Load CSS file content if it exists.
"""
if not path or not os.path.exists(path):
return ""
with open(path, "r", encoding="utf-8") as f:
return f.read()
def _email_logo_html() -> str:
"""
Return logo HTML for emails.
Priority:
1) EMAIL_LOGO_URL (direct URL to logo image)
2) EMAIL_LOGO_PATH (local file; embedded as base64)
3) empty string if neither is set.
"""
logo_url = os.getenv("EMAIL_LOGO_URL")
if logo_url:
return f'
'
logo_path = os.getenv("EMAIL_LOGO_PATH", "/app/src/misc/logo1.png")
if not os.path.exists(logo_path):
return ""
try:
with open(logo_path, "rb") as f:
b64 = base64.b64encode(f.read()).decode("utf-8")
return f'
'
except Exception:
return ""
def build_template_context(**runtime_kwargs: Any) -> Dict[str, Any]:
"""
Build a context dict for templates from:
- environment variables (base, customizable)
- runtime-provided values (override env if present)
Environment variables:
- EMAIL_CONTACT_ADDRESS: value for {contact_email}
- EMAIL_CSS_PATH: path to mail_style.css (optional; we inline it)
- EMAIL_LOGO_URL: URL for email logo (preferred)
- EMAIL_LOGO_PATH: fallback local path for email logo
"""
# Load and inline mail_style.css for consistent email styling
css_path = os.getenv("EMAIL_CSS_PATH", "/app/src/misc/mail_style.css")
css_text = _load_css(css_path)
# Build logo HTML (URL or local fallback)
logo_html = _email_logo_html()
ctx: Dict[str, Any] = {
"contact_email": os.getenv("EMAIL_CONTACT_ADDRESS", "support@example.com"),
"email_css": css_text,
"email_logo": logo_html,
}
# Runtime values override env if provided
if runtime_kwargs:
ctx.update(runtime_kwargs)
return ctx
def load_template(template_name: str, **runtime_kwargs: Any) -> 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()
# Build context from env + runtime
ctx = build_template_context(**runtime_kwargs)
# Replace {placeholder} style variables safely
try:
return template.format(**ctx)
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}")