"""Email digest composition and SMTP delivery.
Provides three public functions:
- is_smtp_configured(settings) — gate check before attempting sends
- compose_digest_html(user_display_name, creator_content_groups, unsubscribe_url)
- send_email(to_address, subject, html_body, settings) — delivers via smtplib
"""
from __future__ import annotations
import logging
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from html import escape
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from config import Settings
logger = logging.getLogger(__name__)
# ── Public API ───────────────────────────────────────────────────────────────
def is_smtp_configured(settings: Settings) -> bool:
"""Return True only if SMTP host and from-address are set."""
return bool(settings.smtp_host) and bool(settings.smtp_from_address)
def compose_digest_html(
user_display_name: str,
creator_content_groups: list[dict],
unsubscribe_url: str,
) -> str:
"""Build an HTML email body for the digest.
Args:
user_display_name: Recipient's display name.
creator_content_groups: List of dicts, each with:
- creator_name (str)
- posts (list of {title, url})
- technique_pages (list of {title, url})
unsubscribe_url: Signed link to toggle off digests.
Returns:
HTML string ready for MIMEText.
"""
if not creator_content_groups:
return _minimal_html(user_display_name, unsubscribe_url)
sections: list[str] = []
for group in creator_content_groups:
creator_name = escape(group.get("creator_name", "Unknown"))
items_html = ""
for post in group.get("posts", []):
title = escape(post.get("title", "Untitled"))
url = escape(post.get("url", "#"))
items_html += (
f''
f'{title}'
f' (post)'
)
for page in group.get("technique_pages", []):
title = escape(page.get("title", "Untitled"))
url = escape(page.get("url", "#"))
items_html += (
f''
f'{title}'
f' (technique)'
)
if items_html:
sections.append(
f'{creator_name}
'
f''
)
body_content = "\n".join(sections) if sections else "No new content this period.
"
return _wrap_html(user_display_name, body_content, unsubscribe_url)
def send_email(
to_address: str,
subject: str,
html_body: str,
settings: Settings,
) -> bool:
"""Send an HTML email via SMTP.
Returns True on success, False on any SMTP error (logged).
Uses a 10-second connection timeout.
"""
if not is_smtp_configured(settings):
logger.warning("send_email called but SMTP is not configured")
return False
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = settings.smtp_from_address
msg["To"] = to_address
msg.attach(MIMEText(html_body, "html", "utf-8"))
try:
if settings.smtp_tls:
server = smtplib.SMTP(settings.smtp_host, settings.smtp_port, timeout=10)
server.starttls()
else:
server = smtplib.SMTP(settings.smtp_host, settings.smtp_port, timeout=10)
if settings.smtp_user:
server.login(settings.smtp_user, settings.smtp_password)
server.sendmail(settings.smtp_from_address, to_address, msg.as_string())
server.quit()
logger.info("Digest email sent to %s", to_address)
return True
except smtplib.SMTPException:
logger.exception("SMTP error sending digest to %s", to_address)
return False
except OSError:
logger.exception("Network error connecting to SMTP server for %s", to_address)
return False
# ── Private helpers ──────────────────────────────────────────────────────────
def _minimal_html(user_display_name: str, unsubscribe_url: str) -> str:
"""Fallback HTML when there's no new content."""
return _wrap_html(
user_display_name,
'No new content from your followed creators this period.
',
unsubscribe_url,
)
def _wrap_html(user_display_name: str, body_content: str, unsubscribe_url: str) -> str:
"""Wrap body content in the standard digest email shell."""
name = escape(user_display_name)
unsub = escape(unsubscribe_url)
return f"""\
"""