101 lines
3.2 KiB
Python
101 lines
3.2 KiB
Python
import logging
|
|
import random
|
|
import secrets
|
|
import string
|
|
import time
|
|
|
|
from app.core.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SMS_CODE_TTL_SECONDS = 300 # 5分钟有效期
|
|
_sms_codes: dict[str, tuple[str, float]] = {}
|
|
|
|
|
|
def _send_sms_aliyun(phone: str, code: str) -> bool:
|
|
"""通过阿里云发送短信"""
|
|
try:
|
|
from alibabacloud_dysmsapi20170525.client import Client
|
|
from alibabacloud_dysmsapi20170525 import models as dysmsapi_models
|
|
from alibabacloud_tea_openapi import models as open_api_models
|
|
|
|
config = open_api_models.Config(
|
|
access_key_id=settings.aliyun_access_key_id,
|
|
access_key_secret=settings.aliyun_access_key_secret,
|
|
)
|
|
config.endpoint = "dysmsapi.aliyuncs.com"
|
|
client = Client(config)
|
|
|
|
request = dysmsapi_models.SendSmsRequest(
|
|
phone_numbers=phone,
|
|
sign_name=settings.aliyun_sms_sign_name,
|
|
template_code=settings.aliyun_sms_template_code,
|
|
template_param=f'{{"code":"{code}"}}',
|
|
)
|
|
response = client.send_sms(request)
|
|
|
|
if response.body.code == "OK":
|
|
logger.info(f"[阿里云短信] 发送成功: {phone}")
|
|
return True
|
|
else:
|
|
logger.error(f"[阿里云短信] 发送失败: {phone}, 错误: {response.body.message}")
|
|
return False
|
|
except ImportError:
|
|
logger.error("[阿里云短信] 未安装依赖，请执行: pip install alibabacloud-dysmsapi20170525")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"[阿里云短信] 发送异常: {phone}, 异常: {str(e)}")
|
|
return False
|
|
|
|
|
|
def _send_sms_local(phone: str, code: str) -> bool:
|
|
"""本地测试模式：将验证码打印到终端"""
|
|
print(f"\n{'='*50}")
|
|
print(f"[短信验证码] 手机号: {phone}")
|
|
print(f"[短信验证码] 验证码: {code}")
|
|
print(f"[短信验证码] 有效期: {SMS_CODE_TTL_SECONDS}秒")
|
|
print(f"{'='*50}\n")
|
|
logger.info(f"[本地短信] 验证码已生成: {phone} -> {code}")
|
|
return True
|
|
|
|
|
|
def send_sms(phone: str, code: str) -> bool:
|
|
"""发送短信验证码（根据配置选择发送方式）"""
|
|
if settings.sms_mode == "aliyun":
|
|
return _send_sms_aliyun(phone, code)
|
|
else:
|
|
return _send_sms_local(phone, code)
|
|
|
|
|
|
def generate_sms_code(phone: str) -> str:
|
|
"""生成短信验证码"""
|
|
clean_expired_codes()
|
|
code = "".join(random.choices(string.digits, k=6))
|
|
sms_id = secrets.token_urlsafe(16)
|
|
_sms_codes[sms_id] = (code, time.time() + SMS_CODE_TTL_SECONDS)
|
|
|
|
# 发送短信
|
|
send_sms(phone, code)
|
|
|
|
return sms_id
|
|
|
|
|
|
def verify_sms_code(sms_id: str, code: str) -> bool:
|
|
"""验证短信验证码"""
|
|
clean_expired_codes()
|
|
record = _sms_codes.pop(sms_id, None)
|
|
if not record:
|
|
return False
|
|
expected, expires_at = record
|
|
if expires_at < time.time():
|
|
return False
|
|
return expected == code.strip()
|
|
|
|
|
|
def clean_expired_codes() -> None:
|
|
"""清理过期的验证码"""
|
|
now = time.time()
|
|
expired = [sms_id for sms_id, (_, expires_at) in _sms_codes.items() if expires_at < now]
|
|
for sms_id in expired:
|
|
_sms_codes.pop(sms_id, None)
|