import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart import time import logging from backend.config import settings logger = logging.getLogger(__name__) # --- 邮件模板 --- EXPIRATION_HTML_TEMPLATE = """ Token 到期提醒

⚠️ Token 即将到期提醒

您好,

您的接龙打卡系统 Token 即将到期,为避免影响自动打卡功能,请尽快刷新您的 Token。

到期时间:{exp_time}
通知时间:{send_time}

请登录系统,前往 用户设置 页面刷新您的 Token,以确保自动打卡功能正常运行。

""" SUCCESS_HTML_TEMPLATE = """ 打卡成功通知

✅ 打卡成功

🎉

您好,

自动打卡已成功完成!

打卡时间:{send_time}
打卡状态:成功 ✓

您无需进行任何操作,系统已自动为您完成打卡。此邮件仅作通知。

""" FAILURE_HTML_TEMPLATE = """ 打卡失败通知

❌ 打卡失败通知

⚠️

您好,

自动打卡失败,需要您的关注!

失败时间:{send_time}
失败原因:Token 已失效(需要登录)

📋 需要您执行以下操作:

  • 登录接龙自动打卡系统
  • 刷新您的 Authorization Token
  • 确认 Token 更新成功

Token 失效是正常现象,通常在一段时间后会自动过期。刷新 Token 后,系统将恢复自动打卡功能。

""" def get_email_settings(): """ 从环境变量读取邮件配置 如果 SMTP_SERVER、SMTP_PORT 或 SMTP_SENDER_EMAIL 有任一为空,则禁用邮件功能 Returns: dict: 邮件配置,如果配置不完整则返回 None """ # 检查必要的邮件配置是否存在 if not settings.SMTP_SERVER or not settings.SMTP_SENDER_EMAIL: logger.debug("邮件配置未完成(SMTP_SERVER 或 SMTP_SENDER_EMAIL 为空),邮件发送功能已禁用") return None if not settings.SMTP_PORT: logger.debug("邮件配置未完成(SMTP_PORT 为空),邮件发送功能已禁用") return None # 返回配置字典 return { 'smtpserver': settings.SMTP_SERVER, 'smtpport': settings.SMTP_PORT, 'senderemail': settings.SMTP_SENDER_EMAIL, 'senderpassword': settings.SMTP_SENDER_PASSWORD, 'use_ssl': settings.SMTP_USE_SSL } def _send_email(to_email: str, subject: str, html_content: str, email_settings: dict) -> bool: """ 发送邮件 Args: to_email: 收件人邮箱 subject: 邮件主题 html_content: HTML 邮件内容 email_settings: 邮件配置 Returns: 是否发送成功 """ try: msg = MIMEMultipart() msg["From"] = email_settings['senderemail'] msg["To"] = to_email msg["Subject"] = subject msg.attach(MIMEText(html_content, 'html', 'utf-8')) # 根据配置选择使用 SSL 或普通 SMTP if email_settings.get('use_ssl', True): with smtplib.SMTP_SSL(email_settings['smtpserver'], int(email_settings['smtpport'])) as server: server.login(email_settings['senderemail'], email_settings['senderpassword']) server.sendmail(msg["From"], msg["To"], msg.as_string()) else: with smtplib.SMTP(email_settings['smtpserver'], int(email_settings['smtpport'])) as server: server.starttls() server.login(email_settings['senderemail'], email_settings['senderpassword']) server.sendmail(msg["From"], msg["To"], msg.as_string()) logger.info(f"已成功向 {to_email} 发送邮件,主题: {subject}") return True except Exception as e: logger.error(f"向 {to_email} 发送邮件时失败: {e}") return False def send_expiration_notification(email: str, jwt_exp: str) -> bool: """ 发送 Token 到期提醒邮件 Args: email: 收件人邮箱 jwt_exp: Token 过期时间戳 Returns: 是否发送成功 """ email_settings = get_email_settings() if not email_settings: return False try: exp_time = time.strftime("%Y年%m月%d日 %H:%M:%S", time.localtime(float(jwt_exp))) send_time = time.strftime("%Y年%m月%d日 %H:%M:%S", time.localtime()) html = EXPIRATION_HTML_TEMPLATE.format( name=email, exp_time=exp_time, send_time=send_time ) return _send_email(email, "接龙管家Token到期通知", html, email_settings) except Exception as e: logger.error(f"发送过期通知邮件失败: {e}") return False def send_success_notification(email: str) -> bool: """ 发送打卡成功通知邮件 Args: email: 收件人邮箱 Returns: 是否发送成功 """ email_settings = get_email_settings() if not email_settings: return False try: send_time = time.strftime("%Y年%m月%d日 %H:%M:%S", time.localtime()) html = SUCCESS_HTML_TEMPLATE.format( name=email, send_time=send_time ) return _send_email(email, "自动打卡成功通知", html, email_settings) except Exception as e: logger.error(f"发送成功通知邮件失败: {e}") return False def send_failure_notification(email: str) -> bool: """ 发送打卡失败通知邮件 Args: email: 收件人邮箱 Returns: 是否发送成功 """ email_settings = get_email_settings() if not email_settings: return False try: send_time = time.strftime("%Y年%m月%d日 %H:%M:%S", time.localtime()) html = FAILURE_HTML_TEMPLATE.format( name=email, send_time=send_time ) return _send_email(email, "打卡失败 - 需要刷新Token", html, email_settings) except Exception as e: logger.error(f"发送失败通知邮件失败: {e}") return False