"""Personal recap email rendering and SMTP delivery with explicit acceptance tracking."""
import html
import re
import smtplib
import ssl
from contextlib import suppress
from datetime import datetime
from email.message import EmailMessage
from email.policy import SMTP as SMTP_POLICY
from email.utils import formataddr, formatdate
from urllib.parse import urlsplit
from ..runtime import get_runtime_settings
class DeliveryError(Exception):
def __init__(self, state: str, detail: str):
self.state, self.detail = state, detail
super().__init__(detail)
class DeliveryCancelled(Exception):
pass
def valid_email(value: str | None) -> str | None:
value = str(value or "").strip()
if (len(value) <= 254 and re.fullmatch(r"[^@\s<>;,\"\\]+@[^@\s<>;,\"\\]+\.[^@\s<>;,\"\\]+", value)
and all(32 < ord(char) < 127 for char in value)):
return value
return None
def month_label(value: str) -> str:
return datetime.strptime(value, "%Y-%m").strftime("%B %Y")
def number(value: float) -> str:
return f"{value:,.0f}"
def document(*, title: str, intro: str, content: str, action: str, url: str, footer: str) -> str:
esc = html.escape
return f'''{esc(title)}
| MAGENT / YOUR MONTH IN VIEWING |
{esc(title)}{esc(intro)} |
| {content} |
| {esc(action)} ↗ |
|
'''
def render_confirmation(username: str, url: str) -> dict:
title = "Your month, delivered."
intro = f"Hi {username}, confirm this email address to receive your personal monthly viewing recap from Magent."
text = f"{intro}\n\nConfirm email recaps: {url}\n\nThis link expires in 24 hours. If you did not request this, ignore this email. No viewing history will be emailed until you confirm."
body = document(title=title, intro=intro,
content='Minutes watched, movies, episodes, your longest run and requests — with a link to your full monthly report.
',
action="Confirm email recaps", url=url,
footer="This link expires in 24 hours. If you did not request this, ignore this email.
No viewing history will be emailed until you confirm.")
return {"subject": "Confirm your Magent email recaps", "body_text": text, "body_html": body}
def render_recap(report: dict, username: str, public_url: str, unsubscribe_url: str, *, test: bool = False) -> dict:
esc = html.escape
month = month_label(report["month"])
previous = month_label(report["comparison_month"])
summary = report["summary"]
metrics = (("Minutes watched", "minutes", summary["minutes"]), ("Movies played", "movies", summary["movies"]),
("Episodes played", "episodes", summary["episodes"]), ("Requests made", "requests", report["requests"]["total"]))
cells, lines = [], []
for label, key, value in metrics:
change = report["changes"][key]
difference = change["difference"]
comparison = ("No change" if difference == 0 else f"{'+' if difference > 0 else '−'}{number(abs(difference))}")
if change["percent"] is not None and difference:
comparison += f" ({'+' if difference > 0 else '−'}{abs(change['percent']):g}%)"
comparison += f" from {previous}"
lines.append(f"{label}: {number(value)}. {comparison}.")
cells.append(f'{label}
{number(value)}{esc(comparison)} | ')
content = '' + ''.join(cells[:2]) + '
' + ''.join(cells[2:]) + '
'
habit = f"{number(summary['active_days'])} days watched · {number(summary['longest_streak'])}-day longest run"
content += f'{esc(habit)}
'
top = report.get("top_titles", [])[:3]
if top:
content += 'Your most watched
'
for item in top:
content += f'{esc(item["title"])}
{number(item["minutes"])} minutes · {number(item["plays"])} plays
'
else:
content += 'No viewing was recorded this month. Your requests are still included.
'
report_url = f"{public_url}/insights/reports?month={report['month']}"
intro = f"Hi {username}, here’s your {month} in viewing. A little look back at the stories you spent time with."
footer = f'You opted in to personal monthly recaps from Magent.
Based on retained Jellystat history. Calendar months use UTC; request statuses are current.
Unsubscribe from recaps · Email preferences'
if test:
intro = "This is your test recap. " + intro
body = document(title=month, intro=intro, content=content, action="Explore your full report", url=report_url, footer=footer)
text = '\n'.join([intro, '', *lines, '', habit, '', 'Most watched:',
*(f"{item['title']}: {number(item['minutes'])} minutes" for item in top), '',
f"Your full report: {report_url}", '', 'Based on retained Jellystat history. Calendar months use UTC; request statuses are current.',
f"Unsubscribe from recaps: {unsubscribe_url}", f"Email preferences: {public_url}/profile#monthly-recaps"])
return {"subject": f"{'[Test] ' if test else ''}Your {month} in viewing · Magent", "body_text": text, "body_html": body}
def send_email(recipient: str, rendered: dict, message_id: str, before_data=lambda: None) -> None:
"""Return only after SMTP accepts DATA. Never retry an ambiguous DATA disconnect.
A stable Message-ID aids diagnosis; it is not an SMTP deduplication guarantee.
See RFC 5321 §4.5.3.2.6 and Python's smtplib exception definitions.
"""
runtime = get_runtime_settings()
sender = valid_email(runtime.magent_notify_email_from_address)
if not sender or not valid_email(recipient):
raise DeliveryError("failed", "A valid sender and recipient email are required.")
message = EmailMessage(policy=SMTP_POLICY)
message["From"] = formataddr((str(runtime.magent_notify_email_from_name or "Magent").replace('\r', '').replace('\n', ''), sender))
message["To"], message["Subject"] = recipient, rendered["subject"]
message["Date"], message["Message-ID"] = formatdate(localtime=False), message_id
message["Auto-Submitted"], message["X-Auto-Response-Suppress"] = "auto-generated", "All"
message.set_content(rendered["body_text"])
message.add_alternative(rendered["body_html"], subtype="html")
payload = message.as_bytes()
smtp, stage = None, "connect"
try:
kwargs = {"timeout": 30, "local_hostname": sender.split('@', 1)[1]}
if runtime.magent_notify_email_use_ssl:
smtp = smtplib.SMTP_SSL(runtime.magent_notify_email_smtp_host, runtime.magent_notify_email_smtp_port,
context=ssl.create_default_context(), **kwargs)
else:
smtp = smtplib.SMTP(runtime.magent_notify_email_smtp_host, runtime.magent_notify_email_smtp_port, **kwargs)
smtp.ehlo_or_helo_if_needed()
if runtime.magent_notify_email_use_tls and not runtime.magent_notify_email_use_ssl:
smtp.starttls(context=ssl.create_default_context())
smtp.ehlo()
if runtime.magent_notify_email_smtp_username:
smtp.login(runtime.magent_notify_email_smtp_username, runtime.magent_notify_email_smtp_password)
code, reply = smtp.mail(sender)
if code != 250:
raise smtplib.SMTPResponseException(code, reply)
code, reply = smtp.rcpt(recipient)
if code not in (250, 251):
raise smtplib.SMTPResponseException(code, reply)
before_data()
stage = "data"
code, reply = smtp.data(payload)
if code != 250:
raise smtplib.SMTPDataError(code, reply)
stage = "accepted"
except smtplib.SMTPResponseException as exc:
state = "retry" if 400 <= exc.smtp_code < 500 else "failed"
raise DeliveryError(state, f"Mail server returned SMTP {exc.smtp_code}.") from exc
except (ssl.SSLError, smtplib.SMTPNotSupportedError, UnicodeError, ValueError) as exc:
raise DeliveryError("failed", "Check the SMTP security and sender settings.") from exc
except (OSError, smtplib.SMTPException) as exc:
state = "unknown" if stage == "data" else "retry"
detail = "Mail server acceptance is unknown; check its logs before taking further action." if state == "unknown" else "Could not reach or finish connecting to the mail server."
raise DeliveryError(state, detail) from exc
finally:
if smtp:
# A failed QUIT after a 250 DATA response must not turn an accepted email into a retry.
with suppress(Exception):
smtp.quit()
with suppress(Exception):
smtp.close()
def message_id(delivery_id: str, public_url: str) -> str:
host = urlsplit(public_url).hostname or "magent.local"
return f""