xiaoxia-saas/packages/application/auth/verification_code_service.py

"""
验证码服务
- 生成验证码
- 校验验证码
- 频控（60s 冷却 + 每日上限）
"""

from __future__ import annotations

import logging
import re
from datetime import UTC, datetime
from typing import Optional

from packages.domain.verification_code import VerificationCode
from packages.ports.verification_code_repository import VerificationCodeRepository

logger = logging.getLogger(__name__)

# 频控参数
RESEND_COOLDOWN_SECONDS = 60  # 重发冷却时间
DAILY_LIMIT = 10  # 每日发送上限
MAX_ATTEMPTS = 5  # 单验证码最大尝试次数
DEFAULT_TTL_SECONDS = 300  # 默认有效期 5 分钟

# 验证码类型
CODE_TYPE_EMAIL_BIND = "email_bind"
CODE_TYPE_PHONE_BIND = "phone_bind"
CODE_TYPE_EMAIL_LOGIN = "email_login"
CODE_TYPE_PHONE_LOGIN = "phone_login"
CODE_TYPE_RESET_PASSWORD = "reset_password"

VALID_CODE_TYPES = {
    CODE_TYPE_EMAIL_BIND,
    CODE_TYPE_PHONE_BIND,
    CODE_TYPE_EMAIL_LOGIN,
    CODE_TYPE_PHONE_LOGIN,
    CODE_TYPE_RESET_PASSWORD,
}


class VerificationCodeService:
    """验证码服务"""

    def __init__(
        self,
        repo: VerificationCodeRepository,
        resend_cooldown: int = RESEND_COOLDOWN_SECONDS,
        daily_limit: int = DAILY_LIMIT,
        max_attempts: int = MAX_ATTEMPTS,
        default_ttl: int = DEFAULT_TTL_SECONDS,
    ):
        self.repo = repo
        self.resend_cooldown = resend_cooldown
        self.daily_limit = daily_limit
        self.max_attempts = max_attempts
        self.default_ttl = default_ttl

    def generate(
        self,
        recipient: str,
        code_type: str,
        ttl_seconds: int | None = None,
        custom_code: str | None = None,
    ) -> tuple[Optional[VerificationCode], Optional[str]]:
        """
        生成验证码

        Returns:
            (验证码实体, 错误信息)
        """
        recipient = recipient.strip()

        # 参数校验
        if not recipient:
            return None, "接收方不能为空"
        if code_type not in VALID_CODE_TYPES:
            return None, f"无效的验证码类型: {code_type}"

        # 频控检查
        can_send, wait_seconds = self._check_rate_limit(recipient, code_type)
        if not can_send:
            if wait_seconds > 0:
                return None, f"发送太频繁，请 {wait_seconds} 秒后再试"
            return None, "今日发送次数已达上限"

        # 生成并保存
        code = VerificationCode.create(
            recipient=recipient,
            code_type=code_type,
            ttl_seconds=ttl_seconds or self.default_ttl,
            custom_code=custom_code,
        )
        self.repo.save(code)

        return code, None

    def verify(
        self,
        recipient: str,
        code_type: str,
        code_value: str,
        consume: bool = True,
    ) -> tuple[bool, Optional[str]]:
        """
        校验验证码

        Args:
            recipient: 接收方（邮箱/手机号）
            code_type: 验证码类型
            code_value: 用户输入的验证码
            consume: 校验成功后是否标记为已使用

        Returns:
            (是否通过, 错误信息)
        """
        recipient = recipient.strip()
        code_value = code_value.strip()

        if not recipient or not code_value:
            return False, "参数不完整"

        # 查找最新的验证码
        latest = self.repo.find_latest(recipient, code_type)
        if not latest:
            return False, "验证码不存在或已过期"

        # 增加尝试次数
        latest.increment_attempts()
        self.repo.save(latest)

        # 检查是否已使用
        if latest.is_used:
            return False, "验证码已使用，请重新获取"

        # 检查是否过期
        if latest.is_expired:
            return False, "验证码已过期，请重新获取"

        # 检查尝试次数
        if latest.attempts > self.max_attempts:
            return False, "验证次数过多，请重新获取验证码"

        # 校验验证码
        if latest.code != code_value:
            return False, "验证码错误"

        # 校验通过，标记为已使用
        if consume:
            latest.mark_used()
            self.repo.save(latest)

        return True, None

    def _check_rate_limit(self, recipient: str, code_type: str) -> tuple[bool, int]:
        """
        频控检查

        Returns:
            (是否允许发送, 需等待秒数)
        """
        # 检查冷却时间
        latest = self.repo.find_latest(recipient, code_type)
        if latest:
            elapsed = (datetime.now(UTC) - latest.created_at).total_seconds()
            if elapsed < self.resend_cooldown:
                wait = int(self.resend_cooldown - elapsed)
                return False, wait

        # 检查每日上限
        today_count = self.repo.count_today(recipient, code_type)
        if today_count >= self.daily_limit:
            return False, 0

        return True, 0


def validate_phone(phone: str) -> tuple[bool, str]:
    """校验手机号格式（中国大陆手机号）"""
    phone = phone.strip()
    if not phone:
        return False, "手机号不能为空"
    # 支持 +86 前缀或纯 11 位
    pattern = r"^(\+86)?1[3-9]\d{9}$"
    if not re.match(pattern, phone):
        return False, "手机号格式不正确"
    return True, ""


def normalize_phone(phone: str) -> str:
    """标准化手机号（去掉 +86 前缀，统一存储格式）"""
    phone = phone.strip()
    if phone.startswith("+86"):
        phone = phone[3:]
    return phone


def validate_email(email: str) -> tuple[bool, str]:
    """校验邮箱格式"""
    email = email.strip()
    if not email:
        return False, "邮箱不能为空"
    pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
    if not re.match(pattern, email):
        return False, "邮箱格式不正确"
    return True, ""