- 发送短信平台非常多,本次以容联云平台发送短信为例
一、短信发送模块开发
1、容联云注册用户(免费注册,完成个人认证)
2、创建应用,并复制APP ID 和 APP TOKEN
3、短信模板
- 如果实际应用需要创建短信模板,仅开发测试不用创建新短信模板,就用使用系统提供的测试模板
- 项目以此模板为例进行开发测试
4、添加测试号码:
- 没用创建短信模板,只能给添加测试号码发送短信
- 用于开发测试使用,最多添加3个手机号
5、参考开发文档,开发演示代码
5.1 查看开发开发文档
5.2 复制调用示例,编写ronglianyun_SMS_demo.py
- appId : 创建应用时系统给的ID(编辑应用时,也会出现)
- accId、accToken : 进应用管理复制即可,注意保密
- tid :容联云通讯创建的模板ID,测试模板id=“1”
- datas :模板中定义的参数
from ronglian_sms_sdk import SmsSDK
accId = '容联云通讯分配的主账号ID'
accToken = '容联云通讯分配的主账号TOKEN'
appId = '容联云通讯分配的应用ID'
def send_message():
sdk = SmsSDK(accId, accToken, appId)
tid = '1' # 应用是系统提供的开发测试模板及编号
mobile = '手机号1,手机号2'
datas = ('变量1', '变量2') # 测试模板变量1:短信验证码,变量2:有效时间(分钟)
resp = sdk.sendMessage(tid, mobile, datas)
print(resp) # resp 是string类型,还是json类型
send_message()
6、开发短信发送模块
- 在libs包下建立ronglianyun包,并建 ccp_SMS.py
from ronglian_sms_sdk import SmsSDK
accId = '容联云通讯分配的主账号ID'
accToken = '容联云通讯分配的主账号TOKEN'
appId = '容联云通讯分配的应用ID'
def send_message():
sdk = SmsSDK(accId, accToken, appId)
tid = '1' # 应用是系统提供的开发测试模板及编号
mobile = '手机号1,手机号2'
datas = ('变量1', '变量2')
resp = sdk.sendMessage(tid, mobile, datas)
result=json.load(resp)
if result["statusCode"]=="000000":
return 0
else:
return -1
if __name__=="__main__":
send_message()
7、优化封装短信发送模块
- 每发送一次短信,就会调用发送短信发送模块时,都会实例化SmsSDK类,对资源消耗较大,应采用单例模式
7.1 定义发送短信的单例类
- 定义CCP类,重写__new__()方法
- 首先判断是否存在一个属性(_instance,自定义的)
- 不存在,调用父类super()的__new__()方法,给属性赋值,并将相关参数传递进去,然后再设置该属性添加一个sdk,赋值为发送短信类实例
- 最后返回该属性
class CCP(object):
"""发送短信的单例类"""
def __new__(cls,*args,**kwargs):
if not hasattr(cls,"_instance"):
cls._instance=super().__new__(cls,*args,**kwargs)
cls._instance.sdk=SmsSDK(accId, accToken, appId)
return cls._instance
7.2 优化发送短信模块
- 用单例化短信发送类优化
- 发送验证码参数化
from ronglian_sms_sdk import SmsSDK
import json
accId = '容联云通讯分配的主账号ID'
accToken = '容联云通讯分配的主账号TOKEN'
appId = '容联云通讯分配的应用ID'
class CCP(object):
"""发送短信的单例类"""
def __new__(cls, *args, **kwargs):
# 如果是第一次实例化,应该返回实例化后的对象,如果是第二次实例化,应该返回上一次实例化后的对象
# 判断是否存在类属性 _instance
if not hasattr(cls, "_instance"):
cls._instance = super().__new__(cls, *args, **kwargs)
cls._instance.sdk = SmsSDK(accId, accToken, appId)
return cls._instance
def send_message(self, tid, mobile, datas):
sdk = self._instance.sdk
# tid = '容联云通讯创建的模板ID'
# mobile = '手机号1,手机号2'
# datas = ('变量1', '变量2')
# tid = '1'
# mobile = '18908656327'
# datas = ('2345', '5')
resp = sdk.sendMessage(tid, mobile, datas)
result = json.loads(resp)
if result["statusCode"] == "000000":
return 0
else:
return -1
if __name__ == "__main__":
c = CCP()
c.send_message("1", "18908656327", ("1234", "5")) # 测试1号模板,手机号,(短信验证码,有效时间)
二、短信接口设计
1、短信验证码逻辑分析
- 首先图片验证码正确后,才向后端发起短信验证码ajax请求
- 后端接收到请求后,首先接收参数、验证参数
- 验证参数:
- 从redis中提取图形验证码,提取后,删除redis中的图形验证码,确保图形验证码只使用一次,防止撞库
- 比对传递的图形验证码是否与redis取出的验理证码相同,如果不同,直接返回错误
- 如果相同,生成短信验证码,并保存到redis中
- 调用短信发送模块,发送短信
- 向前端返回发送状态
2、接口文档:
- 接口名字
- 描述
- URL
- 请求方式
- 传入参数
- 返回值
接口:获取短信验证码
描述:前端访问,可以获取到短信验证码编号
URL: /api/v1.0/image_codes/<re(r’1[345678]\d{9}’)>
请求方式:GET
传入参数:
名字 类型 是否必须 说明 image_code 字符串 是 图片验证码的编号 image_code_id 字符串 是 uuid 返回值:
名字 类型 是否必须 说明 errno 字符串 否 错误代码 errmsg 字符串 否 错误内容
3、后端接口定义(路由定义):
- 路由参数规则采用自定义正则表达式: <re(“正则表达式”)>
- 获取参数,确保两参数一定存在: all([列表]) :列表的每个元素都不为空才返回True,否则返回False
- 从redis中取出图片验证码,如果异常,返回出错
- 判断图片验证码是否过期:取出的值为None,没获取到,没有保存或过期
- 从redis中删除图片验证码:防止撞库,一个图片验证码只能使用一次,无论对错
- 判断图片验证是否正确:从redis中取出是二进制数据,需要编码后,再与传过的参数比较,验证码不区分大小写,故需要把字符串换成同大写或同小写比较
- 判断手机号是否注册过:防止手机号注册过的手机再注册,查看数据库里是有手机号
- 生成短信验证码:"%06d" % random.randint(0, 999999):随机生成0-999999的一个数,不够6位,前面补0
- 保存短信验证码到redis中,并设置过期时间(设置为常量)
- 发送短信
@api.route("/sms_codes/<re(r'1[345678]\d{9}'):mobile_code>")
def get_sms_code(mobile_code):
"""获取短信验证码"""
print("mobile_code=",mobile_code)
# 获取参数
# 图片验证码
image_code = request.args.get('image_code')
# UUID
image_code_id = request.args.get('image_code_id')
print("image_code=",image_code)
print("image_code_id=",image_code_id)
# 校验参数,两个参数都不能为空
if not all([image_code, image_code_id]):
return jsonify(errno=RET.PARAMERR, errmsg='参数不完整')
# 业务逻辑
# 从redis中取出验证码
try:
real_redis_code=redis_store.get('image_code_%s' % image_code_id)
except Exception as e:
logging.error(e)
return jsonify(errno=RET.DBERR, errmsg='redis数据库异常')
# 判断图片验证码是否过期
if real_redis_code is None:
return jsonify(errno=RET.NODATA, errmsg='图片验证码失效')
# 删除redis中的图片验证码
try:
redis_store.delete('image_code_%s' % image_code_id)
except Exception as e:
logging.error(e)
# print(real_image_code) b'RVMJ'
# 与用户填写的图片验证码对比
real_redis_code=real_redis_code.decode()
if image_code.lower()!=real_redis_code.lower():
return jsonify(errno=RET.PARAMERR, errmsg='图片验证码错误')
logging.info("real_redis_code="+real_redis_code)
# 判断手机号是否存在
try:
user = User.query.filter_by(mobile=mobile_code).first()
except Exception as e:
logging.error(e)
else:
if user is not None:
# 表示手机号已经被注册过
return jsonify(errno=RET.DATAEXIST, errmsg='手机号已经存在')
# 生成短信验证码
sms_code = "%06d" % random.randint(0, 999999)
# 保存真实的短信验证码到redis
try:
redis_store.setex("sms_code_%s" % mobile_code, constants.SMS_CODE_REDIS_EXPIRES, sms_code)
except Exception as e:
logging.error(e)
return jsonify(errno=RET.DBERR, errmsg='保存短信验证码异常')
# 发短信
try:
ccp = CCP()
result = ccp.send_message(1,mobile_code, (sms_code, int(constants.SMS_CODE_REDIS_EXPIRES/60)))
except Exception as e:
logging.error(e)
return jsonify(errno=RET.THIRDERR, errmsg='发送异常')
# 返回值
if result == 0:
return jsonify(errno=RET.OK, errmsg='发送成功')
else:
return jsonify(errno=RET.THIRDERR, errmsg='发送失败')
4、优化:避免频繁发送短信验证码逻辑实现
- 防止恶意频繁提交发送短信验证的请求,应该限制,每分钟申请1 次
4.1 逻辑实现分析
- 短信验证码发送成功后,将短信验证码发送成功标志保存到redis中,并设置有效时间
- 收到用户发送短信验证码申请后,先判断图形验证吗是否正确,正确后再判断是否在限制时间内再次请求(在redis查询中查询发送状态,如果存在,表示在限制时间内再次发送请求)
- 然后再去验证手机号是否存在,可防止恶意测试注册用户信息
4.2 代码优化完善
- 提取、校验send_flag
# 判断手机号的操作
try:
send_flag=redis_store.get("send_flag_%s"% mobile_code)
except Exception as e:
logging(e)
else:
if send_flag is not None:
return jsonify(errno=RET.RET.REQERR, errmsg='请求过于频繁')
- 重新写入send_flag
# 保存真实的短信验证码到redis
try:
redis_store.setex("sms_code_%s" % mobile_code, constants.SMS_CODE_REDIS_EXPIRES, sms_code)
# 保存发送给这个手机号的记录
redis_store.setex("send_flag_%s"% mobile_code, constants.SEND_SMS_CODE_EXPIRES, 1)
except Exception as e:
logging.error(e)
return jsonify(errno=RET.DBERR, errmsg='保存短信验证码异常')
5、优化:redis管道
- 上面代码中,向redis中保存短信验证码和发送状态,同一时间执行两次redis操作,可以用管道进行优化
- redis管道 详见 《redis管道》
# 保存真实的短信验证码到redis
try:
# redis管道
pl = redis_store.pipeline()
pl.setex("sms_code_%s" % mobile_code, constants.SMS_CODE_REDIS_EXPIRES, sms_code)
# 保存发送给这个手机号的记录
pl.setex('send_sms_code_%s' % mobile_code, constants.SNED_SMS_CODE_EXPIRES, 1)
pl.execute()
except Exception as e:
logging.error(e)
return jsonify(errno=RET.DBERR, errmsg='保存短信验证码异常')
6、在蓝图中引用
- 将此功能与图片验证码放到一个文件中即可
from flask import Blueprint
api = Blueprint("api_1_0", __name__, url_prefix="/api/v1.0")
from . import demo,verify_code