給你的公眾號添加一個智能機器人
環(huán)境和工具
- 公眾號一個
- 云服務(wù)器一臺
- Python 環(huán)境
- Flask(Python 第三方庫)
- 圖靈機器人賬號
以上就是我們實現(xiàn)公眾號后臺智能對話機器人需要的環(huán)境和工具,前兩個就不多說了。我們代碼使用 Python 開發(fā),所以需要配置好 Python 開發(fā)環(huán)境,安裝 Flash 庫。最后需要注冊一個圖靈機器人賬號,調(diào)用其 API 接口。
歡迎到我的公眾號 碼小白TM 里調(diào)戲機器人,去看看這個機器人智不智能!
Web 服務(wù)器搭建
我們想要接收公眾號后臺發(fā)送的消息,就需要搭建一個 Web 服務(wù)器了。我們在云服務(wù)器上安裝好 Python 和 Flask 后,就可以進行 Web 服務(wù)器的搭建了。
我們首先搭建一個非常簡單的 Web 服務(wù)器,就是在網(wǎng)頁顯示出 HelloWorld!,來跑通我們的代碼。我們來看一下主程序代碼。
main.py
- from flask import Flask
- app = Flask(__name__)@app.route("/")
- def index():
- return "Hello World!"
- if __name__ == "__main__":
- app.run(host='0.0.0.0')
代碼非常簡單,我們直接運行代碼,啟動 Web 服務(wù)器:
- python main.py
運行成功后,我們可以在云服務(wù)器機器瀏覽器上訪問 127.0.0.1,如果我們能看到 Hello World! 就說明我們服務(wù)器啟動成功了。我們也可以在外網(wǎng)機器瀏覽器上訪問你云服務(wù)器的外網(wǎng) IP,來檢測 Web 服務(wù)器是否成功啟動。
公眾號后臺配置和驗證
然后我們?nèi)ス娞柡笈_開發(fā)->基本配置頁找到服務(wù)器配置,可以看到我們需要一個服務(wù)器的 URL 地址、Token 令牌、消息加解密密鑰。

公眾號后臺配置
服務(wù)器的 URL 地址就是:http://服務(wù)器外網(wǎng)IP/wechat,/ 前面是你的云服務(wù)外網(wǎng) IP 地址,后邊是我們在代碼里定義的路由入口(可以自己定義),后邊我們會在代碼中看到。Token 令牌是我們自己定義的,后邊代碼中也會用到。消息加解密密鑰可以通過自動生成,消息加解密方式我們選擇明文模式即可。
你第一次配置的時候下邊會有保存按鈕,這個時候我們先不要點擊,因為我們需要在 Web 服務(wù)器端對消息字段進行解析處理,然后回傳結(jié)果進行驗證。
我們到微信公眾號開發(fā)文檔里看一下消息驗證流程。

