Files
pipeline-lifetime/app/email.py
T

74 lines
4.0 KiB
Python

from __future__ import annotations
import logging
from html import escape
from typing import Any
import resend
from flask import current_app
class EmailConfigurationError(RuntimeError):
"""Raised when transactional email has not been configured."""
class EmailDeliveryError(RuntimeError):
"""Raised when Resend rejects or cannot deliver an email request."""
def render_transactional_email(*, title: str, content: str) -> str:
"""Render a conservative table-based email layout for broad client support."""
return f"""<!doctype html>
<html lang="zh-CN"><body style="margin:0;background:#f3f5f8;color:#172033;font-family:-apple-system,BlinkMacSystemFont,'Segoe UI','Microsoft YaHei',Arial,sans-serif;">
<table role="presentation" width="100%" cellspacing="0" cellpadding="0" style="padding:32px 16px;background:#f3f5f8;"><tr><td align="center">
<table role="presentation" width="100%" cellspacing="0" cellpadding="0" style="max-width:600px;background:#ffffff;border:1px solid #e2e8f0;border-radius:12px;overflow:hidden;">
<tr><td style="padding:24px 32px;background:#005eb8;color:#ffffff;font-size:18px;font-weight:700;">供水管道健康评估系统</td></tr>
<tr><td style="padding:32px;"><h1 style="margin:0 0 12px;font-size:22px;line-height:1.4;color:#172033;">{escape(title)}</h1>{content}</td></tr>
<tr><td style="padding:18px 32px;border-top:1px solid #e2e8f0;color:#64748b;font-size:12px;line-height:1.7;">此邮件由系统自动发送,请勿直接回复。<br>如非本人操作,请忽略此邮件并及时检查账户安全。</td></tr>
</table>
</td></tr></table>
</body></html>"""
def verification_code_email(*, code: str, minutes: int, purpose: str) -> str:
"""Render the verification email without exposing dynamic HTML to callers."""
content = f"""
<p style="margin:0;color:#475569;font-size:15px;line-height:1.8;">你正在进行{escape(purpose)}。请在 {minutes} 分钟内输入下方验证码。</p>
<div style="margin:24px 0;padding:18px;border-radius:8px;background:#eaf3ff;color:#005eb8;text-align:center;font-size:30px;font-weight:700;letter-spacing:8px;">{escape(code)}</div>
<p style="margin:0;color:#64748b;font-size:13px;line-height:1.8;">验证码仅可使用一次,请勿向任何人透露。</p>"""
return render_transactional_email(title="邮箱验证码", content=content)
def password_reset_notice_email(*, username: str, reset_url: str) -> str:
"""Render an administrator-initiated password reset notification."""
safe_url = escape(reset_url, quote=True)
content = f"""
<p style="margin:0;color:#475569;font-size:15px;line-height:1.8;">{escape(username)},管理员已要求你重置密码。</p>
<p style="margin:16px 0 24px;color:#475569;font-size:15px;line-height:1.8;">请通过下方按钮进入找回密码流程,系统会向本邮箱发送一次性验证码。</p>
<p style="margin:0;"><a href="{safe_url}" style="display:inline-block;padding:12px 20px;border-radius:6px;background:#005eb8;color:#ffffff;font-size:14px;font-weight:700;text-decoration:none;">重置密码</a></p>"""
return render_transactional_email(title="请重置密码", content=content)
def send_transactional_email(*, to: str, subject: str, html: str) -> dict[str, Any]:
"""Send one application-generated email through Resend."""
api_key = current_app.config["RESEND_API_KEY"]
from_email = current_app.config["RESEND_FROM_EMAIL"]
if not api_key or not from_email:
raise EmailConfigurationError(
"邮件服务尚未配置,请设置 RESEND_API_KEY 和 RESEND_FROM_EMAIL。"
)
resend.api_key = api_key
try:
return resend.Emails.send(
{
"from": from_email,
"to": [to],
"subject": subject,
"html": html,
}
)
except Exception as exc:
logging.exception("Resend 邮件发送失败: %s", exc)
raise EmailDeliveryError("邮件发送失败,请稍后重试。") from exc