life-echo/api/services/sms_service.py

"""
腾讯云短信服务
"""
import os
import random
import uuid
from datetime import timedelta
from typing import Optional, Tuple

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from tencentcloud.common import credential
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
from tencentcloud.sms.v20210111 import sms_client, models as sms_models

from database.models import SmsVerificationCode, utc_now

# 腾讯云短信配置
TENCENT_SMS_SECRET_ID = os.getenv("TENCENT_SMS_SECRET_ID", "")
TENCENT_SMS_SECRET_KEY = os.getenv("TENCENT_SMS_SECRET_KEY", "")
TENCENT_SMS_SDK_APP_ID = os.getenv("TENCENT_SMS_SDK_APP_ID", "")
TENCENT_SMS_SIGN_NAME = os.getenv("TENCENT_SMS_SIGN_NAME", "")
# 统一使用一个短信模板ID（所有场景共用）
TENCENT_SMS_TEMPLATE_ID = os.getenv("TENCENT_SMS_TEMPLATE_ID", "")
# 模板参数配置：1=仅验证码，2=验证码+过期时间（默认2）
TENCENT_SMS_TEMPLATE_PARAM_COUNT = int(os.getenv("TENCENT_SMS_TEMPLATE_PARAM_COUNT", "2"))

# 验证码配置
CODE_LENGTH = 6  # 验证码长度
CODE_EXPIRE_MINUTES = 5  # 验证码过期时间（分钟）
RATE_LIMIT_SECONDS = 60  # 发送频率限制（秒）


def generate_verification_code() -> str:
    """生成6位随机数字验证码"""
    return ''.join([str(random.randint(0, 9)) for _ in range(CODE_LENGTH)])


def get_template_id_by_purpose(purpose: str) -> str:
    """获取短信模板ID（所有场景共用同一个模板）"""
    return TENCENT_SMS_TEMPLATE_ID


