jiachenlong/backend/app/services/sms.py

# 阿里云短信服务
import os
import random
import string
import time
from datetime import datetime, timedelta
from typing import Optional
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 阿里云短信配置
SMS_CONFIG = {
    "access_key_id": os.getenv("SMS_ACCESS_KEY_ID", "LTAI5tQAx5niD7JQVqGE5acE"),
    "access_key_secret": os.getenv("SMS_ACCESS_KEY_SECRET", "QsQFAEKBkaNynIoKyvdIi3BUyWVZu1"),
    "sign_name": os.getenv("SMS_SIGN_NAME", "苏州算力"),
    "template_code": os.getenv("SMS_TEMPLATE_CODE", "SMS_501590956"),
}

# 验证码缓存
VERIFICATION_CODES = {}

def generate_code(length: int = 6) -> str:
    return ''.join(random.choices(string.digits, k=length))

def send_verification_code(phone: str) -> dict:
    from alibabacloud_dysmsapi20170525 import models
    from alibabacloud_dysmsapi20170525.client import Client
    from alibabacloud_tea_openapi import models as open_models
    
    try:
        code = generate_code()
        
        # 配置客户端
        config = open_models.Config(
            access_key_id=SMS_CONFIG["access_key_id"],
            access_key_secret=SMS_CONFIG["access_key_secret"],
        )
        config.endpoint = "dysmsapi.aliyuncs.com"
        config.region_id = "cn-hangzhou"
        
        client = Client(config)
        
        # 构造请求
        request = models.SendSmsRequest(
            phone_numbers=phone,
            sign_name=SMS_CONFIG["sign_name"],
            template_code=SMS_CONFIG["template_code"],
            template_param=f'{{"code":"{code}"}}'
        )
        
        # 发送
        response = client.send_sms(request)
        
        # 检查结果
        if response.body.code == "OK":
            VERIFICATION_CODES[phone] = {
                "code": code,
                "expire": int(time.time()) + 300
            }
            return {
                "success": True,
                "message": "验证码已发送",
                "expire": 300
            }
        else:
            return {
                "success": False,
                "message": f"发送失败: {response.body.message}"
            }
    except Exception as e:
        return {
            "success": False,
            "message": f"发送失败: {str(e)}"
        }

def verify_code(phone: str, code: str) -> bool:
    if phone not in VERIFICATION_CODES:
        return False
    
    cached = VERIFICATION_CODES[phone]
    if int(time.time()) > cached["expire"]:
        del VERIFICATION_CODES[phone]
        return False
    
    if cached["code"] == code:
        del VERIFICATION_CODES[phone]
        return True
    
    return False