共计 8339 个字符，预计需要花费 21 分钟才能阅读完成。
作为一名独立开发者，最近我在考虑给自己的应用加入付费功能，然后应用的核心功能只需使用激活码付费激活即可。这个需求涉及到了激活码的保存、校验和后台管理，传统的做法可能是自己购买服务器，搭建配置服务器环境，然后创建数据库，编写后端业务逻辑代码，必要的时候还要自己去写一些前端的界面来管理后台数据。
这是一个十分耗时且无趣的工作。本文则独辟蹊径，尝试带大家使用云函数 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 responseOK, 教程结束，
哈？你说没看懂这堆代码？
好吧，我再耐心给大家捋一下，这次可一定要记住了哈~
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.tenc…
3 秒极速部署，立即体验史上最快的 Serverless HTTP 实战开发！
传送门：
- GitHub: github.com/serverless
- 官网：serverless.com
欢迎访问：Serverless 中文网，您可以在 最佳实践 里体验更多关于 Serverless 应用的开发！
推荐阅读：《Serverless 架构：从原理、设计到项目实战》
正文完
