"""发送邮箱验证码服务(准实时) 从数据库读取待发送的邮箱验证码,使用阿里云 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 = """ Lyrria 验证码
Lumi
您的登录验证码
请在 5 分钟内完成校验,保护您的账号安全。
{code}
如非本人操作,请忽略本邮件。验证码勿泄露给他人。
本邮件由系统自动发送,请勿直接回复。
""" 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())