luffcc项目-06-使用容联云发送短信、Celery
一、使用容联云发送短信
https://www.yuntongxun.com/
目录结构
sms.py
import json
from .SmsSDK import SmsSDK
from django.conf import settingsaccId = settings.SMS_INFO.get('ACCID')
accToken = settings.SMS_INFO.get('ACCTOKEN')
appId = settings.SMS_INFO.get('APPID')def send_message(tid, mobile, datas):sdk = SmsSDK(accId, accToken, appId)""" tid = '1'mobile = '18899241027'datas = ('1234', '3')"""resp = sdk.sendMessage(tid, mobile, datas)resp = json.loads(resp)print(resp)return resp.get('statusCode') == '000000'# send_message()
dev.py
SMS_INFO = {'ACCID': '8aaf0708754a3ef2017563ddb22d0773','ACCTOKEN': '0b41612bc8a8429d84b5d37f29178743','APPID': '8aaf0708754a3ef2017563ddb3110779','TID': 1,
}
users/views.py
import random
import refrom django.shortcuts import render
from rest_framework.views import APIView
from rest_framework.generics import CreateAPIView
from rest_framework.response import Response
# Create your views here.
from rest_framework_jwt.views import ObtainJSONWebToken
from rest_framework import status
from users.serializers import CustomeSerializer, RegisterModelSerializer
from .utils import get_user_obj
from . import models
from lyapi.settings import containsfrom lyapi.libs.ronglian_sms_sdk.sms import send_message
from django.conf import settingsclass CustomLoginView(ObtainJSONWebToken):serializer_class = CustomeSerializer#
class CheckPhoneNumber(APIView):def get(self,request):phone_number = request.GET.get('phone')if not re.match('^1[3-9][0-9]{9}$', phone_number):# 格式不对return Response({'error_msg':'手机号格式有误,请重新输入!'}, status=status.HTTP_400_BAD_REQUEST)# 验证唯一性ret = get_user_obj(phone_number)if ret:return Response({'error_msg': '手机号已被注册,请换手机号'}, status=status.HTTP_400_BAD_REQUEST)return Response({'msg': 'ok'})class RegisterView(CreateAPIView):queryset = models.User.objects.all()serializer_class = RegisterModelSerializerimport logging
logger = logging.getLogger('django')
from django_redis import get_redis_connectionclass GetSMSCodeView(APIView):def get(self,request,phone):# 验证是否已经发送过短信了conn = get_redis_connection('sms_code')ret = conn.get('mobile_interval_%s'%phone)if ret:return Response({'msg':'60秒内已经发送过了,别瞎搞'}, status=status.HTTP_400_BAD_REQUEST)# 生成验证码sms_code = "%06d" % random.randint(0,999999)# 保存验证码conn.setex('mobile_%s'%phone, contains.SMS_CODE_EXPIRE_TIME, sms_code) # 设置有效期conn.setex('mobile_interval_%s'%phone, contains.SMS_CODE_INTERVAL_TIME, sms_code) # 设置发送短信的时间间隔# 发送验证码ret = send_message(settings.SMS_INFO.get('TID'), phone, (sms_code, contains.SMS_CODE_EXPIRE_TIME // 60))if not ret:logger.error('{}手机号短信发送失败'.format(phone))return Response({'msg': '短信发送失败,请联系管理员'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)return Response({'msg':'ok'})
Register.vue
<template><div class="box"><img src="../../static/img/Loginbg.3377d0c.jpg" alt=""><div class="register"><div class="login-title"><img src="../../static/img/Logotitle.1ba5466.png" alt=""><p>帮助有志向的年轻人通过努力学习获得体面的工作和生活!</p></div><div class="register_box"><div class="register-title">注册路飞学城</div><div class="inp"><input v-model = "mobile" type="text" placeholder="手机号码" class="user" @blur="checkPhone"><input v-model = "password" type="password" placeholder="密码" class="user"><input v-model = "r_password" type="password" placeholder="确认密码" class="user"><div><input v-model = "sms" type="text" placeholder="输入验证码" class="user" style="width: 62%"><button style="width: 34%;height: 41px;" @click="getSmsCode">{{btn_msg}}</button></div><button class="register_btn" @click="registerHandler">注册</button><p class="go_login" >已有账号 <router-link to="/user/login">直接登录</router-link></p></div></div></div></div>
</template><script>
export default {name: 'Register',data(){return {sms:"",mobile:"",password:"",r_password:"",validateResult:false,interval_time: 60,btn_msg: '点击获取验证码',flag: false,}},created(){},methods:{checkPhone(){let phoneNumber = this.mobile;// 前端校验格式let reg = /^1[3-9][0-9]{9}$/;if (!reg.test(phoneNumber)){this.$message.error("手机号格式不对");return false;}// 发送请求this.$axios.get(`${this.settings.Host}/users/check_phone/?phone=${phoneNumber}`).then((res)=>{console.log(res);}).catch((error)=>{this.$message.error(error.response.data.error_msg);})},registerHandler(){this.$axios.post(`${this.$settings.Host}/users/register/`, {sms: this.sms,phone: this.mobile,password: this.password,r_password: this.r_password,}).then((res)=>{sessionStorage.token = res.data.token;sessionStorage.username = res.data.username;sessionStorage.id = res.data.id;this.$router.push('/');}).catch((error)=>{console.log(error.response);})},// 点击获取验证码getSmsCode(){this.$axios.get(`${this.$settings.Host}/users/sms_code/${this.mobile}/`).then((res)=>{if (!this.flag){this.flag = setInterval(()=>{if (this.interval_time > 0){this.interval_time--;this.btn_msg = `${this.interval_time}秒后重新获取`;}else {this.interval_time = 60;this.btn_msg = '点击获取验证码'clearInterval(this.flag);this.flag = false;}}, 1000)}}).catch((error)=>{this.$message.error(error.response.data.msg);})}},};
</script>...
二、Celery
Celery是一个功能完备即插即用的异步任务队列系统。它适用于异步处理问题,当发送邮件、或者文件上传, 图像处理等等一些比较耗时的操作,可将其异步执行,这样用户不需要等待很久,提高用户体验。
文档:http://docs.jinkan.org/docs/celery/getting-started/index.html
1.Celery的特点是:
- 简单,易于使用和维护,有丰富的文档。
- 高效,单个celery进程每分钟可以处理数百万个任务。
- 灵活,celery中几乎每个部分都可以自定义扩展。
2.Celery的架构
Celery的架构由三部分组成,消息队列(message broker),任务执行单元(worker)和任务执行结果存储(task result store)组成。
一个celery系统可以包含很多的worker和brokerCelery本身不提供消息队列功能,但是可以很方便地和第三方提供的消息中间件进行集成,包括RabbitMQ,Redis,MongoDB等
3.安装
pip install -U celery #-U是update的意思,有就进行更新,没有就安装
#后面单独将celery运行起来就可以了
也可从官方直接下载安装包:https://pypi.python.org/pypi/celery/
tar xvfz celery-0.0.0.tar.gz
cd celery-0.0.0
python setup.py
python setup.py install
4.使用
使用celery第一件要做的最为重要的事情是需要先创建一个Celery实例,我们一般叫做celery应用,或者更简单直接叫做一个app。app应用是我们使用celery所有功能的入口,比如创建任务,管理任务等,在使用celery的时候,app必须能够被其他的模块导入。
一般celery任务目录直接放在项目的根目录下即可,路径:
lyapi/
├── mycelery/├── config.py # 配置文件├── __init__.py ├── main.py # 主程序└── sms/ # 一个目录可以放置多个任务,该目录下存放当前任务执行时需要的模块或依赖,也可以每个任务单独一个目录└── tasks.py # 任务的文件,名称必须是这个!!!
main.py,代码:
from celery import Celeryimport os
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'lyapi.settings.dev')
import django
django.setup()app = Celery()app.config_from_object('mycelery.config')app.autodiscover_tasks(['mycelery.sms'])
配置文件config.py代码:
# 任务队列的链接地址(变量名必须叫这个)
broker_url = 'redis://127.0.0.1:6379/14'
# 结果队列的链接地址(变量名必须叫这个)
result_backend = 'redis://127.0.0.1:6379/15'
创建一个任务文件sms/tasks.py,并创建任务,代码:
from mycelery.main import appfrom lyapi.libs.ronglian_sms_sdk.sms import send_message
from django.conf import settings
from lyapi.settings import containsimport logging
logger = logging.getLogger('django')@app.task(name='smsCode')
def sms_codes(phone, sms_code):# 发送验证码ret = send_message(settings.SMS_INFO.get('TID'), phone, (sms_code, contains.SMS_CODE_EXPIRE_TIME // 60))if not ret:logger.error('{}手机号短信发送失败'.format(phone))return '短信发送成功'@app.task()
def sms_code2():print('xxx2')return '发送短信成功2'
接下来,我们运行celery,在终端,项目根目录下(也就是mycelery的外层目录里面)执行指令
celery -A mycelery.main worker --loglevel=info (或者直接写info也行) #-A是指定celery启动入口
运行起来之后,如果又添加了新的任务,需要重新启动celery。
SmsSDK.py
#####################################################################################################
#
# Copyright (c) 2014 The CCP project authors. All Rights Reserved.
#
# Use of this source code is governed by a Beijing Speedtong Information Technology Co.,Ltd license
# that can be found in the LICENSE file in the root of the web site.
#
# https://www.yuntongxun.com
#
# An additional intellectual property rights grant can be found
# in the file PATENTS. All contributing project authors may
# be found in the AUTHORS file in the root of the source tree.from . import algorithm
import requests
import time
import json
import tracebackclass SmsSDK:"""短信SDK"""# 容联云通讯服务地址url = 'https://app.cloopen.com:8883'# 发送短信URIsendMessageURI = '/2013-12-26/Accounts/{}/SMS/TemplateSMS'def __init__(self, accId, accToken, appId):self.__accId = accIdself.__accToken = accTokenself.__appId = appIddef sendMessage(self, tid: str, mobile: str, datas: tuple) -> str:"""发送短信Args:tid: 短信模板ID,容联云通讯网站自行创建mobile: 下发手机号码,多个号码以英文逗号分隔datas: 模板变量Returns:返回发送结果和发送成功消息ID发送成功示例:{"statusCode":"000000","templateSMS":{"dateCreated":"20130201155306","smsMessageSid":"ff8080813c373cab013c94b0f0512345"}}发送失败示例:{"statusCode": "172001", "statusMsg": "网络错误"}"""timestamp = time.strftime('%Y%m%d%H%M%S', time.localtime())url = self.__buildSendMessageUrl(timestamp)headers = self.__buildHeaders(timestamp)body = self.__buildSendMessageBody(tid, mobile, datas)self.__logRequestInfo(url, headers, body)try:r = requests.post(url, headers=headers, data=body, timeout=(2, 5))if (r.status_code == requests.codes.ok):print('Response body: ', r.text)return r.textelse:return json.dumps({'statusCode': str(r.status_code)})except:traceback.print_exc()return '{"statusCode": "172001", "statusMsg": "网络错误"}'def __buildSendMessageUrl(self, timestamp):"""构建发送短信URL"""return f'{self.url}{self.sendMessageURI.format(self.__accId)}?sig={self.__buildSign(timestamp)}'def __buildSign(self, timestamp):"""构建签名sigArgs:timestamp: 时间字符串 格式:yyyyMMddHHmmssReturns:签名大写字符串"""plaintext = f'{self.__accId}{self.__accToken}{timestamp}'print("Sign plaintext: ", plaintext)return algorithm.md5(plaintext).upper()def __buildHeaders(self, timestamp):"""构建请求报头"""headers = {}headers['Content-Type'] = 'application/json;charset=utf-8'headers['Accept'] = 'application/json'headers['Accept-Charset'] = 'UTF-8'headers['Authorization'] = self.__buildAuthorization(timestamp)return headersdef __buildAuthorization(self, timestamp):"""构建报头AuthorizationArgs:timestamp: 时间字符串 格式:yyyyMMddHHmmssReturns:Authorization字符串"""plaintext = f'{self.__accId}:{timestamp}'print("Authorization plaintext: %s" % plaintext)return algorithm.base64Encoder(plaintext)def __buildSendMessageBody(self, tid, mobile, datas):"""构建发送短信报文"""body = {}body['to'] = mobilebody['appId'] = self.__appIdbody['templateId'] = tidbody['datas'] = datasreturn json.dumps(body)def __logRequestInfo(self, url, headers, body):"""打印请求信息日志"""print('Request url: ', url)print('Request headers: ', headers)print('Request body: ', body)
users/serializers.py
import refrom rest_framework_jwt.serializers import JSONWebTokenSerializer
from rest_framework import serializers
from rest_framework_jwt.compat import get_username_field, PasswordField
from django.utils.translation import ugettext as _
from django.contrib.auth import authenticate, get_user_model
from rest_framework_jwt.settings import api_settings
from . import models
from django.contrib.auth.hashers import make_password
from django_redis import get_redis_connectionfrom .utils import get_user_objUser = get_user_model()
jwt_payload_handler = api_settings.JWT_PAYLOAD_HANDLER
jwt_encode_handler = api_settings.JWT_ENCODE_HANDLER
jwt_decode_handler = api_settings.JWT_DECODE_HANDLER
jwt_get_username_from_payload = api_settings.JWT_PAYLOAD_GET_USERNAME_HANDLERclass CustomeSerializer(JSONWebTokenSerializer):def __init__(self, *args, **kwargs):"""Dynamically add the USERNAME_FIELD to self.fields."""super(JSONWebTokenSerializer, self).__init__(*args, **kwargs)self.fields[self.username_field] = serializers.CharField()self.fields['password'] = PasswordField(write_only=True)self.fields['ticket'] = serializers.CharField(write_only=True)self.fields['randstr'] = serializers.CharField(write_only=True)#def validate(self, attrs):credentials = {self.username_field: attrs.get(self.username_field),'password': attrs.get('password'),'ticket': attrs.get('ticket'),'randstr': attrs.get('randstr'),}# {'username':'root',password:'123'}if all(credentials.values()):user = authenticate(self.context['request'], **credentials) # self.context['request']当前请求的request对象if user:if not user.is_active:msg = _('User account is disabled.')raise serializers.ValidationError(msg)payload = jwt_payload_handler(user)return {'token': jwt_encode_handler(payload),'user': user}else:msg = _('Unable to log in with provided credentials.')raise serializers.ValidationError(msg)else:msg = _('Must include "{username_field}" and "password".')msg = msg.format(username_field=self.username_field)raise serializers.ValidationError(msg)class RegisterModelSerializer(serializers.ModelSerializer):id = serializers.IntegerField(read_only=True)sms = serializers.CharField(max_length=6, min_length=4, write_only=True) # '3333'r_password = serializers.CharField(write_only=True)token = serializers.CharField(read_only=True) #class Meta:model = models.Userfields = ['id', 'phone', 'password', 'r_password', 'sms', 'token']extra_kwargs = {'password': {'write_only': True},}# 校验密码和确认密码def validate(self, attrs):# 校验手机号phone_number = attrs.get('phone')sms = attrs.get('sms')if not re.match('^1[3-9][0-9]{9}$', phone_number):raise serializers.ValidationError('手机号格式不对')ret = get_user_obj(phone_number)if ret:raise serializers.ValidationError('has one!!!')p1 = attrs.get('password')p2 = attrs.get('r_password')if p1 != p2:raise serializers.ValidationError('两次密码不一致,请核对')# 校验验证码conn = get_redis_connection('sms_code')ret = conn.get('mobile_%s'%phone_number)if not ret:raise serializers.ValidationError('验证码已失效')if ret.decode() != sms:raise serializers.ValidationError('验证码错误')return attrsdef create(self, validated_data):validated_data.pop('r_password')validated_data.pop('sms')# 密码加密hash_password = make_password(validated_data['password'])validated_data['password'] = hash_passwordvalidated_data['username'] = validated_data.get('phone')user = models.User.objects.create(**validated_data)jwt_payload_handler = api_settings.JWT_PAYLOAD_HANDLERjwt_encode_handler = api_settings.JWT_ENCODE_HANDLERpayload = jwt_payload_handler(user)token = jwt_encode_handler(payload)user.token = tokenreturn user
users/views.py
import random
import refrom django.shortcuts import render
from rest_framework.views import APIView
from rest_framework.generics import CreateAPIView
from rest_framework.response import Response
# Create your views here.
from rest_framework_jwt.views import ObtainJSONWebToken
from rest_framework import status
from users.serializers import CustomeSerializer, RegisterModelSerializer
from .utils import get_user_obj
from . import models
from lyapi.settings import containsfrom lyapi.libs.ronglian_sms_sdk.sms import send_message
from django.conf import settingsclass CustomLoginView(ObtainJSONWebToken):serializer_class = CustomeSerializer#
class CheckPhoneNumber(APIView):def get(self,request):phone_number = request.GET.get('phone')if not re.match('^1[3-9][0-9]{9}$', phone_number):# 格式不对return Response({'error_msg':'手机号格式有误,请重新输入!'}, status=status.HTTP_400_BAD_REQUEST)# 验证唯一性ret = get_user_obj(phone_number)if ret:return Response({'error_msg': '手机号已被注册,请换手机号'}, status=status.HTTP_400_BAD_REQUEST)return Response({'msg': 'ok'})class RegisterView(CreateAPIView):queryset = models.User.objects.all()serializer_class = RegisterModelSerializerimport logging
logger = logging.getLogger('django')
from django_redis import get_redis_connectionclass GetSMSCodeView(APIView):def get(self,request,phone):# 验证是否已经发送过短信了conn = get_redis_connection('sms_code')ret = conn.get('mobile_interval_%s'%phone)if ret:return Response({'msg':'60秒内已经发送过了,别瞎搞'}, status=status.HTTP_400_BAD_REQUEST)# 生成验证码sms_code = "%06d" % random.randint(0,999999)# 保存验证码conn.setex('mobile_%s'%phone, contains.SMS_CODE_EXPIRE_TIME, sms_code) # 设置有效期conn.setex('mobile_interval_%s'%phone, contains.SMS_CODE_INTERVAL_TIME, sms_code) # 设置发送短信的时间间隔# 发送验证码# ret = send_message(settings.SMS_INFO.get('TID'), phone, (sms_code, contains.SMS_CODE_EXPIRE_TIME // 60))# if not ret:# logger.error('{}手机号短信发送失败'.format(phone))# return Response({'msg': '短信发送失败,请联系管理员'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)from mycelery.sms.tasks import sms_codessms_codes.delay(phone, sms_code)return Response({'msg': 'ok'})
users/urls.py
from rest_framework_jwt.views import obtain_jwt_token, verify_jwt_token
from . import views
from django.urls import path,re_pathurlpatterns = [path(r'login/', views.CustomLoginView.as_view()), #颁发token值的path(r'verify/', verify_jwt_token),path(r'check_phone/', views.CheckPhoneNumber.as_view()),path(r'register/', views.RegisterView.as_view()),re_path(r'sms_code/(?P<phone>1[3-9][0-9]{9})/', views.GetSMSCodeView.as_view()),]
luffcc项目-06-使用容联云发送短信、Celery相关推荐
- Go语言初识应用--容联云发送短信验证码、手机号注册
使用gin框架.gorm映射 所使用的连接容联云参考容联云官方文档,放置到utils中, gin项目结构根据自身需要,大题如下设置: utils--sms.go package main import ...
- 容联云发送短信验证码
setting配置 # 配置jwt的全局认证 REST_FRAMEWORK = {'DEFAULT_AUTHENTICATION_CLASSES':['rest_framework_jwt.authe ...
- Django使用容联云发送短信验证码时提示:172001,网络错误
尝试用django连接容联云发送短信验证码,运行时出现了以下提示:{'172001':'网络错误'}:在网上查阅资料后得知,是因为python升级到2.7.9之后引入了一个新特性,当打开一个https ...
- flask容联云发送短信验证码和异步发送
容联云地址:https://www.yuntongxun.com/ 获取短信验证码流程图 容联云配置 在容联云官网注册一个账号,发送短信验证码必须使用三个ID ACCOUNT SID (主账户ID ...
- python 容联云测试短信接口
文章目录 1.注册容连云 1.1 登录即可看到开发者账号信息 1.2 添加测试账号 2.使用容联云发送代码测 3.在视图函数中使用 3.1写视图函数 3.1 添加路由 1.注册容连云 注册地址 1.1 ...
- Luffy之注册认证(容联云通讯短信验证)
用户的注册认证 前端显示注册页面并调整首页头部和登陆页面的注册按钮的链接. 注册页面Register,主要是通过登录页面进行改成而成. 先构造前端页面 <template><div ...
- 使用容联云获取短信验证码
容联云地址:https://www.yuntongxun.com/ 获取短信验证码流程图 容联云配置 在容联云官网注册一个账号,发送短信验证码必须使用三个ID ACCOUNT SID (主账户ID ...
- django_容联云_短信验证
1017短信验证 02_requests.py """ # @TIME: 2021/10/16 下午10:08 # @FILE: 02_requests.py # @AU ...
- 容联云通讯短信平台JS调用
容联云通讯官网 1.注册 注册成功之后,会赠送8元进行测试 记录账号信息用于代码中配置 添加测试号码用于测试 2.接口配置(js) var md5 = require('blueimp-md5') v ...
最新文章
- truncate table
- Kerberos认证代码分析Can't get Kerberos realm
- .NET 云原生架构师训练营(模块二 基础巩固 配置)--学习笔记
- python源码笔记_python源码学习笔记(二)
- OpenCV与图像处理学习十四——SIFT特征(含代码)
- Win7上Git安装及配置过程 [转]
- Python练习题(四)
- DevComponents.AdvTree 相关笔记
- SQL Cookbook(读书笔记)No.2
- svm 10折交叉验证 matlab,怎么用10折交叉验证程序?
- 声音分贝测试软件源码,C# 电脑录音 实例源码(根据声音分贝大小自动录制)
- Longest Continuous 1
- 應電鍍廠要求把5個ITEM的主單位PRIMARY UOM由L改為KG
- Axure8与Axure9交互差异总结-1 移动元件交互事件的差异
- 用上就不会停下的效率利器—Automator
- 【ResNet残差网络解析】
- c语言程序设计 葛日波,C语言程序设计课程教学改革的探索
- 关于计算机网络实训室的申请书,计算机网络技术综合实训室建设方案.doc
- 2012年10月20日
- 【代码】QQ群最近聊天记录做成词云