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

"""
阿里云短信服务

提供验证码发送功能
"""
import os
import json
import random
import logging
from typing import Optional

logger = logging.getLogger(__name__)

SMS_ACCESS_KEY_ID = os.getenv("SMS_ACCESS_KEY_ID", os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID", ""))
SMS_ACCESS_KEY_SECRET = os.getenv("SMS_ACCESS_KEY_SECRET", os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET", ""))
SMS_SIGN_NAME = os.getenv("SMS_SIGN_NAME", "智创空间")
SMS_TEMPLATE_VERIFY = os.getenv("SMS_TEMPLATE_CODE_VERIFY", "")   # 验证码模板

# 验证码有效期（秒）
CODE_TTL = 300  # 5分钟
# 同一手机号发送间隔（秒）
SEND_INTERVAL = 60


def _get_client():
    from alibabacloud_dysmsapi20170525.client import Client
    from alibabacloud_tea_openapi import models as open_api_models
    config = open_api_models.Config(
        access_key_id=SMS_ACCESS_KEY_ID,
        access_key_secret=SMS_ACCESS_KEY_SECRET,
    )
    config.endpoint = "dysmsapi.aliyuncs.com"
    return Client(config)


def send_sms(phone: str, template_code: str, params: dict) -> bool:
    """发送短信，返回是否成功"""
    try:
        from alibabacloud_dysmsapi20170525 import models as sms_models
        client = _get_client()
        request = sms_models.SendSmsRequest(
            phone_numbers=phone,
            sign_name=SMS_SIGN_NAME,
            template_code=template_code,
            template_param=json.dumps(params, ensure_ascii=False),
        )
        resp = client.send_sms(request)
        if resp.body.code == "OK":
            logger.info(f"短信发送成功: phone={phone}, template={template_code}")
            return True
        else:
            logger.warning(f"短信发送失败: phone={phone}, code={resp.body.code}, msg={resp.body.message}")
            return False
    except Exception as e:
        logger.error(f"短信发送异常: phone={phone}, error={e}")
        return False


class SmsCodeService:
    """验证码服务（基于 Redis）"""

    REDIS_PREFIX_CODE = "sms:code:"
    REDIS_PREFIX_INTERVAL = "sms: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, phone: str) -> tuple[bool, str]:
        """
        发送验证码
        返回 (success, message)
        """
        if not self.redis:
            return False, "短信服务暂不可用"

        # 检查发送间隔
        interval_key = f"{self.REDIS_PREFIX_INTERVAL}{phone}"
        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))

        # 发送短信
        if not SMS_TEMPLATE_VERIFY:
            logger.warning("未配置验证码短信模板 SMS_TEMPLATE_CODE_VERIFY")
            # 开发模式：不实际发送，直接存储
        else:
            ok = send_sms(phone, SMS_TEMPLATE_VERIFY, {"code": code})
            if not ok:
                return False, "短信发送失败，请稍后重试"

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

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

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


# 全局单例
sms_code_service = SmsCodeService()