給你的公眾號添加一個智能機器人

作者：KeepInUp 2020-09-15 13:56:08

開發開發工具

我們代碼使用 Python 開發，所以需要配置好 Python 開發環境，安裝 Flash 庫。最后需要注冊一個圖靈機器人賬號，調用其 API 接口。

環境和工具

公眾號一個
云服務器一臺
Python 環境
Flask(Python 第三方庫)
圖靈機器人賬號

以上就是我們實現公眾號后臺智能對話機器人需要的環境和工具，前兩個就不多說了。我們代碼使用 Python 開發，所以需要配置好 Python 開發環境，安裝 Flash 庫。最后需要注冊一個圖靈機器人賬號，調用其 API 接口。

[[342172]]

歡迎到我的公眾號碼小白TM 里調戲機器人，去看看這個機器人智不智能!

Web 服務器搭建

我們想要接收公眾號后臺發送的消息，就需要搭建一個 Web 服務器了。我們在云服務器上安裝好 Python 和 Flask 后，就可以進行 Web 服務器的搭建了。

我們首先搭建一個非常簡單的 Web 服務器，就是在網頁顯示出 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 服務器：

python main.py

運行成功后，我們可以在云服務器機器瀏覽器上訪問 127.0.0.1，如果我們能看到 Hello World! 就說明我們服務器啟動成功了。我們也可以在外網機器瀏覽器上訪問你云服務器的外網 IP，來檢測 Web 服務器是否成功啟動。

公眾號后臺配置和驗證

然后我們去公眾號后臺開發->基本配置頁找到服務器配置，可以看到我們需要一個服務器的 URL 地址、Token 令牌、消息加解密密鑰。

公眾號后臺配置

服務器的 URL 地址就是：http://服務器外網IP/wechat，/ 前面是你的云服務外網 IP 地址，后邊是我們在代碼里定義的路由入口(可以自己定義)，后邊我們會在代碼中看到。Token 令牌是我們自己定義的，后邊代碼中也會用到。消息加解密密鑰可以通過自動生成，消息加解密方式我們選擇明文模式即可。

你第一次配置的時候下邊會有保存按鈕，這個時候我們先不要點擊，因為我們需要在 Web 服務器端對消息字段進行解析處理，然后回傳結果進行驗證。

我們到微信公眾號開發文檔里看一下消息驗證流程。

驗證流程

根據上圖流程我們進行我們的代碼編寫。

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，不可填寫其他。這里坑了我很久，因為我的服務器上還有我自己的博客，80端口被博客占用了，為了把80端口讓出來，我重新搭了我的博客，修改了端口，導致現在訪問我的博客后邊必須加上修改后的端口。這里如果有同學有更高的方法，還請勞煩告知一下!

好了，現在我們的驗證流程完成，運行我們的 Web 服務器。然后到公眾號后臺配置處點擊保存，如果提示驗證通過，那么恭喜你完成了驗證。

如果提示驗證失敗，我們自己根據報錯提示來查看是哪里的問題。

實現“你問我答”和“圖”上往來

我們驗證成功了，下面就要開始處理粉絲在公眾號發過來的消息了。我們先來實現一個簡單的“你問我答”，粉絲給公眾號一條文本消息，公眾號立馬回復一條相同文本消息給粉絲;還有圖上往來，接受粉絲發送的圖片消息，并立馬回復相同的圖片給粉絲。

我們通過公眾號文檔知道，普通用戶向公眾號發消息是用的 POST 方法，消息分為文本消息、圖片消息、語音消息、視頻消息等很多種。這里我們只對文本消息和圖片消息進行處理和回復(后續你也可以針對其他消息進行處理和回復)。

消息的格式為XML數據包，下面看一下文本消息的實例：

<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 開發者微信號
FromUserName 發送方帳號(一個OpenID)
CreateTime 消息創建時間 (整型)
MsgType 消息類型，文本為text，圖片為image
Content 文本消息內容
MsgId 消息id，64位整型

被動回復文本消息，就是我們回復給用戶的文本消息類型為：

<xml> 
 <ToUserName><![CDATA[粉絲號]]></ToUserName> 
 <FromUserName><![CDATA[公眾號]]></FromUserName> 
 <CreateTime>1460541339</CreateTime> 
 <MsgType><![CDATA[text]]></MsgType> 
 <Content><![CDATA[test]]></Content> 
</xml>

Content 回復的消息內容

回復的圖片消息類型：

<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。

接收和回復消息的格式我們知道了，下面我們來看一下整個流程。

我們根據流程來修改一下我們的代碼。我們增加兩個文件來處理接收的消息，和回復的消息 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 服務器，去公眾號后臺發消息發圖片測試，如果成功的話立馬就會回復你相同的文字和圖片。

微信提供了一個在線測試的平臺，可以很方便的進行開發中的各種測試。

接入圖靈機器人

首先我們去圖靈機器人官網注冊一個賬號。然后在后臺創建一個機器人。

然后我們根據圖靈機器人接入文檔的使用說明：

編碼方式：調用調用圖靈API的各個環節的編碼方式均為 UTF-8
接口地址：http://openapi.tuling123.com/openapi/api/v2
請求方式： HTTP POST
請求參數：參數格式為 json

請求參數示例：

{ 
 "reqType":0, 
    "perception": { 
        "inputText": { 
            "text": "附近的酒店" 
        },        "inputImage": { 
            "url": "imageUrl" 
        },        "selfInfo": { 
            "location": { 
                "city": "北京", 
                "province": "北京", 
                "street": "信息路" 
            }        }    },    "userInfo": { 
        "apiKey": "", 
        "userId": "" 
    }}

輸出參數示例：

{ 
    "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": "親，已幫你找到相關酒店信息" 
            }        }    ]}

其中 apiKey 是可以在我們創建的機器人的參數中找到，userId 是用戶唯一標識。

好了，下面來編寫我們的代碼。我們增加一個 tuling.py 文件來接入圖靈接口。

tuling.py

import json 
import urllibapiKey = '從你創建的機器人獲得' 
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'] 
  # 免費版每天有固定次數，如果超過之后會返回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端口

耶，我們的機器人完成了，馬上迫不及待的去試試。

可以愉快的和你的機器人對話了。

后面發現了第一關注公眾號后的歡迎語沒有了，因為你自己的服務器接管了公眾號的消息，所以原來后臺設置的歡迎語就失效了。在公眾號文檔中看到關注/取消關注屬于事件。其消息的格式如下。

<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，然后回復我們的歡迎語就可以了，很簡單，這個我們就不詳細展開了。

責任編輯：未麗燕來源：今日頭條

公眾號機器人圖靈機器人