巧用 Serverless，輕鬆搭建微信公衆號的智能後臺服務
通常來講，想給微信公衆號增長更多的功能，須要有一臺服務器，來進行公衆號後臺服務的搭建。那麼在 Serverless 架構下，是否有更簡便的方法來實現這麼一個公衆號後臺呢？咱們試試？html
初步搭建
Serverless 原生開發
首先咱們固然要有一個微信公衆號！python
接下來，咱們要爲咱們的函數計算服務申請固定 IP：git
點擊白名單以後，咱們能夠填寫表單，完成固定公網出口 IP 的申請。github
接下來進行代碼開發。數據庫
- 將函數綁定到公衆號後臺，參考文檔：https://developers.weixin.qq.com/doc/offiaccount/Basic_Information/Access_Overview.html
咱們能夠先在函數中按照文檔完成一個基本的鑑定功能：
def checkSignature(param):
'''
文檔地址：https://developers.weixin.qq.com/doc/offiaccount/Basic_Information/Access_Overview.html
:param param:
:return:
'''
signature = param['signature']
timestamp = param['timestamp']
nonce = param["nonce"]
tmparr = [wxtoken, timestamp, nonce]
tmparr.sort()
tmpstr = ''.join(tmparr)
tmpstr = hashlib.sha1(tmpstr.encode("utf-8")).hexdigest()
return tmpstr == signature
再定義一個基本的回覆方法：express
def response(body, status=200):
return {
"isBase64Encoded": False,
"statusCode": status,
"headers": {"Content-Type": "text/html"},
"body": body
}
而後在函數入口處：編程
def main_handler(event, context):
if 'echostr' in event['queryString']: # 接入時的校驗
return response(event['queryString']['echostr'] if checkSignature(event['queryString']) else False)
配置咱們 Yaml：json
# serverless.yml
Weixin_GoServerless:
component: "@serverless/tencent-scf"
inputs:
name: Weixin_GoServerless
codeUri: ./Admin
handler: index.main_handler
runtime: Python3.6
region: ap-shanghai
description: 微信公衆號後臺服務器配置
memorySize: 128
timeout: 20
environment:
variables:
wxtoken: 自定義一個字符串
appid: 暫時不寫
secret: 暫時不寫
events:
- apigw:
name: Weixin_GoServerless
parameters:
protocols:
- https
environment: release
endpoints:
- path: /
method: ANY
function:
isIntegratedResponse: TRUE
執行代碼，完成部署：後端
接下來在公衆號後臺，選擇基本配置：api
選擇修改配置：
這裏要注意：
-
URL，寫咱們剛纔部署完成返回給咱們的地址，而且在最後加一個 /
-
Token，寫咱們 Yaml 中的 wxtoken，兩個地方要保持同樣的字符串
-
EncodingAESKey，能夠點擊隨機生成
-
消息加密方法能夠選擇明文
完成以後，咱們能夠點擊提交：
看到提交成功，就說明咱們已經完成了第一步驟的綁定。接下來，咱們到函數的後臺：
打開固定出口 IP，看到 IP 地址以後，複製 IP 地址：
點擊查看->修改，並將 IP 地址複製粘貼進來，保存。
同時咱們查看開發者 ID 和密碼：
並將這兩個內容複製粘貼，放到咱們環境變量中：
至此，咱們完成了一個公衆號後臺服務的綁定。
爲了方便以後的操做，先獲取一下全局變量：
wxtoken = os.environ.get('wxtoken')
appid = os.environ.get('appid')
secret = os.environ.get('secret')
- 接下來對各個模塊進行編輯（本文只提供部分簡單基礎的模塊，更多功能實現能夠參考微信公衆號文檔實現）
- 獲取 AccessToken 模塊：
def getAccessToken():
'''
文檔地址：https://developers.weixin.qq.com/doc/offiaccount/Basic_Information/Get_access_token.html
正常返回：{"access_token":"ACCESS_TOKEN","expires_in":7200}
異常返回：{"errcode":40013,"errmsg":"invalid appid"}
:return:
'''
url = "https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=%s&secret=%s" % (appid, secret)
accessToken = json.loads(urllib.request.urlopen(url).read().decode("utf-8"))
print(accessToken)
return None if "errcode" in accessToken else accessToken["access_token"]
- 建立自定義菜單模塊：
def setMenu(menu):
'''
文檔地址：https://developers.weixin.qq.com/doc/offiaccount/Custom_Menus/Creating_Custom-Defined_Menu.html
正確返回：{"errcode":0,"errmsg":"ok"}
異常返回：{"errcode":40018,"errmsg":"invalid button name size"}
:return:
'''
accessToken = getAccessToken()
if not accessToken:
return "Get Access Token Error"
url = "https://api.weixin.qq.com/cgi-bin/menu/create?access_token=%s" % accessToken
postData = urllib.parse.urlencode(menu).encode("utf-8")
requestAttr = urllib.request.Request(url=url, data=postData)
responseAttr = urllib.request.urlopen(requestAttr)
responseData = json.loads(responseAttr.read())
return responseData['errmsg'] if "errcode" in responseData else "success"
- 常見消息回覆模塊：
def textXML(body, event):
'''
:param body: {"msg": "test"}
msg: 必填，回覆的消息內容（換行：在content中可以換行，微信客戶端就支持換行顯示）
:param event:
:return:
'''
return """<xml><ToUserName><![CDATA[{toUser}]]></ToUserName>
<FromUserName><![CDATA[{fromUser}]]></FromUserName>
<CreateTime>{time}</CreateTime>
<MsgType><![CDATA[text]]></MsgType>
<Content><![CDATA[{msg}]]></Content></xml>""".format(toUser=event["FromUserName"],
fromUser=event["ToUserName"],
time=int(time.time()),
msg=body["msg"])
def pictureXML(body, event):
'''
:param body: {"media_id": 123}
media_id: 必填，經過素材管理中的接口上傳多媒體文件，獲得的id。
:param event:
:return:
'''
return """<xml><ToUserName><![CDATA[{toUser}]]></ToUserName>
<FromUserName><![CDATA[{fromUser}]]]></FromUserName>
<CreateTime>{time}</CreateTime>
<MsgType><![CDATA[image]]></MsgType>
<Image>
<MediaId><![CDATA[{media_id}]]></MediaId>
</Image></xml>""".format(toUser=event["FromUserName"],
fromUser=event["ToUserName"],
time=int(time.time()),
media_id=body["media_id"])
def voiceXML(body, event):
'''
:param body: {"media_id": 123}
media_id: 必填，經過素材管理中的接口上傳多媒體文件，獲得的id
:param event:
:return:
'''
return """<xml><ToUserName><![CDATA[{toUser}]]></ToUserName>
<FromUserName><![CDATA[{fromUser}]]></FromUserName>
<CreateTime>{time}</CreateTime>
<MsgType><![CDATA[voice]]></MsgType>
<Voice>
<MediaId><![CDATA[{media_id}]]></MediaId>
</Voice></xml>""".format(toUser=event["FromUserName"],
fromUser=event["ToUserName"],
time=int(time.time()),
media_id=body["media_id"])
def videoXML(body, event):
'''
:param body: {"media_id": 123, "title": "test", "description": "test}
media_id: 必填，經過素材管理中的接口上傳多媒體文件，獲得的id
title:：選填，視頻消息的標題
description：選填，視頻消息的描述
:param event:
:return:
'''
return """<xml><ToUserName><![CDATA[{toUser}]]></ToUserName>
<FromUserName><![CDATA[{fromUser}]]></FromUserName>
<CreateTime>{time}</CreateTime>
<MsgType><![CDATA[video]]></MsgType>
<Video>
<MediaId><![CDATA[{media_id}]]></MediaId>
<Title><![CDATA[{title}]]></Title>
<Description><![CDATA[{description}]]></Description>
</Video></xml>""".format(toUser=event["FromUserName"],
fromUser=event["ToUserName"],
time=int(time.time()),
media_id=body["media_id"],
title=body.get('title', ''),
description=body.get('description', ''))
def musicXML(body, event):
'''
:param body: {"media_id": 123, "title": "test", "description": "test}
media_id：必填，縮略圖的媒體id，經過素材管理中的接口上傳多媒體文件，獲得的id
title：選填，音樂標題
description：選填，音樂描述
url：選填，音樂連接
hq_url：選填，高質量音樂連接，WIFI環境優先使用該連接播放音樂
:param event:
:return:
'''
return """<xml><ToUserName><![CDATA[{toUser}]]></ToUserName>
<FromUserName><![CDATA[{fromUser}]]></FromUserName>
<CreateTime>{time}</CreateTime>
<MsgType><![CDATA[music]]></MsgType>
<Music>
<Title><![CDATA[{title}]]></Title>
<Description><![CDATA[{description}]]></Description>
<MusicUrl><![CDATA[{url}]]></MusicUrl>
<HQMusicUrl><![CDATA[{hq_url}]]></HQMusicUrl>
<ThumbMediaId><![CDATA[{media_id}]]></ThumbMediaId>
</Music></xml>""".format(toUser=event["FromUserName"],
fromUser=event["ToUserName"],
time=int(time.time()),
media_id=body["media_id"],
title=body.get('title', ''),
url=body.get('url', ''),
hq_url=body.get('hq_url', ''),
description=body.get('description', ''))
def articlesXML(body, event):
'''
:param body: 一個list [{"title":"test", "description": "test", "picUrl": "test", "url": "test"}]
title：必填，圖文消息標題
description：必填，圖文消息描述
picUrl：必填，圖片連接，支持JPG、PNG格式，較好的效果爲大圖360*200，小圖200*200
url：必填，點擊圖文消息跳轉連接
:param event:
:return:
'''
if len(body["articles"]) > 8: # 最多隻容許返回8個
body["articles"] = body["articles"][0:8]
tempArticle = """<item>
<Title><![CDATA[{title}]]></Title>
<Description><![CDATA[{description}]]></Description>
<PicUrl><![CDATA[{picurl}]]></PicUrl>
<Url><![CDATA[{url}]]></Url>
</item>"""
return """<xml><ToUserName><![CDATA[{toUser}]]></ToUserName>
<FromUserName><![CDATA[{fromUser}]]></FromUserName>
<CreateTime>{time}</CreateTime>
<MsgType><![CDATA[news]]></MsgType>
<ArticleCount>{count}</ArticleCount>
<Articles>
{articles}
</Articles></xml>""".format(toUser=event["FromUserName"],
fromUser=event["ToUserName"],
time=int(time.time()),
count=len(body["articles"]),
articles="".join([tempArticle.format(
title=eveArticle['title'],
description=eveArticle['description'],
picurl=eveArticle['picurl'],
url=eveArticle['url']
) for eveArticle in body["articles"]]))
-
對 main_handler 進行修改，使其：
-
識別綁定功能
-
識別基本信息
-
識別特殊額外請求（例如經過url觸發自定義菜單的更新）
-
總體代碼：
def main_handler(event, context):
print('event: ', event)
if event["path"] == '/setMenu': # 設置菜單接口
menu = {
"button": [
{
"type": "view",
"name": "精彩文章",
"url": "https://mp.weixin.qq.com/mp/homepage?__biz=Mzg2NzE4MDExNw==&hid=2&sn=168bd0620ee79cd35d0a80cddb9f2487"
},
{
"type": "view",
"name": "開源項目",
"url": "https://mp.weixin.qq.com/mp/homepage?__biz=Mzg2NzE4MDExNw==&hid=1&sn=69444401c5ed9746aeb1384fa6a9a201"
},
{
"type": "miniprogram",
"name": "在線編程",
"appid": "wx453cb539f9f963b2",
"pagepath": "/page/index"
}]
}
return response(setMenu(menu))
if 'echostr' in event['queryString']: # 接入時的校驗
return response(event['queryString']['echostr'] if checkSignature(event['queryString']) else False)
else: # 用戶消息/事件
event = getEvent(event)
if event["MsgType"] == "text":
# 文本消息
return response(body=textXML({"msg": "這是一個文本消息"}, event))
elif event["MsgType"] == "image":
# 圖片消息
return response(body=textXML({"msg": "這是一個圖片消息"}, event))
elif event["MsgType"] == "voice":
# 語音消息
pass
elif event["MsgType"] == "video":
# 視頻消息
pass
elif event["MsgType"] == "shortvideo":
# 小視頻消息
pass
elif event["MsgType"] == "location":
# 地理位置消息
pass
elif event["MsgType"] == "link":
# 連接消息
pass
elif event["MsgType"] == "event":
# 事件消息
if event["Event"] == "subscribe":
# 訂閱事件
if event.get('EventKey', None):
# 用戶未關注時，進行關注後的事件推送（帶參數的二維碼）
pass
else:
# 普通關注
pass
elif event["Event"] == "unsubscribe":
# 取消訂閱事件
pass
elif event["Event"] == "SCAN":
# 用戶已關注時的事件推送（帶參數的二維碼）
pass
elif event["Event"] == "LOCATION":
# 上報地理位置事件
pass
elif event["Event"] == "CLICK":
# 點擊菜單拉取消息時的事件推送
pass
elif event["Event"] == "VIEW":
# 點擊菜單跳轉連接時的事件推送
pass
在上述代碼中能夠看到：
if event["MsgType"] == "text":
# 文本消息
return response(body=textXML({"msg": "這是一個文本消息"}, event))
elif event["MsgType"] == "image":
# 圖片消息
return response(body=textXML({"msg": "這是一個圖片消息"}, event))
這裏就是說，當用戶發送了文本消息時候，咱們給用戶回覆一個文本消息：這是一個文本消息。當用戶發送了一個圖片，咱們給用戶返回這是一個圖片消息，用這兩個功能測試咱們這個後臺的連通性：
能夠看到，系統已經能夠正常返回。
有人問了，這樣一個簡單的 Demo 有什麼意義呢？能夠告訴你們，咱們能夠很輕量地經過一個函數來實現微信公衆號的後端服務；這裏都是基礎能力，咱們能夠在這個基礎能力之上，盡情添加創新力，例如：
-
用戶傳過來的是圖片消息，咱們能夠經過一些識圖 API 告訴用戶這個圖片包括了什麼？
-
用戶傳過來的是文字消息，咱們能夠先設定一些幫助信息/檢索信息進行對比，若是沒找到就給用戶開啓聊天功能（這裏涉及到人工智能中的天然語言處理，例如文本類似度檢測）
-
若是用戶發送的是語音，咱們還能夠將其轉成文本，生成對話消息，而後再轉換成語音返回給用戶
-
若是用戶發送了地理位置信息，咱們能夠返回用戶所在經緯度的街景信息，或者周邊生活服務信息等
-
留給你們想象！
使用 Werobot 框架
上面的方法，是經過 Serverless 原生開發的方法進行對接。除此以外，咱們還能夠選擇一些已有的框架，例如werobot等。
WeRoBot 是一個微信公衆號開發框架。經過 Serverless Component 中的tencent-werobot組件快速部署該框架：
Weixin_Werobot:
component: "@serverless/tencent-werobot"
inputs:
functionName: Weixin_Werobot
code: ./test
werobotProjectName: app
werobotAttrName: robot
functionConf:
timeout: 10
memorySize: 256
environment:
variables:
wxtoken: 你的token
apigatewayConf:
protocols:
- http
environment: release
而後新建代碼：
import os
import werobot
robot = werobot.WeRoBot(token=os.environ.get('wxtoken'))
robot.config['SESSION_STORAGE'] = False
robot.config["APP_ID"] = os.environ.get('appid')
robot.config["APP_SECRET"] = os.environ.get('secret')
# @robot.handler 處理全部消息
@robot.handler
def hello(message):
return 'Hello World!'
if __name__ == "__main__":
# 讓服務器監聽在 0.0.0.0:80
robot.config['HOST'] = '0.0.0.0'
robot.config['PORT'] = 80
robot.run()
而且在本地安裝 werobot 相關依賴，完成以後，執行部署：
將下面的這個地址複製到公衆號後臺，開啓調用便可。
參考 Git：https://github.com/serverless-tencent/tencent-werobot
這裏須要注意的是，咱們必定要關掉 Session 或者將 Session 改爲雲數據庫，不能使用本地文件等，例如關閉 Session 配置：
robot.config['SESSION_STORAGE'] = False
文本類似度實現圖文檢索
有時候用戶不知道咱們發了什麼文章，也不清楚每一個文章具體內容，他可能只須要簡單的關鍵詞，來看一下這個公衆號是否有他想要的東西。
例如他搜索：如何上傳文件？或者搜索：如何開發Component？而圖文檢索功能就能夠快速把最相關的歷史文章推送給用戶，這將會是很方便的一件事情。效果圖以下：
經過簡單的問題描述，找到目標結果，這就是咱們作的文章搜索功能。固然，咱們還能夠把它拓展成「客服系統」，這是後話了。
回到正題，咱們在以前的代碼基礎上，新增兩個函數：
- 函數 1：索引創建函數
主要功能：經過觸發該函數，能夠將現有的公衆號數據進行整理，而且創建適當的索引文件，存儲到 COS 中。
# -*- coding: utf8 -*-
import os
import re
import json
import random
from snownlp import SnowNLP
from qcloud_cos_v5 import CosConfig
from qcloud_cos_v5 import CosS3Client
bucket = os.environ.get('bucket')
secret_id = os.environ.get('secret_id')
secret_key = os.environ.get('secret_key')
region = os.environ.get('region')
client = CosS3Client(CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key))
def main_handler(event, context):
response = client.get_object(
Bucket=bucket,
Key=event["key"],
)
response['Body'].get_stream_to_file('/tmp/output.txt')
with open('/tmp/output.txt') as f:
data = json.loads(f.read())
articlesIndex = []
articles = {}
tempContentList = [
"_", " ",
]
for eveItem in data:
for i in range(0, len(eveItem['content']['news_item'])):
content = eveItem['content']['news_item'][i]['content']
content = re.sub(r'<code(.*?)</code>', '_', content)
content = re.sub(r'<.*?>', '', content)
for eve in tempContentList:
content = content.replace(eve, "")
desc = "%s。%s。%s" % (
eveItem['content']['news_item'][i]['title'],
eveItem['content']['news_item'][i]['digest'],
"。".join(SnowNLP(content).summary(3))
)
tempKey = "".join(random.sample('zyxwvutsrqponmlkjihgfedcba', 5))
articlesIndex.append(
{
"media_id": tempKey,
"description": desc
}
)
articles[tempKey] = eveItem['content']['news_item'][i]
client.put_object(
Bucket=bucket,
Body=json.dumps(articlesIndex).encode("utf-8"),
Key=event['index_key'],
EnableMD5=False
)
client.put_object(
Bucket=bucket,
Body=json.dumps(articles).encode("utf-8"),
Key=event['key'],
EnableMD5=False
)
這一部分，可能定製化比較多一些。首先是 tempContentList 變量，這裏能夠寫上一些公衆號中可能出現但不重要的話，例如公衆號末尾的引導關注文案，這些文案通常不參與搜索，因此最好在創建索引的時候進行替換去除。而後咱們還經過上述代碼去掉了 code 標籤裏面的內容，由於代碼也會影響結果。同時我還去掉了 html 標籤。
原始的文件大概是這樣的：
處理好的文件（經過標題+描述+SnowNLP提取的摘要）：
這些文件將存儲到 COS 中。
這一部分的核心就是，正確讓咱們提取出來的 description 儘量準確地描述文章的內容。通常狀況下，標題就是文章的核心，可是標題可能有一些信息丟失。
例如文章「用騰訊雲 Serverless 你要知道他們兩個的區別」，但實際上描述的是 Plugin 和 Component 的區別。雖然標題知道是兩個東西，可是卻缺乏了核心的目標，因此再加上咱們下面的描述：什麼是 Serverless Framework Plugin？什麼是Component？Plugin與Component 有什麼區別？想要入門 Serverless CLI，這兩個產品必須分的清楚，本文將會分享這兩者區別與對應的特色、功能。
固然，加上描述以後內容變得已經至關精確，可是正文中，可能有更加精準的描述或者額外的內容，因此採用的是標題+描述+摘要（textRank 提取出來的前三句，屬於提取式文本）。
- 函數 2: 搜索函數
主要功能：當用戶向微信號發送了指定關鍵詞，經過該函數獲取的結果。
思考：函數 1 和函數 2，均可以集成在以前的函數中，爲何要把這兩個函數單獨拿出來作一個獨立的函數存在呢？將它們放在同一個函數中很差麼？
緣由是 —— 主函數觸發次數相對來講是最多的，並且這個函數自己不須要太多的資源配置（64M 就夠了），而函數 1 和函數 2，可能須要消耗更多的資源，若是三個函數合併放在一塊兒，可能函數的內存大小須要總體調大，知足三個函數需求。這樣可能會消耗更多資源，
例如：主函數觸發了 10 次（64M，每次 1S），函數 1 觸發了 2 次（512 M，每次 5S），函數 2 觸發了 4 次（384M，每次 3S）
若是將三個函數放在一塊兒，資源消耗是：
若是將其變成三個函數來執行，資源消耗是：
前者總計資源消耗 13308，後者 10432。調用次數越多，主函數的調用比例越大，因此節約的資源也就會越多。所以此處建議將資源消耗差距比較大的模塊，分紅不一樣函數進行部署。
import os
import json
import jieba
from qcloud_cos_v5 import CosConfig
from qcloud_cos_v5 import CosS3Client
from collections import defaultdict
from gensim import corpora, models, similarities
bucket = os.environ.get('bucket')
secret_id = os.environ.get('secret_id')
secret_key = os.environ.get('secret_key')
region = os.environ.get('region')
client = CosS3Client(CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key))
def main_handler(event, context):
response = client.get_object(
Bucket=bucket,
Key=event["key"],
)
response['Body'].get_stream_to_file('/tmp/output.txt')
with open('/tmp/output.txt') as f:
data = json.loads(f.read())
articles = []
articlesDict = {}
for eve in data:
articles.append(eve['description'])
articlesDict[eve['description']] = eve['media_id']
sentence = event["sentence"]
documents = []
for eve_sentence in articles:
tempData = " ".join(jieba.cut(eve_sentence))
documents.append(tempData)
texts = [[word for word in document.split()] for document in documents]
frequency = defaultdict(int)
for text in texts:
for word in text:
frequency[word] += 1
dictionary = corpora.Dictionary(texts)
new_xs = dictionary.doc2bow(jieba.cut(sentence))
corpus = [dictionary.doc2bow(text) for text in texts]
tfidf = models.TfidfModel(corpus)
featurenum = len(dictionary.token2id.keys())
sim = similarities.SparseMatrixSimilarity(
tfidf[corpus],
num_features=featurenum
)[tfidf[new_xs]]
answer_list = [(sim[i], articles[i]) for i in range(1, len(articles))]
answer_list.sort(key=lambda x: x[0], reverse=True)
result = []
print(answer_list)
for eve in answer_list:
if eve[0] > 0.10:
result.append(articlesDict[eve[1]])
if len(result) >= 8:
result = result[0:8]
return {"result": json.dumps(result)}
這一部分的代碼也是很簡單，主要是經過文本的類似度對每一個文本進行評分，而後按照評分從高到低進行排序，給定一個閾值（此處設定的閾值爲 0.1），輸出閾值以前的數據。
另外要注意，此處引用了兩個依賴是 jieba 和 gensim，這兩個依賴均可能涉及到二進制文件，因此強烈推薦在 CentOS 系統下進行打包。
接下來就是主函數中的調用，爲了實現上述功能，須要在主函數中新增方法：
- 獲取所有圖文消息
def getTheTotalOfAllMaterials():
'''
文檔地址：https://developers.weixin.qq.com/doc/offiaccount/Asset_Management/Get_the_total_of_all_materials.html
:return:
'''
accessToken = getAccessToken()
if not accessToken:
return "Get Access Token Error"
url = "https://api.weixin.qq.com/cgi-bin/material/get_materialcount?access_token=%s" % accessToken
responseAttr = urllib.request.urlopen(url=url)
return json.loads(responseAttr.read())
def getMaterialsList(listType, count):
'''
文檔地址：https://developers.weixin.qq.com/doc/offiaccount/Asset_Management/Get_materials_list.html
:return:
'''
accessToken = getAccessToken()
if not accessToken:
return "Get Access Token Error"
url = "https://api.weixin.qq.com/cgi-bin/material/batchget_material?access_token=%s" % accessToken
materialsList = []
for i in range(1, int(count / 20) + 2):
requestAttr = urllib.request.Request(url=url, data=json.dumps({
"type": listType,
"offset": 20 * (i - 1),
"count": 20
}).encode("utf-8"), headers={
"Content-Type": "application/json"
})
responseAttr = urllib.request.urlopen(requestAttr)
responseData = json.loads(responseAttr.read().decode("utf-8"))
materialsList = materialsList + responseData["item"]
return materialsList
能夠經過如下代碼調用：
rticlesList = getMaterialsList("news", getTheTotalOfAllMaterials()['news_count'])
- 將圖文消息存儲到 COS，而且經過函數的 Invoke 接口，實現函數間調用：
def saveNewsToCos():
global articlesList
articlesList = getMaterialsList("news", getTheTotalOfAllMaterials()['news_count'])
try:
cosClient.put_object(
Bucket=bucket,
Body=json.dumps(articlesList).encode("utf-8"),
Key=key,
EnableMD5=False
)
req = models.InvokeRequest()
params = '{"FunctionName":"Weixin_GoServerless_GetIndexFile", "ClientContext":"{\\"key\\": \\"%s\\", \\"index_key\\": \\"%s\\"}"}' % (
key, indexKey)
req.from_json_string(params)
resp = scfClient.Invoke(req)
resp.to_json_string()
response = cosClient.get_object(
Bucket=bucket,
Key=key,
)
response['Body'].get_stream_to_file('/tmp/content.json')
with open('/tmp/content.json') as f:
articlesList = json.loads(f.read())
return True
except Exception as e:
print(e)
return False
- 根據搜索反饋回來的 Key 實現文章內容的對應
def searchNews(sentence):
req = models.InvokeRequest()
params = '{"FunctionName":"Weixin_GoServerless_SearchNews", "ClientContext":"{\\"sentence\\": \\"%s\\", \\"key\\": \\"%s\\"}"}' % (
sentence, indexKey)
req.from_json_string(params)
resp = scfClient.Invoke(req)
print(json.loads(json.loads(resp.to_json_string())['Result']["RetMsg"]))
media_id = json.loads(json.loads(json.loads(resp.to_json_string())['Result']["RetMsg"])["result"])
return media_id if media_id else None
最後在 main_handler 中，增長使用邏輯：
邏輯很簡答，就是根據用戶發的消息，去查找對應的結果.拿到結果以後判斷結果個數，若是有 1 個類似內容，則返回一個圖文，若是有多個則返回帶有連接的文本。
另一個邏輯是創建索引，直接是經過 API 網關觸發便可。固然，若是怕不安全或者有須要的話，能夠增長權限堅決的參數：
額外優化：
在接口列表中，咱們能夠看到獲取 accessToken 的接口其實是有次數限制的，每次獲取有效期兩個小時。因此，咱們就要在函數中，對這部份內容作持久化。爲了這個小東西，弄一個 MySQL 並不划算，因此決定用 COS：
def getAccessToken():
'''
文檔地址：https://developers.weixin.qq.com/doc/offiaccount/Basic_Information/Get_access_token.html
正常返回：{"access_token":"ACCESS_TOKEN","expires_in":7200}
異常返回：{"errcode":40013,"errmsg":"invalid appid"}
:return:
'''
global accessToken
# 第一次判斷是判斷本地是否已經有了accessToken，考慮到容器複用狀況
if accessToken:
if int(time.time()) - int(accessToken["time"]) <= 7000:
return accessToken["access_token"]
# 若是本地沒有accessToken，能夠去cos獲取
try:
response = cosClient.get_object(
Bucket=bucket,
Key=accessTokenKey,
)
response['Body'].get_stream_to_file('/tmp/token.json')
with open('/tmp/token.json') as f:
accessToken = json.loads(f.read())
except:
pass
# 這一次是看cos中是否有，若是cos中有的話，再次進行判斷段
if accessToken:
if int(time.time()) - int(accessToken["time"]) <= 7000:
return accessToken["access_token"]
# 若是此時流程還沒中止，則說明accessToken還沒得到到，就須要從接口得到，而且同步給cos
url = "https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=%s&secret=%s" % (appid, secret)
accessTokenResult = json.loads(urllib.request.urlopen(url).read().decode("utf-8"))
accessToken = {"time": int(time.time()), "access_token": accessTokenResult["access_token"]}
print(accessToken)
response = cosClient.put_object(
Bucket=bucket,
Body=json.dumps(accessToken).encode("utf-8"),
Key=accessTokenKey,
EnableMD5=False
)
return None if "errcode" in accessToken else accessToken["access_token"]
固然這段代碼能夠繼續優化，這裏只是一個思路。
除了利用文本類似度進行圖文檢索，咱們還能夠利用雲廠商提供的 AI 能力，爲公衆號增長機器人功能，限於篇幅就先告一段落，感興趣的讀者能夠本身探索一下。
總結
至此，咱們完成了一個簡單的公衆號開發。經過 Serverless 的原生開發思路（也可使用 Werobot 等公衆號開發框架），將公衆號後臺服務部署到 Serverless 架構上。再經過天然語言處理技術（特指文本類似度等）實現了一個圖文檢索功能。
Serverless 架構在開發微信公衆號這種事件驅動類的觸發式場景有着很大的優點，本文也僅僅是一個小探索，更多的功能和應用，能力和價值，仍是要看具體業務。但願讀者能夠經過本文，對 Serverless 架構有更深刻的瞭解。
Serverless Framework 30 天試用計劃
咱們誠邀您來體驗最便捷的 Serverless 開發和部署方式。在試用期內，相關聯的產品及服務均提供免費資源和專業的技術支持，幫助您的業務快速、便捷地實現 Serverless！
One More Thing
3 秒你能作什麼？喝一口水，看一封郵件，仍是 —— 部署一個完整的 Serverless 應用？
複製連接至 PC 瀏覽器訪問：https://serverless.cloud.tencent.com/deploy/express
3 秒極速部署，當即體驗史上最快的 Serverless HTTP 實戰開發！
傳送門：
- GitHub: github.com/serverless
- 官網：serverless.com
歡迎訪問：Serverless 中文網，您能夠在 最佳實踐 裏體驗更多關於 Serverless 應用的開發！
相關文章
- 1. 微信公衆號服務器搭建
- 2. 微信公衆號 智能客服
- 3. 技術小白用python搭建微信公衆號後臺
- 4. 使用python django快速搭建微信公衆號後臺
- 5. 記一次微信公衆號的開發與後臺搭建
- 6. 使用Senparc.Weixin SDK搭建微信公衆號服務程序
- 7. 使用openResty搭建微信公衆號服務
- 8. 用Docker搭建微信公衆號
- 9. 微信公衆號開發（服務器搭建、微信測試號的接入）
- 10. 搭建微信公衆號後臺開發環境
- 更多相關文章...
- • Git 服務器搭建 - Git 教程
- • 服務器上的 XML - XML 教程
- • Spring Cloud 微服務實戰(三) - 服務註冊與發現
- • 適用於PHP初學者的學習線路和建議
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
