mirror of
https://github.com/stardrophere/InsightRadar.git
synced 2026-06-06 01:57:51 +08:00
stardrophere
50 lines
1.3 KiB
Python
50 lines
1.3 KiB
Python
# app/utils/email_utils.py
|
|
import os
|
|
from email.message import EmailMessage
|
|
import aiosmtplib
|
|
from dotenv import load_dotenv
|
|
import asyncio
|
|
|
|
load_dotenv()
|
|
|
|
SMTP_HOST = os.getenv("SMTP_HOST", "smtp.qiye.aliyun.com")
|
|
SMTP_PORT = int(os.getenv("SMTP_PORT", 465))
|
|
SMTP_USER = os.getenv("SMTP_USER", "noreply@yourdomain.com")
|
|
SMTP_PASS = os.getenv("SMTP_PASS", "your_password")
|
|
|
|
|
|
async def send_html_email(
|
|
to_email: str,
|
|
subject: str,
|
|
html_content: str,
|
|
sender_name: str = "AI 新闻",
|
|
sender_email: str = None
|
|
) -> bool:
|
|
"""底层纯异步发送邮件工具"""
|
|
# 如果未指定发送者邮箱，默认使用环境配置中的认证邮箱
|
|
if sender_email is None:
|
|
sender_email = SMTP_USER
|
|
|
|
message = EmailMessage()
|
|
# 动态拼接 From 字段
|
|
message["From"] = f"{sender_name} <{sender_email}>"
|
|
message["To"] = to_email
|
|
message["Subject"] = subject
|
|
|
|
# 设定内容为 HTML
|
|
message.set_content(html_content, subtype="html")
|
|
|
|
try:
|
|
await aiosmtplib.send(
|
|
message,
|
|
hostname=SMTP_HOST,
|
|
port=SMTP_PORT,
|
|
username=SMTP_USER,
|
|
password=SMTP_PASS,
|
|
use_tls=True,
|
|
)
|
|
return True
|
|
except Exception as e:
|
|
print(f"邮件发送至 {to_email} 失败: {str(e)}")
|
|
return False
|