204 lines
7.5 KiB
Python
204 lines
7.5 KiB
Python
"""
|
||
阿里云短信服务模块
|
||
用于发送手机验证码
|
||
"""
|
||
import json
|
||
import random
|
||
import sys
|
||
from typing import Dict, Any, Optional
|
||
from datetime import datetime, timedelta
|
||
|
||
from alibabacloud_dysmsapi20170525.client import Client as Dysmsapi20170525Client
|
||
from alibabacloud_credentials.client import Client as CredentialClient
|
||
from alibabacloud_credentials.models import Config as CredentialConfig
|
||
from alibabacloud_tea_openapi import models as open_api_models
|
||
from alibabacloud_dysmsapi20170525 import models as dysmsapi_20170525_models
|
||
from alibabacloud_tea_util import models as util_models
|
||
|
||
|
||
class AliSmsService:
|
||
"""阿里云短信服务"""
|
||
|
||
def __init__(self, access_key_id: str, access_key_secret: str, sign_name: str, template_code: str):
|
||
"""
|
||
初始化阿里云短信服务
|
||
|
||
Args:
|
||
access_key_id: 阿里云AccessKey ID
|
||
access_key_secret: 阿里云AccessKey Secret
|
||
sign_name: 短信签名
|
||
template_code: 短信模板CODE
|
||
"""
|
||
self.sign_name = sign_name
|
||
self.template_code = template_code
|
||
|
||
# 创建阿里云短信客户端
|
||
credential_config = CredentialConfig(
|
||
type='access_key',
|
||
access_key_id=access_key_id,
|
||
access_key_secret=access_key_secret
|
||
)
|
||
credential = CredentialClient(credential_config)
|
||
config = open_api_models.Config(credential=credential)
|
||
config.endpoint = 'dysmsapi.aliyuncs.com'
|
||
|
||
self.client = Dysmsapi20170525Client(config)
|
||
|
||
# 验证码缓存（简单内存存储，生产环境应使用Redis）
|
||
self._code_cache: Dict[str, Dict[str, Any]] = {}
|
||
|
||
# 验证码配置
|
||
self.code_length = 6 # 验证码长度
|
||
self.code_expire_minutes = 5 # 验证码过期时间（分钟）
|
||
|
||
def _generate_code(self) -> str:
|
||
"""生成随机验证码"""
|
||
return ''.join([str(random.randint(0, 9)) for _ in range(self.code_length)])
|
||
|
||
async def send_verification_code(self, phone: str) -> Dict[str, Any]:
|
||
"""
|
||
发送验证码到指定手机号
|
||
|
||
Args:
|
||
phone: 手机号
|
||
|
||
Returns:
|
||
Dict containing success status and error message if any
|
||
"""
|
||
try:
|
||
# 生成验证码
|
||
code = self._generate_code()
|
||
|
||
print(f"[短信服务] 正在发送验证码到 {phone}，验证码: {code}", file=sys.stderr)
|
||
|
||
# 构建短信请求
|
||
send_sms_request = dysmsapi_20170525_models.SendSmsRequest(
|
||
phone_numbers=phone,
|
||
sign_name=self.sign_name,
|
||
template_code=self.template_code,
|
||
template_param=json.dumps({"code": code})
|
||
)
|
||
|
||
runtime = util_models.RuntimeOptions()
|
||
|
||
# 发送短信
|
||
try:
|
||
resp = self.client.send_sms_with_options(send_sms_request, runtime)
|
||
resp_dict = resp.to_map()
|
||
|
||
print(f"[短信服务] 阿里云响应: {json.dumps(resp_dict, default=str, indent=2, ensure_ascii=False)}", file=sys.stderr)
|
||
|
||
# 检查发送结果
|
||
if resp_dict.get('body', {}).get('Code') == 'OK':
|
||
# 缓存验证码
|
||
self._code_cache[phone] = {
|
||
'code': code,
|
||
'expire_time': datetime.now() + timedelta(minutes=self.code_expire_minutes),
|
||
'sent_at': datetime.now()
|
||
}
|
||
|
||
print(f"[短信服务] 验证码发送成功，手机号: {phone}", file=sys.stderr)
|
||
|
||
return {
|
||
"success": True,
|
||
"message": f"验证码已发送，{self.code_expire_minutes}分钟内有效",
|
||
"code": code # 开发环境返回验证码，生产环境应移除
|
||
}
|
||
else:
|
||
error_msg = resp_dict.get('body', {}).get('Message', '未知错误')
|
||
print(f"[短信服务] 发送失败: {error_msg}", file=sys.stderr)
|
||
return {
|
||
"success": False,
|
||
"error": f"短信发送失败: {error_msg}"
|
||
}
|
||
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
print(f"[短信服务] 发送异常: {error_msg}", file=sys.stderr)
|
||
|
||
# 如果有诊断地址，打印出来
|
||
if hasattr(e, 'data') and e.data:
|
||
recommend = e.data.get('Recommend')
|
||
if recommend:
|
||
print(f"[短信服务] 诊断地址: {recommend}", file=sys.stderr)
|
||
|
||
return {
|
||
"success": False,
|
||
"error": f"短信发送异常: {error_msg}"
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"[短信服务] 发送验证码失败: {str(e)}", file=sys.stderr)
|
||
return {
|
||
"success": False,
|
||
"error": str(e)
|
||
}
|
||
|
||
def verify_code(self, phone: str, code: str) -> Dict[str, Any]:
|
||
"""
|
||
验证手机号和验证码
|
||
|
||
Args:
|
||
phone: 手机号
|
||
code: 用户输入的验证码
|
||
|
||
Returns:
|
||
Dict containing verification result
|
||
"""
|
||
try:
|
||
# 检查验证码是否存在
|
||
if phone not in self._code_cache:
|
||
return {
|
||
"success": False,
|
||
"error": "验证码未发送或已过期，请重新获取"
|
||
}
|
||
|
||
cached_data = self._code_cache[phone]
|
||
|
||
# 检查是否过期
|
||
if datetime.now() > cached_data['expire_time']:
|
||
# 删除过期验证码
|
||
del self._code_cache[phone]
|
||
return {
|
||
"success": False,
|
||
"error": "验证码已过期，请重新获取"
|
||
}
|
||
|
||
# 验证码匹配
|
||
if code == cached_data['code']:
|
||
# 验证成功后删除验证码（一次性使用）
|
||
del self._code_cache[phone]
|
||
|
||
print(f"[短信服务] 验证码验证成功，手机号: {phone}", file=sys.stderr)
|
||
|
||
return {
|
||
"success": True,
|
||
"message": "验证码验证成功"
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"error": "验证码错误，请重新输入"
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"[短信服务] 验证码验证失败: {str(e)}", file=sys.stderr)
|
||
return {
|
||
"success": False,
|
||
"error": str(e)
|
||
}
|
||
|
||
def cleanup_expired_codes(self):
|
||
"""清理过期的验证码"""
|
||
current_time = datetime.now()
|
||
expired_phones = [
|
||
phone for phone, data in self._code_cache.items()
|
||
if current_time > data['expire_time']
|
||
]
|
||
|
||
for phone in expired_phones:
|
||
del self._code_cache[phone]
|
||
|
||
if expired_phones:
|
||
print(f"[短信服务] 已清理 {len(expired_phones)} 个过期验证码", file=sys.stderr)
|