驗證流程
根據(jù)上圖流程我們進行我們的代碼編寫。
main.py
- from flask import Flask
- from flask import request
- import hashlibapp = Flask(__name__)@app.route("/")
- def index(): return "Hello World!"
- # 公眾號后臺消息路由入口@app.route("/wechat", methods=["GET", "POST"])
- def wechat(): # 驗證使用的是GET方法 if request.method == "GET":
- signature = request.args.get('signature')
- timestamp = request.args.get('timestamp')
- nonce = request.args.get('nonce')
- echostr = request.args.get('echostr')
- token = "公眾號后臺填寫的token"
- # 進行排序 dataList = [token, timestamp, nonce] dataList.sort() result = "".join(dataList)
- #哈希加密算法得到hashcode sha1 = hashlib.sha1() sha1.update(result.encode("utf-8"))
- hashcode = sha1.hexdigest() if hashcode == signature:
- return echostr
- else:
- return ""
- if __name__ == "__main__":
- app.run(host='0.0.0.0', port=80) #公眾號后臺只開放了80端口
這里需要注意 http 的端口號固定使用80,不可填寫其他。這里坑了我很久,因為我的服務(wù)器上還有我自己的博客,80端口被博客占用了,為了把80端口讓出來,我重新搭了我的博客,修改了端口,導致現(xiàn)在訪問我的博客后邊必須加上修改后的端口。這里如果有同學有更高的方法,還請勞煩告知一下!
好了,現(xiàn)在我們的驗證流程完成,運行我們的 Web 服務(wù)器。然后到公眾號后臺配置處點擊保存,如果提示驗證通過,那么恭喜你完成了驗證。
如果提示驗證失敗,我們自己根據(jù)報錯提示來查看是哪里的問題。
實現(xiàn)“你問我答”和“圖”上往來
我們驗證成功了,下面就要開始處理粉絲在公眾號發(fā)過來的消息了。我們先來實現(xiàn)一個簡單的“你問我答”, 粉絲給公眾號一條文本消息,公眾號立馬回復一條相同文本消息給粉絲;還有圖上往來,接受粉絲發(fā)送的圖片消息,并立馬回復相同的圖片給粉絲。
我們通過公眾號文檔知道,普通用戶向公眾號發(fā)消息是用的 POST 方法,消息分為文本消息、圖片消息、語音消息、視頻消息等很多種。這里我們只對文本消息和圖片消息進行處理和回復(后續(xù)你也可以針對其他消息進行處理和回復)。
消息的格式為XML數(shù)據(jù)包,下面看一下文本消息的實例:
- <xml>
- <ToUserName><![CDATA[toUser]]></ToUserName>
- <FromUserName><![CDATA[fromUser]]></FromUserName>
- <CreateTime>1348831860</CreateTime>
- <MsgType><![CDATA[text]]></MsgType>
- <Content><![CDATA[this is a test]]></Content>
- <MsgId>1234567890123456</MsgId>
- </xml>
- ToUserName 開發(fā)者微信號
- FromUserName 發(fā)送方帳號(一個OpenID)
- CreateTime 消息創(chuàng)建時間 (整型)
- MsgType 消息類型,文本為text,圖片為image
- Content 文本消息內(nèi)容
- MsgId 消息id,64位整型
被動回復文本消息,就是我們回復給用戶的文本消息類型為:
- <xml>
- <ToUserName><![CDATA[粉絲號]]></ToUserName>
- <FromUserName><![CDATA[公眾號]]></FromUserName>
- <CreateTime>1460541339</CreateTime>
- <MsgType><![CDATA[text]]></MsgType>
- <Content><![CDATA[test]]></Content>
- </xml>
Content 回復的消息內(nèi)容
回復的圖片消息類型:
- <xml>
- <ToUserName><![CDATA[toUser]]></ToUserName>
- <FromUserName><![CDATA[fromUser]]></FromUserName>
- <CreateTime>12345678</CreateTime>
- <MsgType><![CDATA[image]]></MsgType>
- <Image>
- <MediaId><![CDATA[media_id]]></MediaId>
- </Image>
- </xml>
MediaId 通過素材管理中的接口上傳多媒體文件,得到的id。
接收和回復消息的格式我們知道了,下面我們來看一下整個流程。


