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"""
供水管道健康评估系统

{escape(title)}

{content}
此邮件由系统自动发送，请勿直接回复。
如非本人操作，请忽略此邮件并及时检查账户安全。
""" def verification_code_email(*, code: str, minutes: int, purpose: str) -> str: """Render the verification email without exposing dynamic HTML to callers.""" content = f"""

你正在进行{escape(purpose)}。请在 {minutes} 分钟内输入下方验证码。

{escape(code)}

验证码仅可使用一次，请勿向任何人透露。

""" 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"""

{escape(username)}，管理员已要求你重置密码。

请通过下方按钮进入找回密码流程，系统会向本邮箱发送一次性验证码。

重置密码

""" 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