最新版阿里云短信服务发送程序（Python版本）2022-11-08_python实现短信定时发送阿里短信

参考链接：
短信 API 官方文档：https://help.aliyun.com/document_detail/419298.html?spm=5176.25163407.help.dexternal.6ff2bb6ercg9x3
发送短信 官方文档：https://help.aliyun.com/document_detail/419273.htm?spm=a2c4g.11186623.0.0.36855d7aRBSwtP
Python SDK 官方文档：https://help.aliyun.com/document_detail/215764.html?spm=a2c4g.11186623.0.0.6a0c4198XsBJNW

# -*- coding: utf-8 -*-
# @version        : 1.3
# @Creaet Time    : 2021/3/15
# @Author         : LZK
# @File           : aliyun_sms.py
# @Software       : PyCharm
# @Update Time    : 2022/11/08
# @Title          : 最新版阿里云短信服务发送程序（Python版本）2022-11-08

"""
环境要求
Python 3
安装 SDK 核心库 OpenAPI ，使用pip安装包依赖:
pip install alibabacloud_tea_openapi
pip install alibabacloud_dysmsapi20170525
"""

import random
import re
from core.exception import CustomException
from alibabacloud_dysmsapi20170525.client import Client as Dysmsapi20170525Client
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
from alibabacloud_tea_util.client import Client as UtilClient
from core.logger import logger
import datetime


class AliyunSMS:

    # 返回错误码对应：
    doc = "https://help.aliyun.com/document_detail/101346.html"

    def __init__(self, telephone: str, code: str = None):
        self.check_telephone_format(telephone)
        self.telephone = telephone
        if not code:
            self.code = self.get_code()
        else:
            self.code = code
        self.access_key = 'access_key'
        self.access_key_secret = 'access_key_secret'
        self.sign_name = '已审核通过的签名'
        self.template_code = '模板CODE'

    def main(self) -> bool:
        """
        主程序入口，同步方式
        """
        client = self.create_client(self.access_key, self.access_key_secret)
        send_sms_request = dysmsapi_20170525_models.SendSmsRequest(
            phone_numbers=self.telephone,
            sign_name=self.sign_name,
            template_code=self.template_code,
            template_param='{"code":"%s"}' % self.code
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印 API 的返回值
            resp = client.send_sms_with_options(send_sms_request, runtime)
            return self.__validation(resp)
        except Exception as error:
            print(error.message)
            # 如有需要，请打印 error
            UtilClient.assert_as_string(error)

    async def main_async(self) -> bool:
        """
        主程序入口，异步方式
        """
        client = self.create_client(self.access_key, self.access_key_secret)
        send_sms_request = dysmsapi_20170525_models.SendSmsRequest(
            phone_numbers=self.telephone,
            sign_name=self.sign_name,
            template_code=self.template_code,
            template_param='{"code":"%s"}' % self.code
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印 API 的返回值
            resp = await client.send_sms_with_options_async(send_sms_request, runtime)
            return self.__validation(resp)
        except Exception as error:
            print(error.message)
            # 如有需要，请打印 error
            UtilClient.assert_as_string(error.message)

    def __validation(self, resp: dysmsapi_20170525_models.SendSmsResponse) -> bool:
        """
        验证结果并返回
        """
        send_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        if resp.body.code == "OK":
            logger.info(f'{send_time} {self.telephone} 短信发送成功，返回code：{resp.body.code}')
            return True
        else:
            logger.error(f'{send_time} {self.telephone} 短信发送失败，返回code：{resp.body.code}，请参考文档：{self.doc}')
            return False

    @staticmethod
    def get_code(length: int = 6, blend: bool = False) -> str:
        """
        随机获取短信验证码
        短信验证码只支持数字，不支持字母及其他符号

        @param length: 验证码长度
        @param blend: 是否 字母+数字 混合
        """
        code = ""  # 创建字符串变量,存储生成的验证码
        for i in range(length):  # 通过for循环控制验证码位数
            num = random.randint(0, 9)  # 生成随机数字0-9
            if blend:  # 需要字母验证码,不用传参,如果不需要字母的,关键字alpha=False
                upper_alpha = chr(random.randint(65, 90))
                lower_alpha = chr(random.randint(97, 122))
                # 随机选择其中一位
                num = random.choice([num, upper_alpha, lower_alpha])
            code = code + str(num)
        return code

    @staticmethod
    def check_telephone_format(telephone: str):
        """
        检查手机号格式是否合法
        """
        REGEX_TELEPHONE = r'^1(3\d|4[4-9]|5[0-35-9]|6[67]|7[013-8]|8[0-9]|9[0-9])\d{8}$'

        if not re.match(REGEX_TELEPHONE, telephone):
            raise CustomException(msg="手机号码格式不正确", code=400)
        return True

    @staticmethod
    def create_client(
            access_key_id: str,
            access_key_secret: str,
    ) -> Dysmsapi20170525Client:
        """
        使用AK&SK初始化账号Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            # 您的 AccessKey ID,
            access_key_id=access_key_id,
            # 您的 AccessKey Secret,
            access_key_secret=access_key_secret
        )
        # 访问的域名
        config.endpoint = f'dysmsapi.aliyuncs.com'
        return Dysmsapi20170525Client(config)


if __name__ == '__main__':
    AliyunSMS("手机号").main()


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158