个人开发者短信验证码接入方案技术分析与实践

优点：完全可控，定制化强
缺点：开发成本高，需要处理通道对接、内容审核等复杂逻辑
适用场景：有充足开发资源的团队项目

优点：稳定性好，通道冗余
缺点：通常需要企业认证，成本相对较高
适用场景：商业化项目，对稳定性要求高

优点：接入简单，支持个人认证，成本较低
缺点：功能相对基础，高级特性有限
适用场景：个人项目、原型验证、中小型应用

典型案例：Spug推送平台等支持个人认证的服务

[前端/客户端] → [业务服务器] → [短信API] → [用户手机]
                     ↓
                [验证码缓存]

import requests
import random
import redis

class SMSService:
    def __init__(self, api_url, redis_client):
        self.api_url = api_url  # 例如: https://push.spug.cc/send/模板ID
        self.redis_client = redis_client
    
    def generate_code(self, length=6):
        """生成随机验证码"""
        return ''.join([str(random.randint(0, 9)) for _ in range(length)])
    
    def send_verification_code(self, phone, app_name="验证码"):
        """发送验证码"""
        code = self.generate_code()
        
        # 调用短信API（以支持个人认证的轻量化服务为例）
        payload = {
            'name': app_name,
            'code': code,
            'targets': phone
        }
        
        try:
            response = requests.post(self.api_url, json=payload, timeout=10)
            result = response.json()
            
            if result.get('error') == 0:
                # 缓存验证码，设置5分钟过期
                cache_key = f"sms_code:{phone}"
                self.redis_client.setex(cache_key, 300, code)
                return {"success": True, "message": "发送成功"}
            else:
                return {"success": False, "message": result.get('message', '发送失败')}
                
        except Exception as e:
            return {"success": False, "message": f"网络错误: {str(e)}"}

def verify_code(self, phone, input_code):
    """验证码校验"""
    cache_key = f"sms_code:{phone}"
    stored_code = self.redis_client.get(cache_key)
    
    if not stored_code:
        return {"success": False, "message": "验证码已过期"}
    
    if stored_code.decode() == input_code:
        # 验证成功后删除缓存
        self.redis_client.delete(cache_key)
        return {"success": True, "message": "验证成功"}
    else:
        return {"success": False, "message": "验证码错误"}

def check_send_limit(self, phone):
    """检查发送频率限制"""
    limit_key = f"sms_limit:{phone}"
    current_count = self.redis_client.get(limit_key)
    
    if current_count and int(current_count) >= 5:
        return {"allow": False, "message": "发送过于频繁，请稍后再试"}
    
    # 更新计数器，设置1小时过期
    self.redis_client.incr(limit_key)
    self.redis_client.expire(limit_key, 3600)
    
    return {"allow": True}

class SMSHelper {
    constructor(apiEndpoint) {
        this.apiEndpoint = apiEndpoint;
        this.countdown = 0;
    }
    
    async sendCode(phone) {
        if (this.countdown > 0) {
            throw new Error(`请等待 ${this.countdown} 秒后重试`);
        }
        
        try {
            const response = await fetch(`${this.apiEndpoint}/send-sms`, {
                method: 'POST',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify({phone})
            });
            
            const result = await response.json();
            
            if (result.success) {
                this.startCountdown(60);
                return result;
            } else {
                throw new Error(result.message);
            }
        } catch (error) {
            throw new Error(`发送失败: ${error.message}`);
        }
    }
    
    startCountdown(seconds) {
        this.countdown = seconds;
        const timer = setInterval(() => {
            this.countdown--;
            if (this.countdown <= 0) {
                clearInterval(timer);
            }
        }, 1000);
    }
}

# 接口鉴权示例
def authenticate_request(request):
    api_key = request.headers.get('X-API-Key')
    timestamp = request.headers.get('X-Timestamp')
    signature = request.headers.get('X-Signature')
    
    # 验证时间戳（防重放攻击）
    if abs(time.time() - int(timestamp)) > 300:
        return False
    
    # 验证签名
    expected_signature = hmac.new(
        SECRET_KEY.encode(),
        f"{api_key}{timestamp}".encode(),
        hashlib.sha256
    ).hexdigest()
    
    return signature == expected_signature

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()
retry_strategy = Retry(
    total=3,
    status_forcelist=[429, 500, 502, 503, 504],
    method_whitelist=["HEAD", "GET", "OPTIONS", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)

import asyncio
import aiohttp

async def send_sms_async(session, phone, code):
    async with session.post(api_url, json=payload) as response:
        return await response.json()

import logging

logger = logging.getLogger('sms_service')

def log_sms_event(phone, event_type, result):
    logger.info(f"SMS {event_type} - Phone: {phone[:3]}****{phone[-4:]} - Result: {result}")

# 实际可用的代码示例
import requests

def send_sms_via_spug(template_id, phone, code, app_name="验证码"):
    """
    使用Spug平台发送短信验证码
    """
    url = f"https://push.spug.cc/send/{template_id}"
    data = {
        'name': app_name,
        'code': code,
        'targets': phone
    }
    
    try:
        response = requests.post(url, json=data, timeout=10)
        return response.json()
    except Exception as e:
        return {"error": 1, "message": f"请求失败: {str(e)}"}

# 使用示例
result = send_sms_via_spug("A27Lxxxxxx", "186xxxx9898", "123456")