Shulou(Shulou.com)06/01 Report--

This article is about how to simply manage user activation codes based on Serverless with the help of Wechat official account. The editor thinks it is very practical, so I share it with you to learn. I hope you can get something after reading this article. Without saying much, let's take a look at it.

As an independent developer, I have recently been considering adding a paid feature to my application, and then the core function of the application can be activated for a fee using the activation code. This requirement involves the preservation, verification and background management of the activation code. The traditional approach may be to purchase the server, set up the configuration server environment, then create the database, write the back-end business logic code, and write some front-end interfaces to manage the background data if necessary.

This is a very time-consuming and boring job.

Try to use cloud function SCF and object storage COS to quickly write and launch your own user activation codes to manage cloud functions at the backend, and then use your official Wechat account backend as the application foreground to simply manage user activation codes.

As you can see, now we only need to reply to the member @ activation time in our official Wechat account to add and reply a member activation code with a specified period of validity, realizing the need to simply manage user activation codes on Wechat official account.

Operation step 1: create a new python cloud function step 2: write a cloud function

Don't say much, put on the code.

Import jsonfrom wechatpy.replies import ArticlesReplyfrom wechatpy.utils import check_signaturefrom wechatpy.crypto import WeChatCryptofrom wechatpy import parse_message, create_replyfrom wechatpy.exceptions import InvalidSignatureException Whether InvalidAppIdExceptionimport datetimeimport random# enables local debug mode debug = False# Tencent Cloud object storage depends on if debug: from qcloud_cos import CosConfig from qcloud_cos import CosS3Client from qcloud_cos import CosServiceError from qcloud_cos import CosClientErrorelse: from qcloud_cos_v5 import CosConfig from qcloud_cos_v5 import CosS3Client from qcloud_cos_v5 import CosServiceError from qcloud_cos_v5 import CosClientError# configuration bucket appid = '66666666666'secret_id = ublicxxxxxxxxxxxxxxxxxxxsecret_ Key = u'xxxxxxxxxxxxxxx'region = u'ap-chongqing'bucket = 'name'+'-'+appid# Wechat official account docking wecaht_id =' xxxxxxxxxxxxxxx'WECHAT_TOKEN = 'xxxxxxxxxxxxxxxxxxx'encoding_aes_key =' xxxxxxxxxxxxxxxxxxxxxx'# object storage instance config = CosConfig (Secret_id=secret_id Secret_key=secret_key, Region=region) client = CosS3Client (config) # Wechat official account backend message encryption and decryption example crypto = WeChatCrypto (WECHAT_TOKEN, encoding_aes_key, wecaht_id) # cos file read and write 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# get all member activation codes def getvips (): vipMap = {} vipTxt = cosRead ('vips.txt') # read data if len (vipTxt) > 0: vipMap = json.loads (vipTxt) return vipMap# add member activation code 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,''# Delete the member activation code 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# get today's date def today (): return datetime.datetime.now (). Strftime ("% Y-%m-%d") # determine whether the activation code expires def checkVip (t): return t = = today () # randomly generate the activation code def randomKey (): return'. Join (random.sample ('zyxwvutsrqponmlkjihgfedcba0123456789') 6)) # check regularly every day to delete the expired activation code 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 -->

Add a 365-day member activation code flag,vip=addvip (keys [1]) return create_reply (f "your activation code: {vip}) Validity period: {keys [1]} days "if flag else" failed to add, msg) return create_reply ("meow 'omega' milk", 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 (' ha ◔ '◔? \ nI don't know what you sent me ~', 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 ("Meow 'omega' cow", 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 (): # from the timing trigger 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, end of tutorial

Huh? You said you didn't understand this pile of code?

All right, let me be patient and straighten it out for you, but you must remember it this time.

Def main_handler (event, context): if 'Time' in event.keys (): # from the timing trigger 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

Let's start with the entry function of the cloud function. We can judge whether the cloud function is triggered by timer from whether there is a Time in the keys of event.

# check regularly every day to delete expired activation codes 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 -->

Add a 365-day member activation code flag,vip=addvip (keys [1]) return create_reply (f "your activation code: {vip}, validity period: {keys [1]} days" if flag else "add failed", msg) return create_reply ("Meow 'omega' cards", msg)

Then for the text message, we can parse and process the user message according to our own command format.

# whether to enable local debug mode debug = False# Tencent Cloud object storage depends on if debug: from qcloud_cos import CosConfig from qcloud_cos import CosS3Client from qcloud_cos import CosServiceError from qcloud_cos import CosClientErrorelse: from qcloud_cos_v5 import CosConfig from qcloud_cos_v5 import CosS3Client from qcloud_cos_v5 import CosServiceError from qcloud_cos_v5 import CosClientError# configuration bucket appid = '66666666666'secret_id = u'xxxxxxxxxxxxxxx'secret_key = u 'xxxxxxxxxxxxxxx'region = u'ap-chongqing'bucket =' name'+'-'+appid# object storage instance config = CosConfig (Secret_id=secret_id Secret_key=secret_key, Region=region) client = CosS3Client (config) # cos file read and write 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 ase: return False# api gateway response integration 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)} the above is how to simply manage user activation codes based on Serverless with the help of Wechat official account The editor believes that there are some knowledge points that we may see or use in our daily work. I hope you can learn more from this article. For more details, please follow the industry information channel.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.