作為一名獨立開發者，最近我在考慮給自己的應用加入付費功能，然后應用的核心功能只需使用激活碼付費激活即可，這個需求涉及到了激活碼的保存、校驗和后臺管理，傳統的做法可能是自己購買服務器，搭建配置服務器環境，然后創建資料庫，撰寫后端業務邏輯代碼，必要的時候還要自己去寫一些前端的界面來管理后臺資料，
這是一個十分耗時且無趣的作業，本文則獨辟蹊徑，嘗試帶大家使用云函式 SCF 和物件存盤 COS，快速撰寫上線自己的用戶激活碼后端管理云函式，然后把自己的微信公眾號后臺做為應用前臺，簡單管理用戶激活碼，
效果展示
可以看到，現在我們只需要在自己的微信公眾號后臺回復 會員@激活時長，就可以添加并回復一個指定有效期的會員激活碼，實作了在微信公眾號簡單管理用戶激活碼的需求，
操作步驟
第一步：新建 python 云函式
參見之前的系列文章《萬物皆可 Serverless 之使用 SCF+COS 快速開發全堆疊應用》
第二步：撰寫云函式
話不多說，上代碼
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
# 騰訊云物件存盤依賴
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
# 配置存盤桶
appid = '66666666666'
secret_id = u'xxxxxxxxxxxxxxx'
secret_key = u'xxxxxxxxxxxxxxx'
region = u'ap-chongqing'
bucket = 'name'+'-'+appid
# 微信公眾號對接
wecaht_id = 'xxxxxxxxxxxxxxx'
WECHAT_TOKEN = 'xxxxxxxxxxxxxxxxxxx'
encoding_aes_key = 'xxxxxxxxxxxxxxxxxxxxxx'
# 物件存盤實體
config = CosConfig(Secret_id=secret_id, Secret_key=secret_key, Region=region)
client = CosS3Client(config)
#微信公眾號后臺訊息加解密實體
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
#獲取所有會員激活碼
def getvips():
vipMap = {}
vipTxt = cosRead('vips.txt') # 讀取資料
if len(vipTxt) > 0:
vipMap = json.loads(vipTxt)
return vipMap
#添加會員激活碼
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,''
#洗掉會員激活碼
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()
# 隨機生成激活碼
def randomKey():
return ''.join(random.sample('zyxwvutsrqponmlkjihgfedcba0123456789', 6))
#每天定時檢查洗掉過期的激活碼
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網關回應集成
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] == '會員': # 會員@356 --> 添加一個365天的會員激活碼
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搞不明白你給我發了啥~', 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(): # 來自定時觸發器
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
OK, 教程結束，
哈？你說沒看懂這堆代碼？
好吧，我再耐心給大家捋一下，這次可一定要記住了哈~
def main_handler(event, context):
if 'Time' in event.keys(): # 來自定時觸發器
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
先從云函式入口函式開始，我們可以從 event 的 keys 里是否存在 Time 來判斷云函式是否是被定時器觸發的
#每天定時檢查洗掉過期的激活碼
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))
這里設定定時器來觸發云函式是為了每天檢查一遍有沒有激活碼失效了，失效的激活碼會被洗掉掉，
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搞不明白你給我發了啥~', 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")
如果云函式不是通過定時器觸發，那它就是通過后面我們要設定的 api 網關給觸發的，這時候就是我們的微信公眾號后臺發訊息過來了，我們先用 crypto.decrypt\_message 來解密一下訊息，
if msg.type == 'text':
reply = replyMessage(msg)
else:
reply = create_reply('哈? ??？\n搞不明白你給我發了啥~', msg)
然后判斷一下訊息的型別（文字、圖片、語音、視頻或者其他型別），如果不是文字訊息，我們就先暫不處理啦 ~
def replyMessage(msg):
txt = msg.content
if '@' in txt:
keys = txt.split('@')
if keys[0] == '會員': # 會員@356 --> 添加一個365天的會員激活碼
flag,vip=addvip(keys[1])
return create_reply(f"您的激活碼:{vip}，有效期:{keys[1]}天" if flag else "添加失敗", msg)
return create_reply("喵嗚 ?'ω'?", msg)
然后對于文字訊息我們按照自己規定的命令格式來決議處理用戶訊息即可，
# 是否開啟本地debug模式
debug = False
# 騰訊云物件存盤依賴
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
# 配置存盤桶
appid = '66666666666'
secret_id = u'xxxxxxxxxxxxxxx'
secret_key = u'xxxxxxxxxxxxxxx'
region = u'ap-chongqing'
bucket = 'name'+'-'+appid
# 物件存盤實體
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網關回應集成
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)
}
其他的諸如騰訊云 cos 的讀寫和訊息回復的格式之類的問題，就不再細說了，之前的系列文章《萬物皆可 Serverless 之使用 SCF+COS 快速開發全堆疊應用》 里面都有詳細講到，
Serverless Framework 30 天試用計劃
我們誠邀您來體驗最便捷的 Serverless 開發和部署方式，在試用期內，相關聯的產品及服務均提供免費資源和專業的技術支持，幫助您的業務快速、便捷地實作 Serverless！
詳情可查閱：Serverless Framework 試用計劃
One More Thing
3 秒你能做什么？喝一口水，看一封郵件，還是 —— 部署一個完整的 Serverless 應用？
復制鏈接至 PC 瀏覽器訪問：https://serverless.cloud.tencent.com/deploy/express
3 秒極速部署，立即體驗史上最快的 Serverless HTTP 實戰開發！
傳送門：
- GitHub: github.com/serverless
- 官網：serverless.com
歡迎訪問：Serverless 中文網，您可以在 最佳實踐 里體驗更多關于 Serverless 應用的開發！
推薦閱讀：《Serverless 架構：從原理、設計到專案實戰》
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/5320.html
標籤：其他