270 lines
12 KiB
Python
270 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
||
# This file is auto-generated, don't edit it. Thanks.
|
||
import sys
|
||
import os
|
||
import json
|
||
|
||
from typing import List, Union, Dict, Any
|
||
|
||
from alibabacloud_dysmsapi20170525.client import Client as DysmsapiClient
|
||
from alibabacloud_tea_openapi import models as open_api_models
|
||
from alibabacloud_darabonba_env.client import Client as EnvClient
|
||
from alibabacloud_dysmsapi20170525 import models as dysmsapi_models
|
||
from alibabacloud_tea_util.client import Client as UtilClient
|
||
from alibabacloud_tea_console.client import Client as ConsoleClient
|
||
from alibabacloud_darabonba_string.client import Client as StringClient
|
||
from alibabacloud_darabonba_time.client import Client as TimeClient
|
||
from config import settings
|
||
|
||
|
||
|
||
|
||
class SMS:
|
||
def __init__(self):
|
||
self.AccessKey_ID = "LTAI5tEWmH3robZUGHiWYVnP"
|
||
self.AccessKey_Secret = "7LAVepzjXcnEobqlx7WPrvHaFYYEU3"
|
||
|
||
@staticmethod
|
||
def create_client(
|
||
access_key_id: str,
|
||
access_key_secret: str,
|
||
) -> DysmsapiClient:
|
||
"""
|
||
使用AK&SK初始化账号Client
|
||
"""
|
||
config = open_api_models.Config()
|
||
config.access_key_id = access_key_id
|
||
config.access_key_secret = access_key_secret
|
||
return DysmsapiClient(config)
|
||
|
||
@staticmethod
|
||
def main(self,
|
||
phone_number: str,
|
||
template_param: Union[str, Dict[str, Any]] = None,
|
||
template_code: str = settings.sms_template_code,
|
||
sign_name: str = settings.sms_sign_name,
|
||
**kwargs
|
||
) -> dict:
|
||
try:
|
||
|
||
if self.AccessKey_ID is None or self.AccessKey_Secret is None:
|
||
ConsoleClient.log("警告：使用默认的访问密钥，建议通过环境变量设置您自己的访问密钥")
|
||
|
||
client = SMS.create_client(self.AccessKey_ID, self.AccessKey_Secret)
|
||
|
||
# 设置默认值
|
||
phone_number = phone_number
|
||
# sign_name is passed as argument
|
||
# template_code is passed as argument
|
||
|
||
# 构建参数字典
|
||
params = {}
|
||
if template_param:
|
||
if isinstance(template_param, dict):
|
||
params.update(template_param)
|
||
elif isinstance(template_param, str):
|
||
# 尝试解析 JSON 字符串，如果失败则作为 code 字段的值
|
||
try:
|
||
json_param = json.loads(template_param)
|
||
if isinstance(json_param, dict):
|
||
params.update(json_param)
|
||
else:
|
||
params['code'] = template_param
|
||
except json.JSONDecodeError:
|
||
params['code'] = template_param
|
||
|
||
# 添加额外的动态参数
|
||
if kwargs:
|
||
params.update(kwargs)
|
||
|
||
param_code = json.dumps(params)
|
||
|
||
ConsoleClient.log(f"正在尝试发送短信... 模板代码: {template_code}, 签名: {sign_name}, 参数: {param_code}")
|
||
# 1.发送短信
|
||
send_req = dysmsapi_models.SendSmsRequest(
|
||
phone_numbers=phone_number,
|
||
sign_name=sign_name,
|
||
template_code=template_code,
|
||
template_param=param_code
|
||
)
|
||
try:
|
||
send_resp = client.send_sms(send_req)
|
||
code = send_resp.body.code
|
||
if not UtilClient.equal_string(code, 'OK'):
|
||
error_msg = f'发送失败，错误代码: {code}, 错误信息: {send_resp.body.message}'
|
||
ConsoleClient.log(error_msg)
|
||
return {"success": False, "error_message": error_msg}
|
||
|
||
ConsoleClient.log(f'短信发送成功，业务ID: {send_resp.body.biz_id}')
|
||
biz_id = send_resp.body.biz_id
|
||
|
||
# 2. 等待 10 秒后查询结果
|
||
ConsoleClient.log("等待10秒后查询发送状态...")
|
||
UtilClient.sleep(10000)
|
||
|
||
# 3.查询结果
|
||
phone_nums = StringClient.split(phone_number, ',', -1)
|
||
details = []
|
||
for phone_num in phone_nums:
|
||
query_req = dysmsapi_models.QuerySendDetailsRequest(
|
||
phone_number=UtilClient.assert_as_string(phone_num),
|
||
biz_id=biz_id,
|
||
send_date=TimeClient.format('yyyyMMdd'),
|
||
page_size=10,
|
||
current_page=1
|
||
)
|
||
query_resp = client.query_send_details(query_req)
|
||
dtos = query_resp.body.sms_send_detail_dtos.sms_send_detail_dto
|
||
# 打印结果
|
||
for dto in dtos:
|
||
detail = {
|
||
"phone_num": dto.phone_num,
|
||
"send_status": dto.send_status,
|
||
"receive_date": dto.receive_date,
|
||
"err_code": dto.err_code
|
||
}
|
||
details.append(detail)
|
||
if UtilClient.equal_string(f'{dto.send_status}', '3'):
|
||
ConsoleClient.log(f'{dto.phone_num} 发送成功，接收时间: {dto.receive_date}')
|
||
elif UtilClient.equal_string(f'{dto.send_status}', '2'):
|
||
ConsoleClient.log(f'{dto.phone_num} 发送失败')
|
||
else:
|
||
ConsoleClient.log(f'{dto.phone_num} 正在发送中...')
|
||
|
||
return {"success": True, "biz_id": biz_id, "details": details}
|
||
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
if "NoPermission" in error_msg or "403" in error_msg:
|
||
ConsoleClient.log("权限错误：当前账号没有发送短信的权限")
|
||
# ... (keep existing log messages if possible, but for brevity I might shorten)
|
||
elif "InvalidAccessKeyId" in error_msg:
|
||
ConsoleClient.log("错误：AccessKeyId无效")
|
||
elif "SignatureDoesNotMatch" in error_msg:
|
||
ConsoleClient.log("错误：签名不匹配")
|
||
else:
|
||
ConsoleClient.log(f"发生错误: {error_msg}")
|
||
return {"success": False, "error_message": error_msg}
|
||
except Exception as e:
|
||
ConsoleClient.log(f"程序运行错误: {str(e)}")
|
||
return {"success": False, "error_message": str(e)}
|
||
|
||
@staticmethod
|
||
async def main_async(self,
|
||
phone_number: str,
|
||
template_param: Union[str, Dict[str, Any]] = None,
|
||
template_code: str = settings.sms_template_code,
|
||
sign_name: str = settings.sms_sign_name,
|
||
**kwargs
|
||
) -> None:
|
||
try:
|
||
|
||
|
||
client = SMS.create_client(self.AccessKey_ID, self.AccessKey_Secret)
|
||
|
||
# 设置默认值
|
||
phone_number = phone_number
|
||
# sign_name is passed as argument
|
||
# template_code is passed as argument
|
||
|
||
# 构建参数字典
|
||
params = {}
|
||
if template_param:
|
||
if isinstance(template_param, dict):
|
||
params.update(template_param)
|
||
elif isinstance(template_param, str):
|
||
# 尝试解析 JSON 字符串，如果失败则作为 code 字段的值
|
||
try:
|
||
json_param = json.loads(template_param)
|
||
if isinstance(json_param, dict):
|
||
params.update(json_param)
|
||
else:
|
||
params['code'] = template_param
|
||
except json.JSONDecodeError:
|
||
params['code'] = template_param
|
||
|
||
# 添加额外的动态参数
|
||
if kwargs:
|
||
params.update(kwargs)
|
||
|
||
param_code = json.dumps(params)
|
||
|
||
# 如果提供了命令行参数，则使用命令行参数
|
||
|
||
|
||
|
||
ConsoleClient.log(f"正在尝试发送短信... 模板代码: {template_code}, 签名: {sign_name}, 参数: {param_code}")
|
||
# 1.发送短信
|
||
send_req = dysmsapi_models.SendSmsRequest(
|
||
phone_numbers=phone_number,
|
||
sign_name=sign_name,
|
||
template_code=template_code,
|
||
template_param=param_code
|
||
)
|
||
try:
|
||
send_resp = await client.send_sms_async(send_req)
|
||
code = send_resp.body.code
|
||
if not UtilClient.equal_string(code, 'OK'):
|
||
ConsoleClient.log(f'发送失败，错误代码: {code}, 错误信息: {send_resp.body.message}')
|
||
return
|
||
|
||
ConsoleClient.log(f'短信发送成功，业务ID: {send_resp.body.biz_id}')
|
||
biz_id = send_resp.body.biz_id
|
||
|
||
# 2. 等待 10 秒后查询结果
|
||
ConsoleClient.log("等待10秒后查询发送状态...")
|
||
await UtilClient.sleep_async(10000)
|
||
|
||
# 3.查询结果
|
||
phone_nums = StringClient.split(phone_number, ',', -1)
|
||
for phone_num in phone_nums:
|
||
query_req = dysmsapi_models.QuerySendDetailsRequest(
|
||
phone_number=UtilClient.assert_as_string(phone_num),
|
||
biz_id=biz_id,
|
||
send_date=TimeClient.format('yyyyMMdd'),
|
||
page_size=10,
|
||
current_page=1
|
||
)
|
||
query_resp = await client.query_send_details_async(query_req)
|
||
dtos = query_resp.body.sms_send_detail_dtos.sms_send_detail_dto
|
||
# 打印结果
|
||
for dto in dtos:
|
||
if UtilClient.equal_string(f'{dto.send_status}', '3'):
|
||
ConsoleClient.log(f'{dto.phone_num} 发送成功，接收时间: {dto.receive_date}')
|
||
elif UtilClient.equal_string(f'{dto.send_status}', '2'):
|
||
ConsoleClient.log(f'{dto.phone_num} 发送失败')
|
||
else:
|
||
ConsoleClient.log(f'{dto.phone_num} 正在发送中...')
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
if "NoPermission" in error_msg or "403" in error_msg:
|
||
ConsoleClient.log("权限错误：当前账号没有发送短信的权限")
|
||
ConsoleClient.log("可能的原因：")
|
||
ConsoleClient.log("1. 访问密钥无效或已过期")
|
||
ConsoleClient.log("2. 账号未开通短信服务或权限不足")
|
||
ConsoleClient.log("3. 如果使用子账号，需要主账号授予相应权限")
|
||
ConsoleClient.log("解决方法：")
|
||
ConsoleClient.log("1. 登录阿里云控制台，检查账号短信服务开通状态")
|
||
ConsoleClient.log("2. 确认访问密钥是否有效")
|
||
ConsoleClient.log("3. 如需帮助，请联系阿里云客服")
|
||
elif "InvalidAccessKeyId" in error_msg:
|
||
ConsoleClient.log("错误：AccessKeyId无效")
|
||
ConsoleClient.log("请检查您的AccessKeyId是否正确")
|
||
elif "SignatureDoesNotMatch" in error_msg:
|
||
ConsoleClient.log("错误：签名不匹配")
|
||
ConsoleClient.log("请检查您的AccessKeySecret是否正确")
|
||
else:
|
||
ConsoleClient.log(f"发生错误: {error_msg}")
|
||
except Exception as e:
|
||
ConsoleClient.log(f"程序运行错误: {str(e)}")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sms = SMS()
|
||
# 示例 1: 原有方式，仅传递 code
|
||
# sms.main(self=sms, phone_number="18585164448", template_param="6666")
|
||
|
||
# 示例 2: 动态参数传递 (phone, price 等)
|
||
sms.main(self=sms, phone_number="18585164448", code="6666")
|
||
|