我們根據(jù)流程來修改一下我們的代碼。我們增加兩個文件來處理接收的消息,和回復的消息 receive.py,reply.py。
receive.py
- import xml.etree.ElementTree as ET
- def parse_xml(web_data): if len(web_data) == 0:
- return None
- xmlData = ET.fromstring(web_data) msg_type = xmlData.find('MsgType').text
- if msg_type == 'text':
- return TextMsg(xmlData)
- elif msg_type == 'image':
- return ImageMsg(xmlData)
- class Msg(object):
- def __init__(self, xmlData):
- self.ToUserName = xmlData.find('ToUserName').text
- self.FromUserName = xmlData.find('FromUserName').text
- self.CreateTime = xmlData.find('CreateTime').text
- self.MsgType = xmlData.find('MsgType').text
- self.MsgId = xmlData.find('MsgId').text
- class TextMsg(Msg): def __init__(self, xmlData):
- Msg.__init__(self, xmlData) self.Content = xmlData.find('Content').text
- class ImageMsg(Msg): def __init__(self, xmlData):
- Msg.__init__(self, xmlData) self.PicUrl = xmlData.find('PicUrl').text
- self.MediaId = xmlData.find('MediaId').text
reply.py
- import time
- class Msg(object): def __init__(self): pass def send(self): return "success"class TextMsg(Msg): def __init__(self, toUserName, fromUserName, content): self.__dict = dict() self.__dict['ToUserName'] = toUserName self.__dict['FromUserName'] = fromUserName self.__dict['CreateTime'] = int(time.time()) self.__dict['Content'] = content def send(self): XmlForm = """ <xml>
- <ToUserName><![CDATA[{ToUserName}]]></ToUserName>
- <FromUserName><![CDATA[{FromUserName}]]></FromUserName>
- <CreateTime>{CreateTime}</CreateTime>
- <MsgType><![CDATA[text]]></MsgType>
- <Content><![CDATA[{Content}]]></Content>
- </xml>
- """ return XmlForm.format(**self.__dict)class ImageMsg(Msg): def __init__(self, toUserName, fromUserName, mediaId): self.__dict = dict() self.__dict['ToUserName'] = toUserName self.__dict['FromUserName'] = fromUserName self.__dict['CreateTime'] = int(time.time()) self.__dict['MediaId'] = mediaId def send(self): XmlForm = """ <xml>
- <ToUserName><![CDATA[{ToUserName}]]></ToUserName>
- <FromUserName><![CDATA[{FromUserName}]]></FromUserName>
- <CreateTime>{CreateTime}</CreateTime>
- <MsgType><![CDATA[image]]></MsgType>
- <Image>
- <MediaId><![CDATA[{MediaId}]]></MediaId>
- </Image>
- </xml>
- """ return XmlForm.format(**self.__dict)
main.py
- from flask import Flask
- from flask import request
- import hashlibimport receiveimport replyapp = Flask(__name__)@app.route("/")
- def index(): return "Hello World!"
- # 公眾號后臺消息路由入口@app.route("/wechat", methods=["GET", "POST"])
- def wechat(): # 驗證使用的GET方法 if request.method == "GET":
- signature = request.args.get('signature')
- timestamp = request.args.get('timestamp')
- nonce = request.args.get('nonce')
- echostr = request.args.get('echostr')
- token = "公眾號后臺填寫的token"
- # 進行排序 dataList = [token, timestamp, nonce] dataList.sort() result = "".join(dataList)
- #哈希加密算法得到hashcode sha1 = hashlib.sha1() sha1.update(result.encode("utf-8"))
- hashcode = sha1.hexdigest() if hashcode == signature:
- return echostr
- else:
- return ""
- else:
- recMsg = receive.parse_xml(request.data)
- if isinstance(recMsg, receive.Msg):
- toUser = recMsg.FromUserName fromUser = recMsg.ToUserName if recMsg.MsgType == 'text':
- content = recMsg.Content replyMsg = reply.TextMsg(toUser, fromUser, content) return replyMsg.send()
- elif recMsg.MsgType == 'image':
- mediaId = recMsg.MediaId replyMsg = reply.ImageMsg(toUser, fromUser, mediaId) return replyMsg.send()
- else:
- return reply.Msg().send()
- else:
- return reply.Msg().send()
- if __name__ == "__main__":
- app.run(host='0.0.0.0', port=80) #公眾號后臺只開放了80端口
然后我們啟動 Web 服務(wù)器,去公眾號后臺發(fā)消息發(fā)圖片測試,如果成功的話立馬就會回復你相同的文字和圖片。
微信提供了一個在線測試的平臺,可以很方便的進行開發(fā)中的各種測試。
接入圖靈機器人
首先我們?nèi)D靈機器人官網(wǎng)注冊一個賬號。然后在后臺創(chuàng)建一個機器人。
然后我們根據(jù)圖靈機器人接入文檔的使用說明:
- 編碼方式:調(diào)用調(diào)用圖靈API的各個環(huán)節(jié)的編碼方式均為 UTF-8
- 接口地址:http://openapi.tuling123.com/openapi/api/v2
- 請求方式: HTTP POST
- 請求參數(shù):參數(shù)格式為 json
請求參數(shù)示例:
- {
- "reqType":0,
- "perception": {
- "inputText": {
- "text": "附近的酒店"
- }, "inputImage": {
- "url": "imageUrl"
- }, "selfInfo": {
- "location": {
- "city": "北京",
- "province": "北京",
- "street": "信息路"
- } } }, "userInfo": {
- "apiKey": "",
- "userId": ""
- }}
輸出參數(shù)示例:
- {
- "intent": {
- "code": 10005,
- "intentName": "",
- "actionName": "",
- "parameters": {
- "nearby_place": "酒店"
- } }, "results": [
- { "groupType": 1,
- "resultType": "url",
- "values": {
- "url": "http://m.elong.com/hotel/0101/nlist/#indate=2016-12-10&outdate=2016-12-11&keywords=%E4%BF%A1%E6%81%AF%E8%B7%AF"
- } }, { "groupType": 1,
- "resultType": "text",
- "values": {
- "text": "親,已幫你找到相關(guān)酒店信息"
- } } ]}
其中 apiKey 是可以在我們創(chuàng)建的機器人的參數(shù)中找到,userId 是用戶唯一標識。
好了,下面來編寫我們的代碼。我們增加一個 tuling.py 文件來接入圖靈接口。
tuling.py
- import json
- import urllibapiKey = '從你創(chuàng)建的機器人獲得'
- tulingUrl = "http://openapi.tuling123.com/openapi/api/v2"
- # content 是接收的消息,userId 是用戶唯一標識def tulingReply(content, userId): requestData = { "reqType": 0,
- "perception": {
- "inputText": {
- "text": content
- }, "selfInfo": {
- "location": {
- "city": "北京"
- } } }, "userInfo": {
- "apiKey": apiKey,
- "userId": userId
- } } requestData = json.dumps(requestData).encode('utf8')
- http_post = urllib.request.Request( tulingUrl, data=requestData,
- headers={'content-type': 'application/json'})
- response = urllib.request.urlopen(http_post) response_str = response.read().decode('utf8')
- response_dic = json.loads(response_str) results_code = response_dic['intent']['code']
- # 免費版每天有固定次數(shù),如果超過之后會返回4003錯誤碼
- if results_code == 4003:
- results_text = "4003:%s" % response_dic['results'][0]['values']['text']
- else:
- results_text = response_dic['results'][0]['values']['text']
- return results_text
修改 main.py
- from flask import Flask
- from flask import request
- import hashlibimport reimport tulingimport receiveimport replyapp = Flask(__name__)@app.route("/")
- def index(): return "Hello World!"
- # 公眾號后臺消息路由入口@app.route("/wechat", methods=["GET", "POST"])
- def wechat(): # 驗證使用的GET方法 if request.method == "GET":
- signature = request.args.get('signature')
- timestamp = request.args.get('timestamp')
- nonce = request.args.get('nonce')
- echostr = request.args.get('echostr')
- token = "公眾號后臺填寫的token"
- # 進行排序 dataList = [token, timestamp, nonce] dataList.sort() result = "".join(dataList)
- #哈希加密算法得到hashcode sha1 = hashlib.sha1() sha1.update(result.encode("utf-8"))
- hashcode = sha1.hexdigest() if hashcode == signature:
- return echostr
- else:
- return ""
- else:
- recMsg = receive.parse_xml(request.data)
- if isinstance(recMsg, receive.Msg):
- toUser = recMsg.FromUserName fromUser = recMsg.ToUserName if recMsg.MsgType == 'text':
- content = recMsg.Content # userId 長度小于等于32位
- if len(toUser) > 31:
- userid = str(toUser[0:30])
- else:
- userid = str(toUser) userid = re.sub(r'[^A-Za-z0-9]+', '', userid)
- tulingReplay = tuling.tulingReply(content, userid) replyMsg = reply.TextMsg(toUser, fromUser, tulingReplay) return replyMsg.send()
- elif recMsg.MsgType == 'image':
- mediaId = recMsg.MediaId replyMsg = reply.ImageMsg(toUser, fromUser, mediaId) return replyMsg.send()
- else:
- return reply.Msg().send()
- else:
- return reply.Msg().send()
- if __name__ == "__main__":
- app.run(host='0.0.0.0', port=80) #公眾號后臺只開放了80端口
耶,我們的機器人完成了,馬上迫不及待的去試試。

可以愉快的和你的機器人對話了。
后面發(fā)現(xiàn)了第一關(guān)注公眾號后的歡迎語沒有了,因為你自己的服務(wù)器接管了公眾號的消息,所以原來后臺設(shè)置的歡迎語就失效了。在公眾號文檔中看到關(guān)注/取消關(guān)注屬于事件。其消息的格式如下。
- <xml>
- <ToUserName><![CDATA[toUser]]></ToUserName>
- <FromUserName><![CDATA[FromUser]]></FromUserName>
- <CreateTime>123456789</CreateTime>
- <MsgType><![CDATA[event]]></MsgType>
- <Event><![CDATA[subscribe]]></Event>
- </xml>
我們只需要判斷一下 MsgType 是 event,然后 Event 是 subscribe,然后回復我們的歡迎語就可以了,很簡單,這個我們就不詳細展開了。