Tolan/sendmsg/send_email.py

"""发送邮箱验证码服务（准实时）

从数据库读取待发送的邮箱验证码，使用阿里云 DirectMail SMTP 发送。

Prerequisites:
- Python >= 3.7
- 配置文件：send.conf（与脚本同目录）
  - 必需配置：[smtp] 下的 smtp_username 和 smtp_password
  - 必需配置：[database] 下的数据库配置

运行: python send_email.py
"""

import asyncio
import configparser
import os
import sys
import logging
import smtplib
from datetime import datetime
from pathlib import Path
from typing import Optional, Tuple

from email.header import Header
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.utils import formataddr
from sqlalchemy import select, update

# 加载配置文件
config_file = Path(__file__).parent / "send.conf"
config = configparser.ConfigParser()
if config_file.exists():
    config.read(config_file, encoding='utf-8')
else:
    raise FileNotFoundError(f"配置文件不存在: {config_file}")

# 导入数据库相关模块（需要从 backend 目录导入）
backend_dir = Path(__file__).parent.parent / "backend"
sys.path.insert(0, str(backend_dir))
from database import DatabaseManager
from models_db import VerificationCode

logger = logging.getLogger(__name__)

# ------- 从配置文件读取参数 -------
# SMTP 配置
SMTP_HOST = config.get("smtp", "smtp_host", fallback="smtpdm.aliyun.com")
SMTP_PORT_SSL = int(config.get("smtp", "smtp_port_ssl", fallback="465"))
SMTP_USERNAME = config.get("smtp", "smtp_username", fallback="").strip()
SMTP_PASSWORD = config.get("smtp", "smtp_password", fallback="").strip()
EMAIL_SUBJECT = config.get("smtp", "email_subject", fallback="Lyrria 注册登录验证码")

# 数据库配置
MYSQL_HOST = config.get("database", "mysql_host", fallback="").strip()
MYSQL_PORT = config.get("database", "mysql_port", fallback="").strip()
MYSQL_USER = config.get("database", "mysql_user", fallback="").strip()
MYSQL_PASSWORD = config.get("database", "mysql_password", fallback="").strip()
MYSQL_DATABASE = config.get("database", "mysql_database", fallback="").strip()

# 服务配置
POLL_INTERVAL = float(config.get("service", "poll_interval", fallback="2.0"))

# 邮件模板 HTML
EMAIL_BODY_TEMPLATE = """
<!DOCTYPE html>
<html lang="zh-CN">
<head>
  <meta charset="UTF-8" />
  <title>Lyrria 验证码</title>
  <style>
    body {{ margin: 0; padding: 0; background: #28272F; font-family: 'Helvetica Neue', Arial, sans-serif; }}
    .wrap {{ padding: 32px 16px; }}
    .container {{ max-width: 560px; margin: 0 auto; background: #1f1e25; border-radius: 16px; padding: 32px 32px 28px; box-shadow: 0 18px 45px rgba(0,0,0,0.26); color: #e5e7eb; }}
    .logo {{ font-size: 18px; font-weight: 800; color: #6b8bff; letter-spacing: 0.3px; }}
    .hero {{ margin-top: 12px; text-align: center; }}
    .hero img {{ width: 120px; height: auto; filter: drop-shadow(0 10px 18px rgba(0,0,0,0.15)); display: block; margin: 0 auto; }}
    .title {{ margin: 18px 0 8px; font-size: 22px; font-weight: 800; color: #f8fafc; }}
    .subtitle {{ margin: 0 0 16px; color: #cbd5e1; font-size: 14px; }}
    .code-box {{ margin: 16px 0; padding: 16px 20px; background: rgba(255,255,255,0.06); border: 1px solid rgba(255,255,255,0.12); border-radius: 12px; display: inline-block; }}
    .code {{ font-size: 32px; font-weight: 800; letter-spacing: 4px; color: #7ea8ff; }}
    .desc {{ margin: 10px 0 0; color: #d1d5db; font-size: 14px; line-height: 1.6; }}
    .note {{ margin-top: 14px; font-size: 13px; color: #a5b4fc; }}
    .footer {{ margin-top: 22px; font-size: 12px; color: #9ca3af; border-top: 1px solid rgba(255,255,255,0.12); padding-top: 12px; }}
  </style>
</head>
<body style="margin:0;padding:0;background:#28272F;">
  <table role="presentation" cellpadding="0" cellspacing="0" width="100%" style="background:#28272F;">
    <tr>
      <td align="center" style="padding:32px 12px;">
        <div class="wrap">
          <div class="container">
            <div class="logo">Lumi</div>
            <div class="hero">
              <img src="https://www.lyrria.com/img/Lumi-cap.png" alt="Lumi" />
            </div>
            <div class="title">您的登录验证码</div>
            <div class="subtitle">请在 5 分钟内完成校验，保护您的账号安全。</div>
            <div class="code-box">
              <div class="code">{code}</div>
            </div>
            <div class="desc">如非本人操作，请忽略本邮件。验证码勿泄露给他人。</div>
            <div class="note">本邮件由系统自动发送，请勿直接回复。</div>
            <div class="footer">© {year} Lyrria. All rights reserved.</div>
          </div>
        </div>
      </td>
    </tr>
  </table>
</body>
</html>
"""


