是的，你沒聽錯，這一次我來帶大家直接上手運營微信公眾號，
本文來自 Serverless 社區用戶「乂乂又又」供稿
震驚，Awesome，哼，我才不信捏，所謂無圖無真相 ~
效果展示
更多的體驗，可以關注我的微信公眾號： 乂乂又又 （僅供測驗，不要亂搞哈~）
嗯，這次我信了，快點教一下我吧，嚶嚶嚶~
操作步驟
在上一篇《萬物皆可Serverless之使用SCF+COS快速開發全堆疊應用》教程中，
我們用騰訊云無服務器云函式 SCF 和物件存盤實作了一個后端云函式，這個云函式可以根據我們的請求回傳對應的結果，
現在我們將嘗試在這個云函式的基礎上決議微信 XML 訊息，實作公眾號訊息的自動回復，關鍵詞回復，文字選單等功能，
第一步：添加相關依賴
為了快速完成開發，這里我們選擇 python 第三方開源庫 wechatpy 來接入微信公眾平臺，
wechatpy 支持以下功能
- 普通公眾平臺被動回應和主動呼叫 API
- 企業微信 API
- 微信支付 API
- 第三方平臺代公眾號呼叫介面 API
- 小程式云開發 API
可見功能是十分完整的，不僅支持普通公眾平臺主被動呼叫，企業微信和微信支付，甚至還支持第三方平臺代公眾號呼叫介面，拿來運營微信公眾號是十分綽綽有余的~
由于騰訊云函式的運行環境中缺少第三方庫，需要我們自己手動上傳添加依賴，這里我們需要添加的第三方依賴有：wechatpy、otionaldict、xmltodict 以及 timeout\_decorator
其中 wechatpy 需要依賴 otionaldict、xmltodict，timeout\_decorator 是用來限制函式運行時長的，具體的依賴檔案可以自行 pip 安裝后 copy 到云函式專案根目錄，如上圖，
第二步：接入微信公眾號
這里需要記下自己的 AppID、Token 和 EncodingAESKey，訊息加密方式建議選為安全模式，這個頁面先不要關，一會兒上線發布好云函式還需要過來再次修改配置，
第三步：撰寫云函式決議并回復微信公眾號訊息
這一步可以直接參考 wechatpy 的官方檔案
Life is short, show me the code.
這里我就直接上代碼了（原始業務代碼已略去，可以按照自己的需求開發）
import json
import timeout_decorator
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
# 是否開啟本地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
def getReplys():
replyMap = {}
replyTxt = cosRead('Replys.txt') # 讀取資料
if len(replyTxt) > 0:
replyMap = json.loads(replyTxt)
return replyMap
def addReplys(reply):
replyMap = getReplys()
if len(replyMap) > 0:
replyMap[reply]='我是黑名單'
return cosWrite('Replys.txt', json.dumps(replyMap, ensure_ascii=False)) if len(replyMap) > 0 else False
def delReplys(reply):
replyMap = getReplys()
if len(replyMap) > 0:
replyMap.pop(reply)
return cosWrite('Replys.txt', json.dumps(replyMap, ensure_ascii=False)) if len(replyMap) > 0 else False
# 微信公眾號對接
wecaht_id = 'xxxxxxxxxxxxxxx'
WECHAT_TOKEN = 'xxxxxxxxxxxxxxxxxxx'
encoding_aes_key = 'xxxxxxxxxxxxxxxxxxxxxx'
crypto = WeChatCrypto(WECHAT_TOKEN, encoding_aes_key, wecaht_id)
# 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
ip = msg.source
print('請求資訊--->'+ip+'%'+txt) # 用來在騰訊云控制臺列印請求日志
replysTxtMap = getReplys() # 獲取回復關鍵詞
if '@' in txt:
keys = txt.split('@')
if keys[0] == '電影': #do something
return
if keys[0] == '音樂': #do something
return
if keys[0] == '下架': #do something
return
if keys[0] == '上架': #do something
return
if keys[0] == '回復': #do something
return
if keys[0] == '洗掉': #do something
return
elif txt in replysTxtMap.keys(): # 如果訊息在回復關鍵詞內則自動回復
return create_reply(replysTxtMap[txt], 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)
elif msg.type == 'image':
reply = create_reply('哈? ??？\n好端端的，給我發圖片干啥~', msg)
elif msg.type == 'voice':
reply = create_reply('哈? ??？\n好端端的，給我發語音干啥~', msg)
else:
reply = create_reply('哈? ??？\n搞不明白你給我發了啥~', msg)
reply = reply.render()
print('回傳結果--->'+str(reply)) # 用來在騰訊云控制臺列印請求日志
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)
reply = reply.render()
print('回傳結果--->'+str(reply)) # 用來在騰訊云控制臺列印請求日志
reply = crypto.encrypt_message(reply, nonce, timestamp)
return apiReply(reply, txt=True, content_type="application/xml")
@timeout_decorator.timeout(4, timeout_exception=StopIteration)
def myMain(httpMethod, requestParameters, body=''):
return wechat(httpMethod, requestParameters, body=body)
def timeOutReply(httpMethod, requestParameters, body=''):
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)
reply = create_reply("出了點小問題，請稍后再試", msg).render()
print('回傳結果--->'+str(reply)) # 用來在騰訊云控制臺列印請求日志
reply = crypto.encrypt_message(reply, nonce, timestamp)
return apiReply(reply, txt=True, content_type="application/xml")
def main_handler(event, context):
body = ''
httpMethod = event["httpMethod"]
requestParameters = event['queryString']
if 'body' in event.keys():
body = event['body']
try:
response = myMain(httpMethod, requestParameters, body=body)
except:
response = timeOutReply(httpMethod, requestParameters, body=body)
return response
請求引數決議和COS讀寫部分可參考上一篇《萬物皆可 Serverless 之使用 SCF+COS 快速開發全堆疊應用》教程
下面我來捋一下整個云函式的思路
def main_handler(event, context):
body = ''
httpMethod = event["httpMethod"]
requestParameters = event['queryString']
if 'body' in event.keys():
body = event['body']
try:
response = myMain(httpMethod, requestParameters, body=body)
except:
response = timeOutReply(httpMethod, requestParameters, body=body)
return response
我們先從 main\_handler 入手，
這里我們通過 API 網關觸發云函式在 event 里拿到了微信公眾號請求的方法、頭部和請求體，然后傳給 myMain 函式做處理，需要注意的是 myMain 是通過 timeout\_decorator 包裝的限時運行函式，
@timeout_decorator.timeout(4, timeout_exception=StopIteration)
def myMain(httpMethod, requestParameters, body=''):
return wechat(httpMethod, requestParameters, body=body)
當 myMain 函式運行市場超過設定的 4 秒后，就會拋出例外，
然后我們可以通過設定一個 timeOutReply 函式來處理超時后的微信公眾號訊息回復，可是為什么要這么做呢？
可以看到，當云函式運行超時后，微信這邊就會顯示「該公眾號提供的服務器出現故障，請稍后再試」
這對用戶體驗是極不友好的，所以我們需要一個函式超時后的回復來兜底，
那么對于一次微信公眾號后臺訊息請求多長時間算是超時呢？答案是 5 秒左右，從云函式后臺的呼叫日志我們可以得到這個結果，
不過需要注意的是對于用戶的一次訊息請求，微信可能會每隔 1 秒左右重撥一次請求，直到收到服務器第一次回應，另外，超過 3 次應該就不會再重撥了，并且在 5 秒超時后即使云函式呼叫成功并回傳了資料，用戶也不會再接收到訊息了~
所以我們就很有必要將自己的云函式的運行時長限制在 5 秒之內了！
當然只通過配置云函式超時時長得方式來處理是不正確的，因為這樣做云函式超時后就被系統停掉了，并不會向微信回傳訊息，所以從一開始我就匯入了 timeout\_decorator 庫來限制主函式的運行時長，并用一個超時后回復函式來兜底，
另外值得一提的是，在我原始的業務代碼中是有一些爬蟲，這些爬蟲本來我是單執行緒順序執行的，考慮到超時問題，我在微信云函式版這里全部改成了多執行緒運行來壓縮時間，所以如果你也有一些比較耗時的小任務話，也可以嘗試通過多執行緒的方式來壓縮云函式的運行時長，
我們接著向下看：
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)
elif msg.type == 'image':
reply = create_reply('哈? ??？\n好端端的，給我發圖片干啥~', msg)
elif msg.type == 'voice':
reply = create_reply('哈? ??？\n好端端的，給我發語音干啥~', msg)
else:
reply = create_reply('哈? ??？\n搞不明白你給我發了啥~', msg)
reply = reply.render()
print('回傳結果--->'+str(reply)) # 用來在騰訊云控制臺列印請求日志
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)
reply = reply.render()
print('回傳結果--->'+str(reply)) # 用來在騰訊云控制臺列印請求日志
reply = crypto.encrypt_message(reply, nonce, timestamp)
return apiReply(reply, txt=True, content_type="application/xml")
這里的 wechat 函式就是整個微信訊息的決議程序，首先判斷請求方法是 GET 還是 POST，GET 方法只在第一次系結微信后臺時會用到，這時我們會從微信服務器推送的請求引數中拿到 signature, timestamp, echostr 和 nonce 引數，
check_signature(WECHAT_TOKEN, signature, timestamp, nonce)
我們只需根據自己的公眾號 token 和來生成簽名與微信服務器傳過來的 signature 對比看是否一致，若一致就說明我們的訊息加解密驗證是OK的，然后再將 echostr 原樣回傳即可接入微信公眾號后臺，
接入好微信公眾號后，如果有用戶在后臺給我們發送訊息，這里云函式收到的就是 POST 方法，
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)
elif msg.type == 'image':
reply = create_reply('哈? ??？\n好端端的，給我發圖片干啥~', msg)
elif msg.type == 'voice':
reply = create_reply('哈? ??？\n好端端的，給我發語音干啥~', msg)
else:
reply = create_reply('哈? ??？\n搞不明白你給我發了啥~', msg)
reply = reply.render()
print('回傳結果--->'+str(reply)) # 用來在騰訊云控制臺列印請求日志
reply = crypto.encrypt_message(reply, nonce, timestamp)
return apiReply(reply, txt=True, content_type="application/xml")
然后我們根據前面在微信公眾號后臺拿到的 id，token 和 aes 加密 key 來初始化訊息加解密實體并解密還原用戶發送的訊息
# 微信公眾號對接
wecaht_id = 'xxxxxxxxxxxxxxx'
WECHAT_TOKEN = 'xxxxxxxxxxxxxxxxxxx'
encoding_aes_key = 'xxxxxxxxxxxxxxxxxxxxxx'
crypto = WeChatCrypto(WECHAT_TOKEN, encoding_aes_key, wecaht_id)
接著判斷一下訊息型別，不同型別的訊息可自行處理
msg = parse_message(decrypted_xml)
if msg.type == 'text':
reply = replyMessage(msg)
elif msg.type == 'image':
reply = create_reply('哈? ??？ 好端端的，給我發圖片干啥~', msg)
elif msg.type == 'voice':
reply = create_reply('哈? ??？ 好端端的，給我發語音干啥~', msg)
else:
reply = create_reply('哈? ??？ 搞不明白你給我發了啥~', msg)
需要注意的是當一個用戶新關注自己的公眾號時，我們收到的是一個其他型別的訊息，也就是上面的最后一個判斷項，這里你可以自己設定新關注用戶的歡迎語
reply = create_reply('哈? ??？\n搞不明白你給我發了啥~', msg)
reply = reply.render()
print('回傳結果--->'+str(reply)) # 用來在騰訊云控制臺列印請求日志
reply = crypto.encrypt_message(reply, nonce, timestamp)
return apiReply(reply, txt=True, content_type="application/xml")
之后我們通過 create\_reply 來快速創建一個文本回復，并通過 render() 來生成 xml 回復訊息文本，因為我之前在后臺設定的是安全模式，所以還需要把 xml 重新通過 crypto.encrypt\_message 方法加密，然后才能把加密后的回復訊息回傳給微信服務器，
上一篇文章我有提到我們不能直接回傳訊息，需要按照特定的格式回傳資料（API 網關需要開啟回應集成）
# 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)
}
第四步：上線發布云函式、添加 API 網關觸發器、啟用回應集成
參考上一篇教程 《萬物皆可 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/6772.html
標籤：其他