"""异步邮件发送服务（aiosmtplib）"""
import aiosmtplib
from email.message import EmailMessage
from typing import Optional
from decimal import Decimal
async def send_email(
to: str,
subject: str,
html_body: str,
text_body: Optional[str] = None,
smtp_config: Optional[dict] = None,
) -> bool:
"""
异步发送邮件，失败返回 False 不抛异常。
smtp_config 可选，格式：
{
"host": "smtp.example.com",
"port": 587,
"user": "...",
"password": "...",
"from_name": "SME Store",
"from_email": "...",
"use_tls": True,
"enabled": True,
}
若 disabled 或 config 为空，跳过发送（不报错）。
"""
if not smtp_config or not smtp_config.get("enabled"):
print(f"[Email] SMTP disabled or not configured, skipping: {to} / {subject}")
return False
host = smtp_config.get("host", "")
port = smtp_config.get("port", 587)
user = smtp_config.get("user", "")
password = smtp_config.get("password", "")
from_name = smtp_config.get("from_name", "SME Store")
from_email = smtp_config.get("from_email", user)
use_tls = smtp_config.get("use_tls", True)
if not host or not from_email:
print(f"[Email] SMTP not fully configured, skipping: {to} / {subject}")
return False
msg = EmailMessage()
msg["From"] = f"{from_name} <{from_email}>"
msg["To"] = to
msg["Subject"] = subject
msg.set_content(text_body or html_body, subtype="plain")
msg.add_alternative(html_body, subtype="html")
try:
await aiosmtplib.send(
msg,
hostname=host,
port=port,
username=user,
password=password,
start_tls=use_tls,
)
print(f"[Email] Sent to {to}: {subject}")
return True
except Exception as e:
print(f"[Email] Failed to send to {to}: {e}")
return False
def render_order_email(
customer_name: str,
order_no: str,
items: list[dict],
total: str,
status: str,
shipping_address: str,
) -> str:
"""生成订单确认邮件 HTML"""
status_map = {
"pending": "待支付",
"paid": "已支付",
"shipped": "已发货",
"completed": "已完成",
"cancelled": "已取消",
}
status_text = status_map.get(status, status)
items_html = "".join(
f"| {item['name']} | {item.get('variant', '')} | "
f"x{item['quantity']} | {item['price']} |
"
for item in items
)
return f"""
订单确认 - {order_no}
亲爱的 {customer_name}，您的订单已确认。
合计：{total}
收货地址：{shipping_address}
订单状态：{status_text}
"""
def build_smtp_config(settings_obj) -> dict:
"""把 TenantSettings ORM 对象转成 send_email 需要的 dict 格式"""
if not settings_obj:
return {}
return {
"host": settings_obj.smtp_host or "",
"port": settings_obj.smtp_port or 587,
"user": settings_obj.smtp_user or "",
"password": settings_obj.smtp_password or "",
"from_name": settings_obj.smtp_from_name or "SME Store",
"from_email": settings_obj.smtp_from_email or "",
"use_tls": bool(settings_obj.smtp_use_tls),
"enabled": bool(settings_obj.smtp_enabled),
}