Maas-Model-platform/backend/app/services/email_service.py

"""
邮件服务

提供邮箱验证码发送功能，基于 SMTP 协议
"""
import os
import random
import logging
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import Optional

logger = logging.getLogger(__name__)

# 邮件配置（从环境变量读取）
SMTP_HOST = os.getenv("SMTP_HOST", "smtp.qq.com")
SMTP_PORT = int(os.getenv("SMTP_PORT", "465"))
SMTP_USER = os.getenv("SMTP_USER", "")
SMTP_PASSWORD = os.getenv("SMTP_PASSWORD", "")  # QQ邮箱授权码
SMTP_FROM_NAME = os.getenv("SMTP_FROM_NAME", "智创空间")
SMTP_USE_SSL = os.getenv("SMTP_USE_SSL", "true").lower() == "true"

# 验证码有效期（秒）
EMAIL_CODE_TTL = 300  # 5分钟
# 同一邮箱发送间隔（秒）
EMAIL_SEND_INTERVAL = 60


def _send_email(to_email: str, subject: str, html_content: str) -> bool:
    """发送邮件，返回是否成功"""
    if not SMTP_USER or not SMTP_PASSWORD:
        logger.warning("未配置 SMTP_USER 或 SMTP_PASSWORD，邮件发送跳过（开发模式）")
        return True  # 开发模式下直接返回成功

    try:
        msg = MIMEMultipart("alternative")
        msg["Subject"] = subject
        msg["From"] = f"{SMTP_FROM_NAME} <{SMTP_USER}>"
        msg["To"] = to_email

        part = MIMEText(html_content, "html", "utf-8")
        msg.attach(part)

        if SMTP_USE_SSL:
            server = smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, timeout=10)
        else:
            server = smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=10)
            server.starttls()

        server.login(SMTP_USER, SMTP_PASSWORD)
        server.sendmail(SMTP_USER, [to_email], msg.as_string())
        server.quit()
        logger.info(f"邮件发送成功: to={to_email}, subject={subject}")
        return True
    except Exception as e:
        logger.error(f"邮件发送失败: to={to_email}, error={e}")
        return False


def _build_code_email(code: str, scene: str) -> tuple[str, str]:
    """构建验证码邮件内容，返回 (subject, html)"""
    scene_map = {
        "register": "注册",
        "login": "登录",
        "reset_password": "重置密码",
    }
    scene_name = scene_map.get(scene, "验证")
    subject = f"【智创空间】{scene_name}验证码"
    html = f"""
    <div style="font-family: Arial, sans-serif; max-width: 480px; margin: 0 auto; padding: 32px; background: #f9fafb; border-radius: 12px;">
      <h2 style="color: #1d4ed8; margin-bottom: 8px;">智创空间</h2>
      <p style="color: #374151; font-size: 15px;">您正在进行<strong>{scene_name}</strong>操作，验证码为：</p>
      <div style="background: #eff6ff; border: 1px solid #bfdbfe; border-radius: 8px; padding: 20px; text-align: center; margin: 20px 0;">
        <span style="font-size: 36px; font-weight: bold; letter-spacing: 8px; color: #1d4ed8;">{code}</span>
      </div>
      <p style="color: #6b7280; font-size: 13px;">验证码有效期 <strong>5 分钟</strong>，请勿泄露给他人。</p>
      <p style="color: #9ca3af; font-size: 12px; margin-top: 24px;">如非本人操作，请忽略此邮件。</p>
    </div>
    """
    return subject, html


class EmailCodeService:
    """邮箱验证码服务（基于 Redis）"""

    REDIS_PREFIX_CODE = "email:code:"
    REDIS_PREFIX_INTERVAL = "email:interval:"

    def __init__(self):
        from app.core.redis import redis_manager
        self._redis_manager = redis_manager

    @property
    def redis(self):
        return self._redis_manager.get_client()

    async def send_code(self, email: str, scene: str = "register") -> tuple[bool, str]:
        """
        发送邮箱验证码
        返回 (success, message)
        """
        if not self.redis:
            return False, "邮件服务暂不可用"

        # 检查发送间隔
        interval_key = f"{self.REDIS_PREFIX_INTERVAL}{email}"
        if await self.redis.exists(interval_key):
            ttl = await self.redis.ttl(interval_key)
            return False, f"请 {ttl} 秒后再试"

        # 生成6位验证码
        code = str(random.randint(100000, 999999))

        # 发送邮件
        subject, html = _build_code_email(code, scene)
        ok = _send_email(email, subject, html)
        if not ok:
            return False, "邮件发送失败，请稍后重试"

        # 存储验证码
        code_key = f"{self.REDIS_PREFIX_CODE}{email}"
        await self.redis.setex(code_key, EMAIL_CODE_TTL, code)
        # 设置发送间隔
        await self.redis.setex(interval_key, EMAIL_SEND_INTERVAL, "1")

        logger.info(f"邮箱验证码已发送: email={email}, scene={scene}, code={code}")
        return True, "验证码已发送"

    async def verify_code(self, email: str, code: str, delete_after: bool = True) -> bool:
        """验证邮箱验证码，验证成功后默认删除"""
        if not self.redis:
            logger.warning(f"Redis 不可用，邮箱验证码验证失败: email={email}")
            return False
        code_key = f"{self.REDIS_PREFIX_CODE}{email}"
        stored = await self.redis.get(code_key)
        logger.info(f"邮箱验证码校验: email={email}, input={code}, stored={stored}")
        if stored and stored == code:
            if delete_after:
                await self.redis.delete(code_key)
            return True
        return False


# 全局单例
email_code_service = EmailCodeService()