"""Shared SMTP send helper for worker services. Worker services historically called ``smtplib.SMTP("localhost", 25)`` directly, which fails inside the workers container (no MTA on localhost:25) -- so every authorization / signing-link / delivery email silently failed with ``[Errno 111] Connection refused`` and customers never received them. This routes all worker email through the same authenticated SMTP relay the rest of the system uses (Carbonio at co.carrierone.com:587 by default), configured via the standard SMTP_* env vars. Usage: from scripts.workers.worker_email import send_worker_email ok = send_worker_email(to, subject, html, attachments=[(filename, bytes, mime)]) """ from __future__ import annotations import logging import os import smtplib import sys from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.application import MIMEApplication # Shared HTML -> plaintext converter (scripts/_email_plaintext.py). Adding a # text/plain part to every message is a deliverability win: a multipart/ # alternative with ONLY an HTML part is malformed, and HTML-only mail is a # spam-score signal. Import is path-tolerant (module run from /app or scripts/). try: from scripts._email_plaintext import html_to_text except ImportError: # pragma: no cover - fallback when scripts/ is on sys.path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from _email_plaintext import html_to_text # type: ignore LOG = logging.getLogger("workers.email") SMTP_HOST = os.getenv("SMTP_HOST", "co.carrierone.com") SMTP_PORT = int(os.getenv("SMTP_PORT", "587")) SMTP_USER = os.getenv("SMTP_USER", "noreply@performancewest.net") SMTP_PASS = os.getenv("SMTP_PASS", "") SMTP_FROM = os.getenv("SMTP_FROM", "Performance West ") def send_worker_email( to: str, subject: str, html: str, *, text: str | None = None, attachments: list[tuple[str, bytes, str]] | None = None, cc: str | None = None, ) -> bool: """Send an email via the configured SMTP relay. Returns True on success. attachments: list of (filename, content_bytes, subtype) e.g. ("x.pdf", b"...", "pdf"). Never raises -- logs and returns False so callers stay non-fatal. """ try: msg = MIMEMultipart("mixed") msg["From"] = SMTP_FROM msg["To"] = to if cc: msg["Cc"] = cc msg["Subject"] = subject alt = MIMEMultipart("alternative") # Always include a text/plain part. If the caller did not supply one, # derive it from the HTML so the message is a well-formed multipart/ # alternative (HTML-only is malformed + a spam signal). RFC 2046: list # the plain part first (least->most preferred). plain = text if text else html_to_text(html) if plain: alt.attach(MIMEText(plain, "plain")) alt.attach(MIMEText(html, "html")) msg.attach(alt) for fname, content, subtype in (attachments or []): part = MIMEApplication(content, _subtype=subtype) part.add_header("Content-Disposition", "attachment", filename=fname) msg.attach(part) recipients = [to] + ([cc] if cc else []) with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=30) as s: s.starttls() if SMTP_USER and SMTP_PASS: s.login(SMTP_USER, SMTP_PASS) s.sendmail(SMTP_FROM, recipients, msg.as_string()) LOG.info("Worker email sent to %s (subject=%r)", to, subject[:60]) return True except Exception as exc: # noqa: BLE001 LOG.warning("Worker email send failed to %s: %s", to, exc) return False