180 lines
6.9 KiB
Python
180 lines
6.9 KiB
Python
import jingrow
|
|
import requests
|
|
import time
|
|
import hmac
|
|
import hashlib
|
|
import base64
|
|
import json
|
|
from urllib.parse import quote
|
|
import random
|
|
|
|
class AliyunSMSClient:
|
|
"""阿里云短信客户端，采用单例模式"""
|
|
_instance = None
|
|
|
|
@classmethod
|
|
def get_instance(cls):
|
|
"""获取单例实例"""
|
|
if cls._instance is None:
|
|
cls._instance = cls()
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
"""初始化阿里云短信配置"""
|
|
if AliyunSMSClient._instance is not None:
|
|
raise Exception("请使用 AliyunSMSClient.get_instance() 获取实例")
|
|
|
|
# 初始化配置
|
|
self.initialize()
|
|
|
|
def initialize(self):
|
|
"""初始化配置信息"""
|
|
try:
|
|
# 直接尝试获取 Jcloud Settings
|
|
settings = jingrow.get_single("Jcloud Settings")
|
|
if settings:
|
|
self.access_key_id = settings.get("aliyun_access_key_id")
|
|
self.access_secret = settings.get_password("aliyun_access_secret") if settings.get("aliyun_access_secret") else None
|
|
else:
|
|
# 设置默认值
|
|
self.access_key_id = None
|
|
self.access_secret = None
|
|
jingrow.log_error("阿里云SMS客户端: Jcloud Settings 尚未配置，请在设置中完成配置")
|
|
except Exception as e:
|
|
jingrow.log_error(f"阿里云SMS客户端初始化: {str(e)}")
|
|
self.access_key_id = None
|
|
self.access_secret = None
|
|
|
|
def send_sms(self, phone_numbers, template_code, template_param, sign_name):
|
|
"""发送短信的主方法"""
|
|
if not self.access_key_id or not self.access_secret:
|
|
jingrow.log_error("阿里云凭据未设置，无法发送短信")
|
|
return {"status": "error", "message": "缺少阿里云访问凭据"}
|
|
|
|
# 确保接收号码是字符串格式
|
|
if isinstance(phone_numbers, list):
|
|
phone_numbers = ','.join(phone_numbers)
|
|
|
|
# 确保模板参数是JSON字符串
|
|
if isinstance(template_param, dict):
|
|
template_param = json.dumps(template_param)
|
|
|
|
# 阿里云 API 请求参数
|
|
parameters = {
|
|
"Action": "SendSms",
|
|
"Version": "2017-05-25",
|
|
"RegionId": "cn-hangzhou",
|
|
"PhoneNumbers": phone_numbers,
|
|
"SignName": sign_name,
|
|
"TemplateCode": template_code,
|
|
"TemplateParam": template_param,
|
|
"AccessKeyId": self.access_key_id,
|
|
"Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
"SignatureMethod": "HMAC-SHA1",
|
|
"SignatureVersion": "1.0",
|
|
"SignatureNonce": str(int(time.time() * 1000)),
|
|
"Format": "JSON"
|
|
}
|
|
|
|
# 计算签名并添加到参数中
|
|
parameters["Signature"] = self._compute_signature(parameters)
|
|
|
|
# 生成完整的请求 URL
|
|
url = "https://dysmsapi.aliyuncs.com/?" + "&".join(f"{key}={self._percent_encode(value)}" for key, value in parameters.items())
|
|
|
|
try:
|
|
response = requests.get(url)
|
|
try:
|
|
result = response.json()
|
|
except:
|
|
result = {"Code": "ParseError", "Message": "无法解析响应数据"}
|
|
jingrow.log_error(f"无法解析API响应: {response.text}")
|
|
|
|
# 判断发送结果
|
|
if result.get("Code") == "OK":
|
|
return {
|
|
"status": "success",
|
|
"message": "短信发送成功",
|
|
"receiver": phone_numbers,
|
|
"content": template_param
|
|
}
|
|
else:
|
|
jingrow.log_error(f"短信发送失败: {result}")
|
|
return {
|
|
"status": "failed",
|
|
"message": result.get("Message", "发送失败"),
|
|
"receiver": phone_numbers,
|
|
"content": template_param,
|
|
"response": result
|
|
}
|
|
except Exception as e:
|
|
jingrow.log_error(f"短信发送异常: {str(e)}")
|
|
return {
|
|
"status": "error",
|
|
"message": f"发送过程中发生错误: {str(e)}",
|
|
"receiver": phone_numbers,
|
|
"content": template_param
|
|
}
|
|
|
|
def _percent_encode(self, string):
|
|
"""对字符串进行URL编码"""
|
|
if not isinstance(string, str):
|
|
string = str(string)
|
|
return quote(string, safe='')
|
|
|
|
def _compute_signature(self, parameters):
|
|
"""计算签名"""
|
|
sorted_parameters = sorted(parameters.items())
|
|
query_string = '&'.join(f'{self._percent_encode(k)}={self._percent_encode(v)}' for k, v in sorted_parameters)
|
|
string_to_sign = f'GET&%2F&{self._percent_encode(query_string)}'
|
|
h = hmac.new((self.access_secret + "&").encode(), string_to_sign.encode(), hashlib.sha1)
|
|
signature = base64.b64encode(h.digest()).decode()
|
|
return signature
|
|
|
|
# 改为在首次调用时初始化
|
|
sms_client = None
|
|
|
|
def get_sms_client():
|
|
"""获取短信客户端实例，确保在首次调用时才初始化"""
|
|
global sms_client
|
|
if sms_client is None:
|
|
sms_client = AliyunSMSClient.get_instance()
|
|
return sms_client
|
|
|
|
def send_custom_sms(phone_numbers, message_content, sign_name, template_code):
|
|
|
|
client = get_sms_client()
|
|
result = client.send_sms(phone_numbers, template_code, message_content, sign_name)
|
|
|
|
return result
|
|
|
|
def generate_verification_code(length=4):
|
|
"""生成指定长度的随机数字验证码"""
|
|
return ''.join([str(random.randint(0, 9)) for _ in range(length)])
|
|
|
|
def send_verification_code(mobile_no, sign_name, template_code):
|
|
"""生成并发送验证码到指定手机号，并将验证码存储在缓存中"""
|
|
verification_code = generate_verification_code()
|
|
cache_key = f"verification_code:{template_code}:{mobile_no}"
|
|
jingrow.cache().set_value(cache_key, verification_code, expires_in_sec=600) # 验证码有效期10分钟
|
|
|
|
message_content = {"code": verification_code}
|
|
return send_custom_sms(mobile_no, message_content, sign_name, template_code)
|
|
|
|
def verify_code(mobile_no, verification_code, template_code):
|
|
"""验证用户输入的验证码是否正确"""
|
|
cache_key = f"verification_code:{template_code}:{mobile_no}"
|
|
cached_code = jingrow.cache().get_value(cache_key)
|
|
if cached_code and cached_code == verification_code:
|
|
jingrow.cache().delete_value(cache_key)
|
|
return True
|
|
return False
|
|
|
|
def send_renew_sms(phone_numbers, days_remaining, site_end_date):
|
|
|
|
template_code = "SMS_489640674" # 网站续费通知短信模板编码
|
|
sign_name = "向日葵网络" # 短信签名名称
|
|
message_content = {
|
|
"site_end_date": str(site_end_date)
|
|
}
|
|
return send_custom_sms(phone_numbers, message_content, sign_name, template_code) |