def build_message(
    sender: str,
    recipients: list[str],
    subject: str,
    body_html: str,
) -> MIMEMultipart:
    """构建 HTML 邮件消息"""
    msg = MIMEMultipart("alternative")
    msg["Subject"] = Header(subject, "utf-8")
    msg["From"] = formataddr(["Lyrria Support", sender])
    msg["To"] = ",".join(recipients)
    msg.attach(MIMEText(body_html, "html", "utf-8"))
    return msg


async def send_email_async(
    email: str,
    code: str,
    smtp_username: str,
    smtp_password: str,
) -> Tuple[bool, Optional[str]]:
    """
    发送邮件验证码（异步包装）
    
    Args:
        email: 收件人邮箱
        code: 验证码
        smtp_username: SMTP 用户名
        smtp_password: SMTP 密码
    
    Returns:
        (是否成功, 错误信息)
    """
    try:
        # 构建邮件内容
        body = EMAIL_BODY_TEMPLATE.format(code=code, year=datetime.now().year)
        msg = build_message(
            sender=smtp_username,
            recipients=[email],
            subject=EMAIL_SUBJECT,
            body_html=body,
        )
        
        # 使用 asyncio.to_thread 将同步 SMTP 调用转为异步
        def _send():
            with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT_SSL) as client:
                client.login(smtp_username, smtp_password)
                client.sendmail(smtp_username, [email], msg.as_string())
        
        await asyncio.to_thread(_send)
        
        logger.info(f"✅ 邮件发送成功: email={email}, code_id={code[:4]}***")
        return True, None
        
    except smtplib.SMTPAuthenticationError as e:
        error_msg = f"SMTP authentication error: {e}"
        logger.error(f"❌ 邮件发送认证失败: email={email}, {error_msg}")
        return False, error_msg
    except smtplib.SMTPException as e:
        error_msg = f"SMTP error: {e}"
        logger.error(f"❌ 邮件发送SMTP错误: email={email}, {error_msg}")
        return False, error_msg
    except Exception as e:
        error_msg = f"Unexpected error: {e}"
        logger.error(f"❌ 邮件发送异常: email={email}, {error_msg}")
        return False, error_msg


async def process_email_codes(db_manager: DatabaseManager):
    """扫描并发送邮箱验证码"""
    if not SMTP_USERNAME or not SMTP_PASSWORD:
        logger.error("❌ SMTP 配置缺失，请在 send.conf 文件中设置 smtp_username 和 smtp_password")
        return 0, 0
    
    async with db_manager.get_session() as session:
        # 查询未发送的邮箱验证码
        result = await session.execute(
            select(VerificationCode)
            .where(
                VerificationCode.send_type == "email",
                VerificationCode.is_sent == False,
                VerificationCode.is_valid == True,
                VerificationCode.email.isnot(None)
            )
            .order_by(VerificationCode.created_at.asc())
            .limit(50)  # 每次处理50条
        )
        codes = result.scalars().all()
        
        if not codes:
            return 0, 0
        
        logger.info(f"📋 找到 {len(codes)} 条待发送的邮箱验证码")
        
        success_count = 0
        fail_count = 0
        
        for code_record in codes:
            try:
                # 发送邮件
                success, error_msg = await send_email_async(
                    code_record.email,
                    code_record.code,
                    SMTP_USERNAME,
                    SMTP_PASSWORD
                )
                
                # 🔥 使用显式 UPDATE 语句更新数据库（确保数据正确保存）
                now = datetime.now()
                code_id = code_record.code_id
                
                if success:
                    # 成功：更新 is_sent、sent_at 和 metadata
                    metadata_dict = code_record.metadata_json or {}
                    metadata_dict["send_status"] = "success"
                    metadata_dict["send_at"] = now.isoformat()
                    
                    await session.execute(
                        update(VerificationCode)
                        .where(VerificationCode.code_id == code_id)
                        .values(
                            is_sent=True,
                            sent_at=now,
                            metadata_json=metadata_dict
                        )
                    )
                    await session.commit()
                    success_count += 1
                else:
                    # 失败：更新 metadata，并将验证码标记为失效，避免重复发送
                    metadata_dict = code_record.metadata_json or {}
                    metadata_dict["send_status"] = "failed"
                    metadata_dict["error"] = error_msg
                    metadata_dict["send_at"] = now.isoformat()
                    
                    await session.execute(
                        update(VerificationCode)
                        .where(VerificationCode.code_id == code_id)
                        .values(
                            is_valid=False,
                            metadata_json=metadata_dict,
                        )
                    )
                    await session.commit()
                    fail_count += 1
                
            except Exception as e:
                logger.error(f"❌ 处理验证码失败: code_id={code_record.code_id}, error={e}")
                try:
                    # 异常：更新 metadata，且将验证码标记为失效，避免后续重复重试
                    metadata_dict = code_record.metadata_json or {}
                    metadata_dict["send_status"] = "error"
                    metadata_dict["error"] = str(e)
                    metadata_dict["send_at"] = datetime.now().isoformat()
                    
                    await session.execute(
                        update(VerificationCode)
                        .where(VerificationCode.code_id == code_record.code_id)
                        .values(
                            is_valid=False,
                            metadata_json=metadata_dict,
                        )
                    )
                    await session.commit()
                except Exception as update_error:
                    logger.error(f"❌ 更新错误状态失败: code_id={code_record.code_id}, error={update_error}")
                fail_count += 1
        
        if success_count > 0 or fail_count > 0:
            logger.info(f"📊 处理完成: 成功 {success_count} 条, 失败 {fail_count} 条")
        
        return success_count, fail_count


