199 lines
14 KiB
Python
199 lines
14 KiB
Python
"""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, kicker: str = 'YOUR MONTH IN VIEWING') -> str:
|
||
esc = html.escape
|
||
return f'''<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1"><meta name="color-scheme" content="dark"><title>{esc(title)}</title><style>@media(max-width:280px){{.email-metrics td{{display:block!important;width:auto!important;padding:16px 0!important}}.email-metrics tr{{display:block!important}}}}</style></head>
|
||
<body style="margin:0;padding:0;background:#131315;color:#e5e1e4;font-family:Arial,Helvetica,sans-serif">
|
||
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="background:#131315"><tr><td align="center" style="padding:24px 12px">
|
||
<table role="presentation" width="600" cellpadding="0" cellspacing="0" style="width:100%;max-width:600px;table-layout:fixed;background:#1c1b1d;border:1px solid #363338;border-radius:16px">
|
||
<tr><td style="padding:32px 24px 8px;color:#c7bdff;font-size:12px;letter-spacing:2px;font-weight:bold">MAGENT <span style="color:#918b98;letter-spacing:0">/ {esc(kicker)}</span></td></tr>
|
||
<tr><td style="padding:12px 24px"><h1 style="margin:0 0 16px;font-size:32px;line-height:1.2;color:#f3eef6">{esc(title)}</h1><p style="margin:0;color:#bdb6c3;font-size:15px;line-height:1.7;overflow-wrap:anywhere">{esc(intro)}</p></td></tr>
|
||
<tr><td style="padding:12px 24px">{content}</td></tr>
|
||
<tr><td style="padding:20px 24px 32px"><a href="{esc(url, quote=True)}" style="display:inline-block;padding:15px 22px;border-radius:8px;background:#c7bdff;color:#211b30;text-decoration:none;font-size:14px;font-weight:bold">{esc(action)} ↗</a></td></tr>
|
||
</table><table role="presentation" width="600" style="width:100%;max-width:600px"><tr><td style="padding:22px 18px;color:#a69fac;font-size:12px;line-height:1.7;text-align:center">{footer}</td></tr></table>
|
||
</td></tr></table></body></html>'''
|
||
|
||
|
||
def render_confirmation(username: str, url: str) -> dict:
|
||
title = "Your month, delivered."
|
||
intro = f"Hi {username}, confirm this email address to receive personal viewing reports from Magent. You choose whether to request them yourself or also receive automatic monthly emails."
|
||
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='<p style="color:#bdb6c3;font-size:14px;line-height:1.7">Minutes watched, movies, episodes, your longest run and requests — with a link to your full monthly report.</p>',
|
||
action="Confirm email recaps", url=url,
|
||
footer="This link expires in 24 hours. If you did not request this, ignore this email.<br>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, requested: bool = False) -> dict:
|
||
esc = html.escape
|
||
month = month_label(report["month"])
|
||
previous = month_label(report["comparison_month"])
|
||
if report.get('is_partial'):
|
||
month += ' so far'
|
||
previous += ' (same elapsed period, capped at month end)' if report.get('comparison_capped') else ' (same elapsed period)'
|
||
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'<td width="50%" valign="top" style="padding:16px 10px;border-bottom:1px solid #363338"><span style="color:#bdb6c3;font-size:12px">{label}</span><br><strong style="display:block;margin:10px 0;color:#e0d8ff;font-size:30px">{number(value)}</strong><span style="color:#a69fac;font-size:11px;line-height:1.6">{esc(comparison)}</span></td>')
|
||
content = '<table role="presentation" class="email-metrics" width="100%" cellpadding="0" cellspacing="0" style="table-layout:fixed"><tr>' + ''.join(cells[:2]) + '</tr><tr>' + ''.join(cells[2:]) + '</tr></table>'
|
||
habit = f"{number(summary['active_days'])} days watched · {number(summary['longest_streak'])}-day longest run"
|
||
content += f'<p style="color:#e5e1e4;font-size:14px;line-height:1.7;margin:24px 0">{esc(habit)}</p>'
|
||
patterns = report.get("patterns", {})
|
||
if patterns:
|
||
detail = f"Average play: {number(patterns['average_play_minutes'])} min. Longest play: {number(patterns['longest_play_minutes'])} min. Weekend viewing: {number(patterns['weekend_percent'])}%."
|
||
lines.append(detail)
|
||
content += f'<p style="padding:18px;background:#242334;border-radius:12px;color:#d8cfff;line-height:1.8">{esc(detail)}</p>'
|
||
for heading, rows in (("Your week in viewing (UTC)", patterns["weekdays"]), ("Movies, TV and more", patterns["media"])):
|
||
peak = max(1, *(row["minutes"] for row in rows))
|
||
content += f'<h2 style="font-size:18px;color:#e5e1e4">{heading}</h2><table role="presentation" width="100%" cellspacing="0" cellpadding="0">'
|
||
for row in rows:
|
||
width = round(row["minutes"] / peak * 100)
|
||
content += f'<tr><td style="padding:8px 0;color:#bdb6c3;font-size:12px;width:100px">{esc(row["name"])}</td><td style="padding:8px"><table role="presentation" width="{width}%" cellspacing="0" cellpadding="0"><tr><td height="8" style="background:{"#8cdbdd" if width else "transparent"};border-radius:4px;font-size:0"> </td></tr></table></td><td style="width:65px;color:#e0d8ff;font-size:12px;text-align:right">{number(row["minutes"])} min</td></tr>'
|
||
lines.append(f"{row['name']}: {number(row['minutes'])} minutes")
|
||
content += '</table>'
|
||
top = report.get("top_titles", [])[:3]
|
||
if top:
|
||
content += '<h2 style="font-size:18px;color:#e5e1e4;margin:24px 0 8px">Your most watched</h2>'
|
||
for item in top:
|
||
artwork = item.get("email_artwork", "")
|
||
if artwork.startswith(("cid:", "data:image/")):
|
||
content += f'<img src="{esc(artwork, quote=True)}" alt="{esc(item["title"], quote=True)}" width="80" style="display:block;border-radius:10px;margin-top:20px" />'
|
||
content += f'<p style="font-size:14px;line-height:1.6;color:#e5e1e4;margin:12px 0;overflow-wrap:anywhere">{esc(item["title"])}<br><span style="font-size:12px;color:#a69fac">{number(item["minutes"])} minutes · {number(item["plays"])} plays</span></p>'
|
||
else:
|
||
content += '<p style="font-size:14px;color:#bdb6c3;line-height:1.7">No viewing was recorded this month. Your requests are still included.</p>'
|
||
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 enabled personal report emails from Magent.<br>Based on retained Jellystat history. Calendar months use UTC; request statuses are current.<br><a href="{esc(unsubscribe_url, quote=True)}" style="color:#c7bdff">Unsubscribe from recaps</a> · <a href="{esc(public_url + "/profile#monthly-recaps", quote=True)}" style="color:#c7bdff">Email preferences</a>'
|
||
if requested:
|
||
intro = 'You requested this report. ' + intro
|
||
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")
|
||
html_part = message.get_payload()[-1]
|
||
for attachment in rendered.get('inline_images', []):
|
||
html_part.add_related(
|
||
attachment['data'], maintype='image', subtype=attachment.get('subtype', 'jpeg'), cid=f"<{attachment['cid']}>",
|
||
filename=attachment['cid'].split('@')[0] + '.' + attachment.get('subtype', 'jpeg'), disposition='inline')
|
||
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"<magent-recap-{delivery_id}@{host}>"
|