"""邮件发送服务""" from __future__ import annotations import logging import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from typing import Optional logger = logging.getLogger(__name__) def send_email( smtp_host: str, smtp_port: int, smtp_user: str, smtp_password: str, use_tls: bool, to_email: str, subject: str, body: str, html_body: Optional[str] = None, ) -> dict: """ 发送邮件 Args: smtp_host: SMTP 服务器地址 smtp_port: SMTP 端口 smtp_user: SMTP 用户名 smtp_password: SMTP 密码 use_tls: 是否使用 TLS to_email: 收件人邮箱 subject: 邮件主题 body: 纯文本内容 html_body: HTML 内容（可选） Returns: {"ok": True/False, "message": "..."} """ try: # 创建邮件 msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = smtp_user msg["To"] = to_email # 添加纯文本内容 part1 = MIMEText(body, "plain", "utf-8") msg.attach(part1) # 添加 HTML 内容 if html_body: part2 = MIMEText(html_body, "html", "utf-8") msg.attach(part2) # 连接 SMTP 服务器 if use_tls: server = smtplib.SMTP(smtp_host, smtp_port, timeout=30) server.starttls() else: server = smtplib.SMTP_SSL(smtp_host, smtp_port, timeout=30) server.login(smtp_user, smtp_password) server.sendmail(smtp_user, [to_email], msg.as_string()) server.quit() logger.info("Email sent successfully to %s: %s", to_email, subject) return {"ok": True, "message": "邮件发送成功"} except smtplib.SMTPAuthenticationError as e: logger.error("SMTP authentication failed: %s", e) return {"ok": False, "message": "SMTP 认证失败，请检查用户名和密码"} except smtplib.SMTPConnectError as e: logger.error("SMTP connection failed: %s", e) return {"ok": False, "message": "无法连接到 SMTP 服务器"} except Exception as e: logger.exception("Failed to send email") return {"ok": False, "message": str(e)} def send_traffic_alert_email( smtp_host: str, smtp_port: int, smtp_user: str, smtp_password: str, use_tls: bool, to_email: str, machine_name: str, aws_service: str, current_traffic_gb: float, limit_gb: float, traffic_type: str, # "total" or "upload" ) -> dict: """ 发送流量预警邮件 Args: machine_name: 机器名称 aws_service: 服务类型（ec2/lightsail） current_traffic_gb: 当前流量（GB） limit_gb: 限制流量（GB） traffic_type: 流量类型（total=总流量, upload=上传流量） """ if traffic_type == "total": traffic_desc = "总流量（上传+下载）" else: traffic_desc = "上传流量" subject = f"[ProxyAuto] 流量预警 - {machine_name}" body = f""" 流量预警通知机器名称: {machine_name} 服务类型: {aws_service.upper()} 预警类型: {traffic_desc}超限当前{traffic_desc}: {current_traffic_gb:.2f} GB 设定限制: {limit_gb:.2f} GB 超出: {current_traffic_gb - limit_gb:.2f} GB 系统已自动暂停该机器的 IP 自动更换任务。 --- ProxyAuto Pro 自动通知 """ html_body = f"""

警告：{traffic_desc}已超出设定限制！

机器名称 {machine_name}

服务类型 {aws_service.upper()}

当前{traffic_desc} {current_traffic_gb:.2f} GB

设定限制 {limit_gb:.2f} GB

超出流量 {current_traffic_gb - limit_gb:.2f} GB

系统已自动暂停该机器的 IP 自动更换任务。请登录控制台查看详情。

""" return send_email( smtp_host=smtp_host, smtp_port=smtp_port, smtp_user=smtp_user, smtp_password=smtp_password, use_tls=use_tls, to_email=to_email, subject=subject, body=body, html_body=html_body, )