def send_sms_via_tencent(phone: str, code: str, purpose: str) -> Tuple[bool, str]:
    """
    通过腾讯云发送短信验证码
    
    Args:
        phone: 手机号
        code: 验证码
        purpose: 用途
        
    Returns:
        Tuple[bool, str]: (是否发送成功, 错误消息)
    """
    # 检查配置是否完整
    if not TENCENT_SMS_SECRET_ID or not TENCENT_SMS_SECRET_KEY:
        error_msg = "腾讯云短信服务配置不完整，请设置 TENCENT_SMS_SECRET_ID 和 TENCENT_SMS_SECRET_KEY 环境变量"
        print(f"错误: {error_msg}")
        return False, error_msg
    
    if not TENCENT_SMS_SDK_APP_ID or not TENCENT_SMS_SIGN_NAME or not TENCENT_SMS_TEMPLATE_ID:
        error_msg = "腾讯云短信服务配置不完整，请设置 TENCENT_SMS_SDK_APP_ID、TENCENT_SMS_SIGN_NAME 和 TENCENT_SMS_TEMPLATE_ID 环境变量"
        print(f"错误: {error_msg}")
        return False, error_msg
    
    # 创建认证对象和客户端（复用）
    cred = credential.Credential(TENCENT_SMS_SECRET_ID, TENCENT_SMS_SECRET_KEY)
    client = sms_client.SmsClient(cred, "ap-guangzhou")
    
    # 尝试发送短信，如果参数不匹配则自动重试另一种配置
    param_configs = [
        (1, [code]),  # 单参数：仅验证码
        (2, [code, str(CODE_EXPIRE_MINUTES)])  # 双参数：验证码+过期时间
    ]
    
    # 优先使用配置的参数数量
    if TENCENT_SMS_TEMPLATE_PARAM_COUNT == 1:
        param_configs = [param_configs[0], param_configs[1]]  # 先试单参数，再试双参数
    else:
        param_configs = [param_configs[1], param_configs[0]]  # 先试双参数，再试单参数
    
    last_error = None
    for param_count, template_params in param_configs:
        try:
            # 创建请求对象
            req = sms_models.SendSmsRequest()
            req.SmsSdkAppId = TENCENT_SMS_SDK_APP_ID
            req.SignName = TENCENT_SMS_SIGN_NAME
            req.TemplateId = get_template_id_by_purpose(purpose)
            req.TemplateParamSet = template_params
            req.PhoneNumberSet = [f"+86{phone}"]
            
            print(f"调试: 尝试发送短信 - 模板ID={req.TemplateId}, 参数数量={param_count}, 参数={template_params}")
            
            # 发送短信
            resp = client.SendSms(req)
            
            # 检查发送结果
            if resp.SendStatusSet and len(resp.SendStatusSet) > 0:
                status = resp.SendStatusSet[0]
                if status.Code == "Ok":
                    print(f"调试: 短信发送成功，使用的参数数量={param_count}")
                    return True, ""
                else:
                    error_msg = f"短信发送失败: {status.Code} - {status.Message}"
                    print(f"调试: {error_msg}")
                    last_error = error_msg
                    
                    # 如果是参数不匹配错误，尝试下一个配置
                    if "TemplateParamSetNotMatchApprovedTemplate" in status.Code or "FailedOperation.TemplateParamSetNotMatchApprovedTemplate" in status.Code:
                        print(f"调试: 参数不匹配，尝试下一个配置...")
                        continue
                    else:
                        return False, error_msg
            
            last_error = "短信发送失败: 未收到有效响应"
            print(f"调试: {last_error}")
            
        except TencentCloudSDKException as e:
            error_msg = f"腾讯云SDK异常: {e}"
            print(f"调试: {error_msg}")
            last_error = error_msg
            
            # 如果是参数不匹配错误，尝试下一个配置
            if "TemplateParamSetNotMatchApprovedTemplate" in str(e) or "FailedOperation.TemplateParamSetNotMatchApprovedTemplate" in str(e):
                print(f"调试: 参数不匹配，尝试下一个配置...")
                continue
            
            # 其他错误直接返回
            if "SmsSdkAppIdVerifyFail" in str(e) or "UnauthorizedOperation.SmsSdkAppIdVerifyFail" in str(e):
                error_msg = "短信服务配置错误: SmsSdkAppId 验证失败，请检查 TENCENT_SMS_SDK_APP_ID 是否正确，以及该 AppId 是否属于当前 API 密钥对应的账户"
            elif "InvalidCredential" in str(e) or "secret id should not be none" in str(e).lower():
                error_msg = "短信服务配置错误: API 密钥无效，请检查 TENCENT_SMS_SECRET_ID 和 TENCENT_SMS_SECRET_KEY 是否正确"
            elif "UnauthorizedOperation" in str(e):
                error_msg = f"短信服务授权失败: {e.message if hasattr(e, 'message') else str(e)}"
            
            return False, error_msg
        except Exception as e:
            error_msg = f"发送短信异常: {str(e)}"
            print(f"调试: {error_msg}")
            return False, error_msg
    
    # 所有配置都失败了
    if "TemplateParamSetNotMatchApprovedTemplate" in str(last_error) or "FailedOperation.TemplateParamSetNotMatchApprovedTemplate" in str(last_error):
        error_msg = f"短信模板参数不匹配: 已尝试单参数和双参数配置，均失败。请检查腾讯云控制台中的模板配置（模板ID: {get_template_id_by_purpose(purpose)}），确认模板实际需要的参数数量和格式，然后设置正确的 TENCENT_SMS_TEMPLATE_PARAM_COUNT 环境变量（1=仅验证码，2=验证码+过期时间）"
    else:
        error_msg = last_error or "短信发送失败"
    
    return False, error_msg


async def check_rate_limit(db: AsyncSession, phone: str) -> Tuple[bool, int]:
    """
    检查发送频率限制
    
    Args:
        db: 数据库会话
        phone: 手机号
        
    Returns:
        Tuple[bool, int]: (是否允许发送, 剩余等待秒数)
    """
    # 查询最近的验证码记录
    stmt = select(SmsVerificationCode).where(
        SmsVerificationCode.phone == phone
    ).order_by(
        SmsVerificationCode.created_at.desc()
    )
    result = await db.execute(stmt)
    recent_code = result.scalar_one_or_none()
    
    if not recent_code:
        return True, 0
    
    # 计算距离上次发送的时间
    now = utc_now()
    time_diff = (now - recent_code.created_at).total_seconds()
    
    if time_diff < RATE_LIMIT_SECONDS:
        remaining = int(RATE_LIMIT_SECONDS - time_diff)
        return False, remaining
    
    return True, 0


