107 lines
2.9 KiB
Python
107 lines
2.9 KiB
Python
|
# 阿里云短信服务
|
|||
|
|
import os
|
|||
|
|
import random
|
|||
|
|
import string
|
|||
|
|
import time
|
|||
|
|
from datetime import datetime, timedelta
|
|||
|
|
from typing import Optional
|
|||
|
|
|
|||
|
|
# 阿里云短信配置
|
|||
|
|
SMS_CONFIG = {
|
|||
|
|
"access_key_id": os.getenv("SMS_ACCESS_KEY_ID", "LTAI5t6HUnpFBLEK9194kPVG"),
|
|||
|
|
"access_key_secret": os.getenv("SMS_ACCESS_KEY_SECRET", "LEr4Q8yRxb8D5b24cKfCwlt4MMoke1"),
|
|||
|
|
"sign_name": "阿里云",
|
|||
|
|
"template_code": "100001",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 验证码缓存（生产环境建议用Redis）
|
|||
|
|
# 格式: { phone: { code: "123456", expire: 1234567890 } }
|
|||
|
|
VERIFICATION_CODES = {}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def generate_code(length: int = 6) -> str:
|
|||
|
|
"""生成6位数字验证码"""
|
|||
|
|
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(6)
|
|||
|
|
|
|||
|
|
# 配置客户端
|
|||
|
|
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 # 5分钟有效
|
|||
|
|
}
|
|||
|
|
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
|
|||
|
|
|
|||
|
|
stored = VERIFICATION_CODES[phone]
|
|||
|
|
|
|||
|
|
# 检查是否过期
|
|||
|
|
if int(time.time()) > stored["expire"]:
|
|||
|
|
del VERIFICATION_CODES[phone]
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
# 验证码匹配
|
|||
|
|
if stored["code"] == code:
|
|||
|
|
# 验证成功，删除验证码
|
|||
|
|
del VERIFICATION_CODES[phone]
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
def check_code_exists(phone: str) -> bool:
|
|||
|
|
"""检查是否已发送过验证码"""
|
|||
|
|
return phone in VERIFICATION_CODES
|