溫馨提示×
您好，登錄后才能下訂單哦！
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶服務(wù)條款》
您好，登錄后才能下訂單哦！
本篇文章給大家分享的是有關(guān)如何基于Serverless借助微信公眾號(hào)簡(jiǎn)單管理用戶激活碼，小編覺(jué)得挺實(shí)用的，因此分享給大家學(xué)習(xí)，希望大家閱讀完這篇文章后可以有所收獲，話不多說(shuō)，跟著小編一起來(lái)看看吧。
作為一名獨(dú)立開發(fā)者，最近我在考慮給自己的應(yīng)用加入付費(fèi)功能，然后應(yīng)用的核心功能只需使用激活碼付費(fèi)激活即可。這個(gè)需求涉及到了激活碼的保存、校驗(yàn)和后臺(tái)管理，傳統(tǒng)的做法可能是自己購(gòu)買服務(wù)器，搭建配置服務(wù)器環(huán)境，然后創(chuàng)建數(shù)據(jù)庫(kù)，編寫后端業(yè)務(wù)邏輯代碼，必要的時(shí)候還要自己去寫一些前端的界面來(lái)管理后臺(tái)數(shù)據(jù)。
這是一個(gè)十分耗時(shí)且無(wú)趣的工作。
嘗試帶大家使用云函數(shù) SCF 和對(duì)象存儲(chǔ) COS，快速編寫上線自己的用戶激活碼后端管理云函數(shù)，然后把自己的微信公眾號(hào)后臺(tái)做為應(yīng)用前臺(tái)，簡(jiǎn)單管理用戶激活碼。
可以看到，現(xiàn)在我們只需要在自己的微信公眾號(hào)后臺(tái)回復(fù) 會(huì)員@激活時(shí)長(zhǎng)，就可以添加并回復(fù)一個(gè)指定有效期的會(huì)員激活碼，實(shí)現(xiàn)了在微信公眾號(hào)簡(jiǎn)單管理用戶激活碼的需求。
話不多說(shuō)，上代碼
import json
from wechatpy.replies import ArticlesReply
from wechatpy.utils import check_signature
from wechatpy.crypto import WeChatCrypto
from wechatpy import parse_message, create_reply
from wechatpy.exceptions import InvalidSignatureException, InvalidAppIdException
import datetime
import random
# 是否開啟本地debug模式
debug = False
# 騰訊云對(duì)象存儲(chǔ)依賴
if debug:
from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
from qcloud_cos import CosServiceError
from qcloud_cos import CosClientError
else:
from qcloud_cos_v5 import CosConfig
from qcloud_cos_v5 import CosS3Client
from qcloud_cos_v5 import CosServiceError
from qcloud_cos_v5 import CosClientError
# 配置存儲(chǔ)桶
appid = '66666666666'
secret_id = u'xxxxxxxxxxxxxxx'
secret_key = u'xxxxxxxxxxxxxxx'
region = u'ap-chongqing'
bucket = 'name'+'-'+appid
# 微信公眾號(hào)對(duì)接
wecaht_id = 'xxxxxxxxxxxxxxx'
WECHAT_TOKEN = 'xxxxxxxxxxxxxxxxxxx'
encoding_aes_key = 'xxxxxxxxxxxxxxxxxxxxxx'
# 對(duì)象存儲(chǔ)實(shí)例
config = CosConfig(Secret_id=secret_id, Secret_key=secret_key, Region=region)
client = CosS3Client(config)
#微信公眾號(hào)后臺(tái)消息加解密實(shí)例
crypto = WeChatCrypto(WECHAT_TOKEN, encoding_aes_key, wecaht_id)
# cos 文件讀寫
def cosRead(key):
try:
response = client.get_object(Bucket=bucket, Key=key)
txtBytes = response['Body'].get_raw_stream()
return txtBytes.read().decode()
except CosServiceError as e:
return ""
def cosWrite(key, txt):
try:
response = client.put_object(
Bucket=bucket,
Body=txt.encode(encoding="utf-8"),
Key=key,
)
return True
except CosServiceError as e:
return False
#獲取所有會(huì)員激活碼
def getvips():
vipMap = {}
vipTxt = cosRead('vips.txt') # 讀取數(shù)據(jù)
if len(vipTxt) > 0:
vipMap = json.loads(vipTxt)
return vipMap
#添加會(huì)員激活碼
def addvip(days):
vip=randomKey()
vipMap = getvips()
if len(vipMap) > 0:
vipMap[vip] = (datetime.datetime.now()+datetime.timedelta(days=days)).strftime("%Y-%m-%d")
return cosWrite('vips.txt', json.dumps(vipMap, ensure_ascii=False)),vip if len(vipMap) > 0 else False,''
#刪除會(huì)員激活碼
def delvip(vip):
vipMap = getvips()
if len(vipMap) > 0:
vipMap.pop(vip)
return cosWrite('vips.txt', json.dumps(vipMap, ensure_ascii=False)) if len(vipMap) > 0 else False
# 獲取今日日期
def today():
return datetime.datetime.now().strftime("%Y-%m-%d")
# 判斷激活碼是否到期
def checkVip(t):
return t == today()
# 隨機(jī)生成激活碼
def randomKey():
return ''.join(random.sample('zyxwvutsrqponmlkjihgfedcba0123456789', 6))
#每天定時(shí)檢查刪除過(guò)期的激活碼
def check_del_vips():
vipMap = getvips()
if len(vipMap) < 1:
return
for vip in vipMap.keys():
if not checkVip(vipMap[vip]):
vipMap.pop(vip)
return cosWrite('vips.txt', json.dumps(vipMap, ensure_ascii=False))
# api網(wǎng)關(guān)響應(yīng)集成
def apiReply(reply, txt=False, content_type='application/json', code=200):
return {
"isBase64Encoded": False,
"statusCode": code,
"headers": {'Content-Type': content_type},
"body": json.dumps(reply, ensure_ascii=False) if not txt else str(reply)
}
def replyMessage(msg):
txt = msg.content
if '@' in txt:
keys = txt.split('@')
if keys[0] == '會(huì)員': # 會(huì)員@356 --> 添加一個(gè)365天的會(huì)員激活碼
flag,vip=addvip(keys[1])
return create_reply(f"您的激活碼:{vip}，有效期:{keys[1]}天" if flag else "添加失敗", msg)
return create_reply("喵嗚 ?'ω'?", msg)
def wechat(httpMethod, requestParameters, body=''):
if httpMethod == 'GET':
signature = requestParameters['signature']
timestamp = requestParameters['timestamp']
nonce = requestParameters['nonce']
echo_str = requestParameters['echostr']
try:
check_signature(WECHAT_TOKEN, signature, timestamp, nonce)
except InvalidSignatureException:
echo_str = 'error'
return apiReply(echo_str, txt=True, content_type="text/plain")
elif httpMethod == 'POST':
msg_signature = requestParameters['msg_signature']
timestamp = requestParameters['timestamp']
nonce = requestParameters['nonce']
try:
decrypted_xml = crypto.decrypt_message(
body,
msg_signature,
timestamp,
nonce
)
except (InvalidAppIdException, InvalidSignatureException):
return
msg = parse_message(decrypted_xml)
if msg.type == 'text':
reply = replyMessage(msg)
else:
reply = create_reply('哈? ??？\n搞不明白你給我發(fā)了啥~', msg)
reply = reply.render()
reply = crypto.encrypt_message(reply, nonce, timestamp)
return apiReply(reply, txt=True, content_type="application/xml")
else:
msg = parse_message(body)
reply = create_reply("喵嗚 ?'ω'?", msg).render()
reply = crypto.encrypt_message(reply, nonce, timestamp)
return apiReply(reply, txt=True, content_type="application/xml")
def main_handler(event, context):
if 'Time' in event.keys(): # 來(lái)自定時(shí)觸發(fā)器
return check_del_vips()
httpMethod = event["httpMethod"]
requestParameters = event['queryString']
body = event['body'] if 'body' in event.keys() else ''
response = wechat(httpMethod, requestParameters, body=body)
return responseOK, 教程結(jié)束，
哈？你說(shuō)沒(méi)看懂這堆代碼？
好吧，我再耐心給大家捋一下，這次可一定要記住了哈~
def main_handler(event, context): if 'Time' in event.keys(): # 來(lái)自定時(shí)觸發(fā)器 return check_del_vips() httpMethod = event["httpMethod"] requestParameters = event['queryString'] body = event['body'] if 'body' in event.keys() else '' response = wechat(httpMethod, requestParameters, body=body) return response
先從云函數(shù)入口函數(shù)開始，我們可以從 event 的 keys 里是否存在 Time 來(lái)判斷云函數(shù)是否是被定時(shí)器觸發(fā)的
#每天定時(shí)檢查刪除過(guò)期的激活碼
def check_del_vips():
vipMap = getvips()
if len(vipMap) < 1:
return
for vip in vipMap.keys():
if not checkVip(vipMap[vip]):
vipMap.pop(vip)
return cosWrite('vips.txt', json.dumps(vipMap, ensure_ascii=False))這里設(shè)置定時(shí)器來(lái)觸發(fā)云函數(shù)是為了每天檢查一遍有沒(méi)有激活碼失效了，失效的激活碼會(huì)被刪除掉。
def wechat(httpMethod, requestParameters, body=''):
if httpMethod == 'GET':
signature = requestParameters['signature']
timestamp = requestParameters['timestamp']
nonce = requestParameters['nonce']
echo_str = requestParameters['echostr']
try:
check_signature(WECHAT_TOKEN, signature, timestamp, nonce)
except InvalidSignatureException:
echo_str = 'error'
return apiReply(echo_str, txt=True, content_type="text/plain")
elif httpMethod == 'POST':
msg_signature = requestParameters['msg_signature']
timestamp = requestParameters['timestamp']
nonce = requestParameters['nonce']
try:
decrypted_xml = crypto.decrypt_message(
body,
msg_signature,
timestamp,
nonce
)
except (InvalidAppIdException, InvalidSignatureException):
return
msg = parse_message(decrypted_xml)
if msg.type == 'text':
reply = replyMessage(msg)
else:
reply = create_reply('哈? ??？\n搞不明白你給我發(fā)了啥~', msg)
reply = reply.render()
reply = crypto.encrypt_message(reply, nonce, timestamp)
return apiReply(reply, txt=True, content_type="application/xml")
else:
msg = parse_message(body)
reply = create_reply("喵嗚 ?'ω'?", msg).render()
reply = crypto.encrypt_message(reply, nonce, timestamp)
return apiReply(reply, txt=True, content_type="application/xml")如果云函數(shù)不是通過(guò)定時(shí)器觸發(fā)，那它就是通過(guò)后面我們要設(shè)置的 api 網(wǎng)關(guān)給觸發(fā)的，這時(shí)候就是我們的微信公眾號(hào)后臺(tái)發(fā)消息過(guò)來(lái)了，我們先用 crypto.decrypt\_message 來(lái)解密一下消息。
if msg.type == 'text':
reply = replyMessage(msg)
else:
reply = create_reply('哈? ??？\n搞不明白你給我發(fā)了啥~', msg)然后判斷一下消息的類型（文字、圖片、語(yǔ)音、視頻或者其他類型），如果不是文字消息，我們就先暫不處理啦 ~
def replyMessage(msg):
txt = msg.content
if '@' in txt:
keys = txt.split('@')
if keys[0] == '會(huì)員': # 會(huì)員@356 --> 添加一個(gè)365天的會(huì)員激活碼
flag,vip=addvip(keys[1])
return create_reply(f"您的激活碼:{vip}，有效期:{keys[1]}天" if flag else "添加失敗", msg)
return create_reply("喵嗚 ?'ω'?", msg)然后對(duì)于文字消息我們按照自己規(guī)定的命令格式來(lái)解析處理用戶消息即可。
# 是否開啟本地debug模式
debug = False
# 騰訊云對(duì)象存儲(chǔ)依賴
if debug:
from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
from qcloud_cos import CosServiceError
from qcloud_cos import CosClientError
else:
from qcloud_cos_v5 import CosConfig
from qcloud_cos_v5 import CosS3Client
from qcloud_cos_v5 import CosServiceError
from qcloud_cos_v5 import CosClientError
# 配置存儲(chǔ)桶
appid = '66666666666'
secret_id = u'xxxxxxxxxxxxxxx'
secret_key = u'xxxxxxxxxxxxxxx'
region = u'ap-chongqing'
bucket = 'name'+'-'+appid
# 對(duì)象存儲(chǔ)實(shí)例
config = CosConfig(Secret_id=secret_id, Secret_key=secret_key, Region=region)
client = CosS3Client(config)
# cos 文件讀寫
def cosRead(key):
try:
response = client.get_object(Bucket=bucket, Key=key)
txtBytes = response['Body'].get_raw_stream()
return txtBytes.read().decode()
except CosServiceError as e:
return ""
def cosWrite(key, txt):
try:
response = client.put_object(
Bucket=bucket,
Body=txt.encode(encoding="utf-8"),
Key=key,
)
return True
except CosServiceError as e:
return False
# api網(wǎng)關(guān)響應(yīng)集成
def apiReply(reply, txt=False, content_type='application/json', code=200):
return {
"isBase64Encoded": False,
"statusCode": code,
"headers": {'Content-Type': content_type},
"body": json.dumps(reply, ensure_ascii=False) if not txt else str(reply)
}以上就是如何基于Serverless借助微信公眾號(hào)簡(jiǎn)單管理用戶激活碼，小編相信有部分知識(shí)點(diǎn)可能是我們?nèi)粘９ぷ鲿?huì)見(jiàn)到或用到的。希望你能通過(guò)這篇文章學(xué)到更多知識(shí)。更多詳情敬請(qǐng)關(guān)注億速云行業(yè)資訊頻道。
免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。
億速云公眾號(hào)
手機(jī)網(wǎng)站二維碼