async def send_verification_code(
    db: AsyncSession,
    phone: str,
    purpose: str,
    ip_address: Optional[str] = None
) -> Tuple[bool, str, int]:
    """
    发送验证码
    
    Args:
        db: 数据库会话
        phone: 手机号
        purpose: 用途 (register/login/reset_password/change_phone)
        ip_address: 请求IP地址
        
    Returns:
        Tuple[bool, str, int]: (是否成功, 消息, 过期时间秒数)
    """
    # 检查频率限制
    can_send, remaining = await check_rate_limit(db, phone)
    if not can_send:
        return False, f"发送过于频繁，请{remaining}秒后再试", 0
    
    # 生成验证码
    code = generate_verification_code()
    
    # 发送短信
    success, error_msg = send_sms_via_tencent(phone, code, purpose)
    
    if not success:
        # 如果错误消息为空，使用默认消息
        if not error_msg:
            error_msg = "短信发送失败，请稍后重试"
        return False, error_msg, 0
    
    # 保存到数据库
    expires_at = utc_now() + timedelta(minutes=CODE_EXPIRE_MINUTES)
    verification_code = SmsVerificationCode(
        id=str(uuid.uuid4()),
        phone=phone,
        code=code,
        purpose=purpose,
        expires_at=expires_at,
        ip_address=ip_address
    )
    
    db.add(verification_code)
    await db.commit()
    
    return True, "验证码已发送", CODE_EXPIRE_MINUTES * 60


async def verify_code(
    db: AsyncSession,
    phone: str,
    code: str,
    purpose: str
) -> Tuple[bool, str]:
    """
    验证验证码
    
    Args:
        db: 数据库会话
        phone: 手机号
        code: 验证码
        purpose: 用途
        
    Returns:
        Tuple[bool, str]: (是否验证成功, 消息)
    """
    # 查询最新的未使用验证码
    stmt = select(SmsVerificationCode).where(
        SmsVerificationCode.phone == phone,
        SmsVerificationCode.purpose == purpose,
        SmsVerificationCode.is_used == False,
        SmsVerificationCode.is_expired == False
    ).order_by(
        SmsVerificationCode.created_at.desc()
    )
    result = await db.execute(stmt)
    verification = result.scalar_one_or_none()
    
    if not verification:
        return False, "验证码不存在或已使用"
    
    # 检查是否过期
    now = utc_now()
    if now > verification.expires_at:
        verification.is_expired = True
        await db.commit()
        return False, "验证码已过期"
    
    # 验证验证码
    if verification.code != code:
        return False, "验证码错误"
    
    # 标记为已使用
    verification.is_used = True
    verification.verified_at = now
    await db.commit()
    
    return True, "验证成功"


async def cleanup_expired_codes(db: AsyncSession) -> int:
    """
    清理过期的验证码
    
    Args:
        db: 数据库会话
        
    Returns:
        int: 清理的记录数
    """
    from sqlalchemy import delete
    
    now = utc_now()
    
    # 标记过期的验证码
    stmt = select(SmsVerificationCode).where(
        SmsVerificationCode.expires_at < now,
        SmsVerificationCode.is_expired == False
    )
    result = await db.execute(stmt)
    expired_codes = result.scalars().all()
    
    count = 0
    for code in expired_codes:
        code.is_expired = True
        count += 1
    
    await db.commit()
    
    # 删除7天前的记录
    seven_days_ago = now - timedelta(days=7)
    delete_stmt = delete(SmsVerificationCode).where(
        SmsVerificationCode.created_at < seven_days_ago
    )
    result = await db.execute(delete_stmt)
    old_codes_count = result.rowcount
    
    await db.commit()
    
    return count + old_codes_count