async def main_loop():
    """主循环：准实时轮询并发送验证码"""
    # 如果配置文件中提供了数据库配置，临时设置环境变量
    if MYSQL_HOST:
        os.environ["MYSQL_HOST"] = MYSQL_HOST
    if MYSQL_PORT:
        os.environ["MYSQL_PORT"] = MYSQL_PORT
    if MYSQL_USER:
        os.environ["MYSQL_USER"] = MYSQL_USER
    if MYSQL_PASSWORD:
        os.environ["MYSQL_PASSWORD"] = MYSQL_PASSWORD
    if MYSQL_DATABASE:
        os.environ["MYSQL_DATABASE"] = MYSQL_DATABASE
    
    db_manager = DatabaseManager()
    
    try:
        # 初始化数据库连接
        await db_manager.connect()
        logger.info("✅ 数据库连接成功")
        
        logger.info(f"🚀 开始准实时发送服务（轮询间隔: {POLL_INTERVAL}秒）...")
        
        # 主循环：持续轮询
        while True:
            try:
                await process_email_codes(db_manager)
            except Exception as e:
                logger.error(f"❌ 处理验证码时出错: {e}", exc_info=True)
            
            # 等待下次轮询
            await asyncio.sleep(POLL_INTERVAL)
            
    except KeyboardInterrupt:
        logger.info("🛑 收到中断信号，正在关闭...")
    except Exception as e:
        logger.error(f"❌ 服务异常: {e}", exc_info=True)
        sys.exit(1)
    finally:
        await db_manager.disconnect()
        logger.info("✅ 服务已关闭")


def setup_logging():
    """配置日志：正常日志输出到 send_email.log，错误日志输出到 send_email_error.log"""
    log_dir = Path(__file__).parent
    log_file = log_dir / "send_email.log"
    error_log_file = log_dir / "send_email_error.log"
    
    # 创建格式化器
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )
    
    # 配置根日志记录器
    root_logger = logging.getLogger()
    root_logger.setLevel(logging.INFO)
    
    # 清除已有的处理器
    root_logger.handlers.clear()
    
    # 1. 控制台处理器（INFO及以上）
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(formatter)
    root_logger.addHandler(console_handler)
    
    # 2. 文件处理器（所有日志，INFO及以上）
    file_handler = logging.FileHandler(log_file, encoding='utf-8')
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(formatter)
    root_logger.addHandler(file_handler)
    
    # 3. 错误日志文件处理器（ERROR及以上）
    error_file_handler = logging.FileHandler(error_log_file, encoding='utf-8')
    error_file_handler.setLevel(logging.ERROR)
    error_file_handler.setFormatter(formatter)
    root_logger.addHandler(error_file_handler)
    
    logger.info(f"✅ 日志配置完成：正常日志 -> {log_file}, 错误日志 -> {error_log_file}")


if __name__ == "__main__":
    setup_logging()
    asyncio.run(main_loop())