参考：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, timedeltafrom flask import request, jsonify
import randomfrom utils import Sample
from my_app.crud.crud_SMS import CRUDSMSapp = 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_iddata = 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 sysfrom typing import Listfrom 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@staticmethoddef 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/Dysmsapiconfig.endpoint = f'dysmsapi.aliyuncs.com'return Dysmsapi20170525Client(config)@staticmethoddef 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 Trueexcept Exception as error:# 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。# 错误 messageprint("{} 发送验证码失败{}".format(phone_numbers, str(error)))return False@staticmethoddef 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 Trueexcept Exception as error:...