admin
249 lines
7.4 KiB
Python
249 lines
7.4 KiB
Python
"""
|
|
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 _logo1_inline_html() -> str:
|
|
"""
|
|
Return an inline <img> tag for logo1.png as base64.
|
|
Falls back to empty string if not found.
|
|
"""
|
|
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'<img src="data:image/png;base64,{b64}" alt="A.P.Strom" style="max-width:180px; display:block; margin:0 auto 10px auto;"/>'
|
|
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_PATH: path to logo1.png (default: /app/src/misc/logo1.png)
|
|
"""
|
|
# 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 inline logo HTML
|
|
logo_html = _logo1_inline_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}")
|