如何基于Serverless借助微信公眾號簡單管理用戶激活碼

本篇文章給大家分享的是有關如何基于Serverless借助微信公眾號簡單管理用戶激活碼，小編覺得挺實用的，因此分享給大家學習，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著小編一起來看看吧。

作為一名獨立開發者，最近我在考慮給自己的應用加入付費功能，然后應用的核心功能只需使用激活碼付費激活即可。這個需求涉及到了激活碼的保存、校驗和后臺管理，傳統的做法可能是自己購買服務器，搭建配置服務器環境，然后創建數據庫，編寫后端業務邏輯代碼，必要的時候還要自己去寫一些前端的界面來管理后臺數據。

這是一個十分耗時且無趣的工作。

嘗試帶大家使用云函數 SCF 和對象存儲 COS，快速編寫上線自己的用戶激活碼后端管理云函數，然后把自己的微信公眾號后臺做為應用前臺，簡單管理用戶激活碼。

可以看到，現在我們只需要在自己的微信公眾號后臺回復 會員@激活時長，就可以添加并回復一個指定有效期的會員激活碼，實現了在微信公眾號簡單管理用戶激活碼的需求。

操作步驟

第一步：新建 python 云函數

第二步：編寫云函數

話不多說，上代碼

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)
    }

以上就是如何基于Serverless借助微信公眾號簡單管理用戶激活碼，小編相信有部分知識點可能是我們日常工作會見到或用到的。希望你能通過這篇文章學到更多知識。更多詳情敬請關注億速云行業資訊頻道。