170 lines
4.6 KiB
Python
170 lines
4.6 KiB
Python
"""
|
|
阿里云短信服务工具类
|
|
"""
|
|
import json
|
|
import random
|
|
import logging
|
|
from datetime import timedelta
|
|
from django.conf import settings
|
|
from django.utils import timezone
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
try:
|
|
from alibabacloud_dysmsapi20170525.client import Client
|
|
from alibabacloud_tea_openapi.models import Config
|
|
from alibabacloud_tea_util.models import RuntimeOptions
|
|
from alibabacloud_dysmsapi20170525.models import SendSmsRequest
|
|
SMS_SDK_AVAILABLE = True
|
|
except ImportError:
|
|
SMS_SDK_AVAILABLE = False
|
|
|
|
|
|
class SMSClient:
|
|
"""阿里云短信客户端"""
|
|
|
|
_instance = None
|
|
|
|
def __new__(cls):
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
cls._instance._initialized = False
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
if self._initialized:
|
|
return
|
|
|
|
if not SMS_SDK_AVAILABLE:
|
|
self.client = None
|
|
self._initialized = True
|
|
return
|
|
|
|
sms_config = settings.ALIYUN_SMS
|
|
if not sms_config.get('ACCESS_KEY_ID'):
|
|
self.client = None
|
|
self._initialized = True
|
|
return
|
|
|
|
config = Config(
|
|
access_key_id=sms_config['ACCESS_KEY_ID'],
|
|
access_key_secret=sms_config['ACCESS_KEY_SECRET'],
|
|
endpoint='dysmsapi.aliyuncs.com',
|
|
)
|
|
self.client = Client(config)
|
|
self.sign_name = sms_config['SIGN_NAME']
|
|
self.template_code = sms_config['TEMPLATE_CODE']
|
|
self._initialized = True
|
|
|
|
def send_code(self, phone, code):
|
|
"""
|
|
发送验证码短信
|
|
:param phone: 手机号
|
|
:param code: 验证码
|
|
:return: (success: bool, error_msg: str)
|
|
"""
|
|
if not self.client:
|
|
logger.warning('SMS SDK 未配置，验证码: %s -> %s', phone, code)
|
|
return True, ''
|
|
|
|
try:
|
|
request = SendSmsRequest(
|
|
phone_numbers=phone,
|
|
sign_name=self.sign_name,
|
|
template_code=self.template_code,
|
|
template_param=json.dumps({'code': code}),
|
|
)
|
|
runtime = RuntimeOptions()
|
|
response = self.client.send_sms_with_options(request, runtime)
|
|
|
|
if response.body.code == 'OK':
|
|
logger.info('短信发送成功: %s', phone)
|
|
return True, ''
|
|
else:
|
|
msg = response.body.message or response.body.code
|
|
logger.error('短信发送失败: %s, %s', phone, msg)
|
|
return False, msg
|
|
except Exception as e:
|
|
logger.error('短信发送异常: %s, %s', phone, str(e))
|
|
return False, str(e)
|
|
|
|
|
|
def get_sms_client():
|
|
"""获取短信客户端单例"""
|
|
return SMSClient()
|
|
|
|
|
|
def generate_code():
|
|
"""生成随机验证码"""
|
|
length = settings.ALIYUN_SMS.get('CODE_LENGTH', 6)
|
|
return ''.join([str(random.randint(0, 9)) for _ in range(length)])
|
|
|
|
|
|
def send_sms_code(phone):
|
|
"""
|
|
发送短信验证码的完整流程
|
|
:param phone: 手机号
|
|
:return: (success: bool, error_msg: str)
|
|
"""
|
|
from apps.users.models import SmsCode
|
|
|
|
sms_config = settings.ALIYUN_SMS
|
|
interval = sms_config.get('SEND_INTERVAL', 60)
|
|
expire_seconds = sms_config.get('CODE_EXPIRE', 300)
|
|
|
|
# 检查发送频率
|
|
recent = SmsCode.objects.filter(
|
|
phone=phone,
|
|
created_at__gte=timezone.now() - timedelta(seconds=interval),
|
|
).exists()
|
|
if recent:
|
|
return False, '验证码发送过于频繁，请稍后再试'
|
|
|
|
# 生成验证码
|
|
code = generate_code()
|
|
|
|
# 发送短信
|
|
client = get_sms_client()
|
|
ok, err = client.send_code(phone, code)
|
|
if not ok:
|
|
return False, err or '短信发送失败'
|
|
|
|
# 保存到数据库
|
|
SmsCode.objects.create(
|
|
phone=phone,
|
|
code=code,
|
|
expire_at=timezone.now() + timedelta(seconds=expire_seconds),
|
|
)
|
|
|
|
return True, ''
|
|
|
|
|
|
def verify_sms_code(phone, code):
|
|
"""
|
|
校验短信验证码
|
|
:param phone: 手机号
|
|
:param code: 验证码
|
|
:return: (valid: bool, error_msg: str)
|
|
"""
|
|
# DEBUG 模式下，万能验证码跳过校验
|
|
if settings.DEBUG and code == '999999':
|
|
return True, ''
|
|
|
|
from apps.users.models import SmsCode
|
|
|
|
record = SmsCode.objects.filter(
|
|
phone=phone,
|
|
code=code,
|
|
is_used=False,
|
|
expire_at__gt=timezone.now(),
|
|
).order_by('-created_at').first()
|
|
|
|
if not record:
|
|
return False, '验证码无效或已过期'
|
|
|
|
# 标记已使用
|
|
record.is_used = True
|
|
record.save(update_fields=['is_used'])
|
|
|
|
return True, ''
|