阿里云验证码短信发送服务搭建（flask)

参考：https://next.api.aliyun.com/api-tools/sdk/Dysmsapi?version=2017-05-25&language=java-async-tea&tab=primer-doc

我们需要思考验证服务一些要求：

1.验证码只能被验证一次，所以需要状态字段
2.验证码有失效时间，超出时间后则失效
3.验证码有限制次数，比如1min只能发送一次，1h只能发送xx次，一天只能发送xx次
4.验证码由random包生成四位随机数字

因此，我们创建数据表

CREATE TABLE `verification_code` (
  `id` int NOT NULL AUTO_INCREMENT,
  `phone` varchar(11) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT '' COMMENT '手机号码',
  `code` varchar(6) DEFAULT '' COMMENT '验证码',
  `channel` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT '' COMMENT '渠道名称',
  `template_code` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT '' COMMENT '模板代码',
  `status` tinyint DEFAULT '10' COMMENT '状态',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP,
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  `type` tinyint DEFAULT '1' COMMENT '类型 1:验证码短信 2:通知短信',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=9 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

项目结构如下：

web.py是我们的主文件

注意作者此处使用了数据库操作方法更改状态，读者可以使用redis等数据库操作

from datetime import datetime, timedelta
 
from flask import request, jsonify
import random

from utils import Sample
from my_app.crud.crud_SMS import CRUDSMS

app = Flask(__name__)


# 统一响应数据格式
def response(code=200, message="请求成功！", data=None):
    res = {
        'code': code,
        'message': message,
        'data': data if data is not None else {}
    }
    return jsonify(res)


# 定义一个错误处理器，捕获所有的异常
@app.errorhandler(Exception)
def handle_error(e):
    return response(400, str(e))


@app.route('/sms_message/verify_code', methods=['POST'])
def verify():
    data = request.get_json()
    phone = data.get('phone')
    code = data.get('code')
    channel = data.get('channel')
    result = CRUDSMS.get(phone, channel)
    current_time = datetime.now()
    # 判断 验证码是否有效
    if result:

        if current_time - result['create_time'] > timedelta(minutes=5):
            return response(401, message="验证码已过期")
        # 验证码是否正确
        if code == result['code']:
            if result['status'] == 99:
                return response(401, message="验证码已被验证过了，请勿重复使用")
            else:
                CRUDSMS.update(result["id"],{"status":99}) #过期失效
                return response(200, message="验证成功")
        else:
            return response(405, message="验证码错误")
    else:
        return response(406, message="该手机未发送过验证码")


@app.route('/sms_message/getVerificationCode', methods=['POST'])
def getVerificationCode():
    data = request.get_json()
    channel = data.get('channel')
    phone = data.get('phone')
    # 随机生成四位数字
    random_number_str = ''.join([str(random.randint(0, 9)) for _ in range(4)])
    template_code = "xxx"

    result = CRUDSMS.get(phone, channel)
    current_time = datetime.now()
    # 判断 验证码是在1min之内发送过
    if result and current_time - result['create_time'] < timedelta(minutes=1):
        return response(401, message="短信1分钟之内已经发送过，请稍后再试")
    # 判断用户最近一小时发过几次短信
    # 计算过去 1 小时的时间点
    one_hour_ago = current_time - timedelta(hours=1)
    one_day_ago = current_time - timedelta(days=1)
    past_hour_message_num = CRUDSMS.count(phone, channel, one_hour_ago)
    past_day_message_num = CRUDSMS.count(phone, channel, one_day_ago)
    if past_hour_message_num > 5:
        return response(402, message="短信1小时之内已经发送过5条，超出上限，请稍后再试")
    if past_day_message_num > 10:
        return response(403, message="短信1天之内已经发送过10条，超出上限，请稍后再试")
    result = Sample.getVerificationCode(template_code, phone, random_number_str)
    if result:
        CRUDSMS.create(
            {"type": 1, "channel": channel, "phone": phone, "code": random_number_str, "template_code": template_code,
             "create_time": datetime.now().strftime('%Y-%m-%d %H:%M:%S')})
        return response(200, data=random_number_str)

    return response(400, message="短信发送失败")


@app.route('/sms_message/sendNotice', methods=['POST'])
def sendNotice():
    # 从请求中获取union_id
    data = request.get_json()
    template_code = data.get('template_code')
    channel = data.get('channel')
    template_param = data.get('template_param','')
    phone = data.get('phone')
    result = Sample.sendNotice(template_code, phone, template_param)
    if result:
        # 存入数据库
        CRUDSMS.create({"type": 2, "channel": channel, "phone": phone, "code": "", "template_code": template_code,
                        "create_time": datetime.now().strftime('%Y-%m-%d %H:%M:%S')})
        return response(200)

    return response(400, message="短信发送失败")


if __name__ == '__main__':
    app.run(debug=True, host="0.0.0.0", port=5000)

utils.py文件是操作阿里云接口的方法:
填写sign_name、 ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET

# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
import os
import sys

from typing import List

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 logger import normal_log,error_log
ALIBABA_CLOUD_ACCESS_KEY_ID = ""
ALIBABA_CLOUD_ACCESS_KEY_SECRET = ""
class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> Dysmsapi20170525Client:
        """
        使用AK&SK初始化账号Client
        @return: Client
        @throws Exception
        """
        # 工程代码泄露可能会导致 AccessKey 泄露，并威胁账号下所有资源的安全性。以下代码示例仅供参考。
        # 建议使用更安全的 STS 方式，更多鉴权访问方式请参见：https://help.aliyun.com/document_detail/378659.html。
        config = open_api_models.Config(
            # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_ID。,
            access_key_id=ALIBABA_CLOUD_ACCESS_KEY_ID,
            # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_SECRET。,
            access_key_secret=ALIBABA_CLOUD_ACCESS_KEY_SECRET
        )
        # Endpoint 请参考 https://api.aliyun.com/product/Dysmsapi
        config.endpoint = f'dysmsapi.aliyuncs.com'
        return Dysmsapi20170525Client(config)

    @staticmethod
    def getVerificationCode(template_code,phone_numbers,code):
        client = Sample.create_client()
        send_sms_request = dysmsapi_20170525_models.SendSmsRequest(
            phone_numbers=phone_numbers,
            sign_name='xxx',
            template_code = template_code,
            template_param  = '{"code":"%s"}'%code,
        )
        try:
            # 复制代码运行请自行打印 API 的返回值
            client.send_sms_with_options(send_sms_request, util_models.RuntimeOptions())
            normal_log.logger.info("{} 发送验证码成功".format(phone_numbers))
            return True

        except Exception as error:
            # 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
            # 错误 message
            print("{} 发送验证码失败{}".format(phone_numbers, str(error)))
            return False


    @staticmethod
    def sendNotice(
            template_code, phone_numbers,template_param
    ):
        client = Sample.create_client()
        send_sms_request = dysmsapi_20170525_models.SendSmsRequest(
            phone_numbers=phone_numbers,
            sign_name='',
            template_code=template_code,
            template_param=template_param
        )
        try:
            # 复制代码运行请自行打印 API 的返回值
            client.send_sms_with_options(send_sms_request, util_models.RuntimeOptions()) 
            return True


        except Exception as error:
            ...