pip3 install tencentcloud-sdk-python -i https://pypi.douban.com/simple
url
# 发送短信
path('tencent_sms/', TencentSmsCodeAPIView.as_view()),
view(如果不使用celery的话,使用普通短信发送函数调用接口)
import random
from rest_framework import status
from rest_framework.views import APIView
from rest_framework.response import Response
from utils.common.tencent_sms import TencentSmsCode
class TencentSmsCodeAPIView(APIView):
"""腾讯短信发送(基于v3版本)"""
authentication_classes = []
def get(self, request, *args, **kwargs):
telephone = request.GET.get('phone')
tpl = request.GET.get('tpl')
code = str(random.randrange(1000, 9999))
# print(telephone, tpl, code)
t_sms = TencentSmsCode(telephone, [code], tpl)
# 普通短信发送函数调用
# ret = t_sms.sms_check()
# 异步短信发送函数调用
ret = t_sms.sms_check_celery()
if not ret:
# print(ret)
return Response(data='ok', status=status.HTTP_200_OK)
data = {
'status': ret.get("status"),
'msg': ret.get("msg"),
}
# print(ret)
return Response(data=data, status=status.HTTP_400_BAD_REQUEST)
TencentSmsCode
import re
import json
from tencentcloud.common import credential
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
from tencentcloud.sms.v20210111 import sms_client, models
from django.conf import settings
from django_redis import get_redis_connection
# pip3 install tencentcloud-sdk-python
class TencentSmsCode(object):
"""
phone str
code [str,str]
tpl_id str
"""
def __init__(self, telephone, code, tpl):
self.telephone = telephone
self.code = code
self.tpl = tpl
self.tpl_id = ''
def __data(self, status, msg):
"""返回格式封装"""
data = {
"status": status,
"msg": msg
}
return data
def __check(self):
"""进行校验是否合法"""
ret_telephone = self.__check_telephone()
if not ret_telephone:
data = self.__data(1001, '手机号格式不正确')
return data
# 进行校验是否有tpl
ret_tpl = self.__check_tpl()
if not ret_tpl:
data = self.__data(1002, '模板不正确')
return data
# 验证码的生成 redis进行查询是否过期 redis的存储
ret_code = self.__set_code()
if not ret_code:
data = self.__data(1003, '60秒只能发送1次')
return data
def sms_check(self):
"""
正常发送
"""
ret = self.__check()
self.__send_msg()
return ret
def sms_check_celery(self):
"""
celery进行异步发送
:return:
"""
ret = self.__check()
from auc_celery.sms.tasks import send_sms
# 执行任务
tid = send_sms.delay(self.telephone, self.code, self.tpl)
return ret
def __send_msg(self):
try:
phone = '+86%s' % self.telephone
cred = credential.Credential(settings.TENCENT_SECRET_ID, settings.TENCENT_SECRET_KEY)
httpProfile = HttpProfile()
httpProfile.endpoint = "sms.tencentcloudapi.com"
clientProfile = ClientProfile()
clientProfile.httpProfile = httpProfile
client = sms_client.SmsClient(cred, "ap-guangzhou", clientProfile)
req = models.SendSmsRequest()
params = {
"PhoneNumberSet": [phone, ], # 手机号
"SmsSdkAppId": settings.TENCENT_APP_ID, # 应用AppId,v2要应用sdk/key,v3新增了密钥,取消了key
"SignName": settings.TENCENT_SIGN, # 签名的模版名称
"TemplateId": self.tpl_id, # 短信模板的id,str格式
"TemplateParamSet": self.code # 格式必须是 ["验证码",参数] 参数可以为空,具体看自己的模板要求
}
req.from_json_string(json.dumps(params))
resp = client.SendSms(req)
# print(resp.to_json_string())
if resp.SendStatusSet[0].Fee == 1:
return True
return
except TencentCloudSDKException as err:
print('err', err)
def __check_telephone(self):
"""
手机号校验
:param phone: 手机号
:return: True/None
"""
ret = re.match('^1[3-9][0-9]{9}$', self.telephone)
if ret:
return True
return
def __check_tpl(self):
tpl = self.tpl
# print('tpl:', tpl)
tpl_id = settings.TENCENT_SMS_TEMPLATES.get(tpl)
if tpl_id:
# print('tpl_id', tpl_id)
self.tpl_id = tpl_id
return True
return
def __set_code(self):
"""
生成验证码存放redis
:param phone: 手机号
:return:
"""
# 生成code
code = self.code[0]
print(code)
# 保存redis
conn = get_redis_connection('sms_code')
pipe = conn.pipeline() # 创建管道
pipe.multi() # 开启批量 *** 作 (事务,)
# # 后端倒计时校验 查看redis是否有,有抛异常
init_code = conn.get('init_code_%s' % self.telephone)
if init_code:
return
# 保存正常验证码
pipe.set('sms_code_%s' % self.telephone, code, ex=60 * 60 * 24 * 14)
# 保存倒计时码
pipe.set('init_code_%s' % self.telephone, code, ex=60)
pipe.execute() # 批量执行了
return code
settings.py
# redis
# 上面是django项目settings中的其他配置....
CACHES = {
"default": {
"BACKEND": "django_redis.cache.RedisCache",
"LOCATION": "redis://106.14.42.253:3306", # 安装redis的主机的 IP 和 端口
"OPTIONS": {
"CLIENT_CLASS": "django_redis.client.DefaultClient",
"CONNECTION_POOL_KWARGS": {
"max_connections": 1000,
"encoding": 'utf-8'
},
"PASSWORD": "foobared" # redis密码
}
},
# 提供存储短信验证码
"sms_code": {
"BACKEND": "django_redis.cache.RedisCache",
"LOCATION": "redis://106.14.42.253:3306/2",
"OPTIONS": {
"CLIENT_CLASS": "django_redis.client.DefaultClient",
"CONNECTION_POOL_KWARGS": {
"max_connections": 1000,
"encoding": 'utf-8'
},
"PASSWORD": "foobared" # redis密码
}
},
}
# ############################# 腾讯云短信配置 ##########################
TENCENT_SECRET_ID = "AKIDbz7yZyxxxxxxxUO1MQDjzxf5E6h1" # 密钥id
TENCENT_SECRET_KEY = "4rMbZGgxxxxxxxz5ukJbcMp1vYv0o" # 密钥key
TENCENT_CITY = "ap-guangzhou" # 地域,默认即可,
TENCENT_APP_ID = "14xxxxx1766" # 应用id,v2要应用id/key,v3新增了密钥
TENCENT_SIGN = "python后端接口开发" # 签名的模版名称
# 短信模板
TENCENT_SMS_TEMPLATES = {
# 单个参数
'register': "993292",
'login': "993155"
}
异步调用配置(需要安装django-redis)
目录结构
auc_celery(文件夹,一般放在根目录下)
sms(包)
__init__.py
task.py
__init__.py
config.py
main.py
main.py
from celery import Celery
# 创建celery实例对象
app = Celery("auction") # celery对象可以创建多个,所以我们最好给我们当前的celery应用起个名字,比如叫做dbj
# 把celery和django进行组合,需要识别和加载django的配置文件
import os
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'auction_api.settings')
# 如果只是使用了logging日志功能的话可以不写以下两句,因为logging是python提供的模块,但是将来可能使用celery来执行其他的django任务,所以我们先写上
import django
django.setup()
# 通过app对象加载配置
app.config_from_object("auc_celery.config")
# 加载任务
# 参数必须必须是一个列表,里面的每一个任务都是任务的路径名称
# app.autodiscover_tasks(["任务1","任务2"])
app.autodiscover_tasks(["auc_celery.sms", ])
config.py
# 配置文件
# 1. 有redis有密码
# 任务队列的链接地址(变量名必须叫这个)
broker_url = 'redis://:[email protected]:3306/14'
# 结果队列的链接地址(变量名必须叫这个)
result_backend = 'redis://:[email protected]:3306/15'
# 2. redis无密码写法
# 任务队列的链接地址(变量名必须叫这个)
# broker_url = 'redis://127.0.0.1:6379/14'
# 结果队列的链接地址(变量名必须叫这个)
# result_backend = 'redis://127.0.0.1:6379/15'
task.py
# celery的任务必须写在tasks.py的文件中,别的文件名称不识别!!!
from auc_celery.main import app
from utils.common.tencent_sms import TencentSmsCode
@app.task(name="send_sms") # name表示设置任务的名称,如果不填写,则默认使用函数名(路径)做为任务名
def send_sms(telephone, code, tpl):
t_sms = TencentSmsCode(telephone, code, tpl)
t_sms.sms_check()
return 'ok'
启动celery
celery -A auc_celery.main worker -l info -P eventlet
settings.py
# redis
# 上面是django项目settings中的其他配置....
CACHES = {
"default": {
"BACKEND": "django_redis.cache.RedisCache",
"LOCATION": "redis://106.14.42.253:3306", # 安装redis的主机的 IP 和 端口
"OPTIONS": {
"CLIENT_CLASS": "django_redis.client.DefaultClient",
"CONNECTION_POOL_KWARGS": {
"max_connections": 1000,
"encoding": 'utf-8'
},
"PASSWORD": "foobared" # redis密码
}
},
# 提供存储短信验证码
"sms_code": {
"BACKEND": "django_redis.cache.RedisCache",
"LOCATION": "redis://106.14.42.253:3306/2",
"OPTIONS": {
"CLIENT_CLASS": "django_redis.client.DefaultClient",
"CONNECTION_POOL_KWARGS": {
"max_connections": 1000,
"encoding": 'utf-8'
},
"PASSWORD": "foobared" # redis密码
}
},
}
# ############################# 腾讯云短信配置 ##########################
TENCENT_SECRET_ID = "AKIDbz7yZyxxxxxxxUO1MQDjzxf5E6h1" # 密钥id
TENCENT_SECRET_KEY = "4rMbZGgxxxxxxxz5ukJbcMp1vYv0o" # 密钥key
TENCENT_CITY = "ap-guangzhou" # 地域,默认即可,
TENCENT_APP_ID = "14xxxxx1766" # 应用id,v2要应用id/key,v3新增了密钥
TENCENT_SIGN = "python后端接口开发" # 签名的模版名称
# 短信模板
TENCENT_SMS_TEMPLATES = {
# 单个参数
'register': "993292",
'login': "993155"
}
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)