用python开发微信公众平台聊天机器人

最近用python写了一个聊天机器人的微信公众号，网上找的开发文档参差不齐，官方文档也比较老旧，还有部分小问题。于是，分享一下我的思路。

开发环境

windows sever 2008+python3.6+Flask

1.搭建服务器

服务器：随便租个什么服务器，或者用弄个能连外网的机子。
1.1 在服务器上装python3.6开发环境，个人喜欢装Anaconda，省得手动装各种包。
1.2 新建一个test.py文件，写如下代码：

from flask import Flask, request, make_responseapp = Flask(__name__)
app.debug = True@app.route('/')  # 默认网址
def index():return '测试页面'if __name__ == '__main__':app.run(host="0.0.0.0", port=80,debug=True)

打开终端，cd到目录下，运行这个文件
python test.py
弹出 * Running on http://0.0.0.0:80/ (Press CTRL+C to quit) 说明成功了。
注意这里的端口必须是80或443，微信公众号只支持这两个端口。
1.3 在另一台机子上打开浏览器，输入服务器的ip地址，弹出‘测试页面’表示服务器搭建成功。
注意：使用阿里云服务器显示连接超时，需要配置安全组规则，具体方法百度。

2. 配置微信公众号

2.1 首先当然是申请一个公众号啦。注意：素材和菜单的开发权限必须要通过认证，未认证只能发发文字，具体可以去接口权限查看。
2.2 开发者基本配置

点击修改配置后，选择提交肯定是验证token失败，因为还需要完成代码逻辑。
2.3 改动 test.py 为app.py.代码如下：

# -*- coding: utf-8 -*-
#  filename: app.py
from flask import Flask, request, make_response
import hashlib
import handleapp = Flask(__name__)
app.debug = True@app.route('/')  # 默认网址
def index():return '测试页面'@app.route('/wx', methods=['GET', 'POST'])
def wechat_auth():  # 处理微信请求的处理函数，get方法用于认证，post方法取得微信转发的数据if request.method == 'GET':token = 'xxxx'  # 这里填公众号里设置的token。data = request.argssignature = data.get('signature', '')timestamp = data.get('timestamp', '')# print(timestamp)nonce = data.get('nonce', '')echostr = data.get('echostr', '')s = [timestamp, nonce, token]s.sort()s = ''.join(s)s = s.encode(encoding='utf-8')#加密后的token匹配上signatureif (hashlib.sha1(s).hexdigest() == signature):return make_response(echostr)else:return 'signature is error'else:rec = request.stream.read() # 接收消息#print(rec)dispatcher = handle.MsgHandler(rec)data = dispatcher.dispatch()with open("./debug.log", "a") as file:file.write(data)response = make_response(data)response.content_type = 'application/xml'return response #发送消息if __name__ == '__main__':app.run(host="0.0.0.0", port=80,debug=True)

这里只用到wechat_auth函数中的get请求部分，else后面是post请求，暂且不管，import handle是用来处理消息的，先不要看。
运行app.py python app.py
2.4 运行app.py成功后，点击公众号界面的提交按钮，token验证成功即可。

3. 消息回复流程

核心文件

app.py ：服务器主程序，接收和发送消息。
handle.py：消息处理枢纽。
receive.py：解析接收到的消息
reply.py：回复消息。（可以添加复杂的回复功能）

流程图

有了以上几个模块就可以完成简单的文字消息回复了。

4.access_token

官方文档：access_token是公众号的全局唯一接口调用凭据，公众号调用各接口时都需使用access_token。开发者需要进行妥善保存。access_token的存储至少要保留512个字符空间。access_token的有效期目前为2个小时，需定时刷新，重复获取将导致上次获取的access_token失效。
可以看到获取一次access_token的有效期是2小时，那么我们需要写一个定时程序，每2小时获取一次access_token。保存在文本文档access_token.txt中。
basic.py ：定时获取access_token

# -*- coding: utf-8 -*-
#  filename: basic.py
import urllib.request
import time
import jsonclass Basic():def __init__(self):self.__accessToken = ''def __real_get_access_token(self):appId = "xxxxx"  #配置公众号时得到的id和secretappSecret = "xxxxxx"postUrl = ("https://api.weixin.qq.com/cgi-bin/token?grant_type="               "client_credential&appid=%s&secret=%s" % (appId, appSecret))urlResp = urllib.request.urlopen(postUrl)urlResp = json.loads(urlResp.read())self.__accessToken = urlResp['access_token']def get_access_token(self):self.__real_get_access_token()return self.__accessTokenif __name__ == "__main__":basic = Basic()while True:token = basic.get_access_token()with open('access_token.txt','w',encoding='utf-8') as f:f.write(token)time.sleep(7000)

运行basic.py就能持续不断的获取access_token啦

5.临时素材添加与获取

如果你向公众号发送语音、图片、视频等非文本消息，app.py只会接收到一个这样的消息（消息模板）,例如图片消息：

<xml><ToUserName><![CDATA[接收人]]></ToUserName><FromUserName><![CDATA[发送人]]></FromUserName><CreateTime>创建时间</CreateTime><MsgType><![CDATA[image]]></MsgType><Image><MediaId><![CDATA[素材id]]></MediaId></Image></xml>

这样一段消息经过receive.py的解析后，我们拿到“素材id”，然后向微信服务器发送一个请求，才能下载到对应的素材。
同理，如果公众号向用户发送图片也需要先上传图片到微信服务器，服务器会返回一个“素材id”，然后公众号再发送一段类似上面的消息，填入返回的”素材id“就可以了，其它类型消息模板可以去微信公众号官方文档查看。
media.py：素材上传和下载。

# -*- coding: utf-8 -*-
# filename: media.py
import requests
import json
import osclass Media(object):def __init__(self):with open('access_token.txt','r',encoding='utf-8') as f:   #获取access_tokentoken = f.read()self.accessToken = tokendef get(self, mediaId, filepath):postUrl = "https://api.weixin.qq.com/cgi-bin/media/get?access_token=%s&media_id=%s" % (self.accessToken, mediaId)urlResp = requests.post(postUrl)headers = urlResp.headersif ('Content-Type: application/json\r\n' in headers) or ('Content-Type: text/plain\r\n' in headers):jsonDict = json.loads(urlResp.content)print(jsonDict)else:# print(urlResp.content.decode('ascii'))buffer = urlResp.content   #素材的二进制with open(filepath, "wb") as f:f.write(buffer)print("get successful")def _add(self, type,filepath):with open(filepath,"rb") as f:img = f.read()filename = os.path.split(filepath)[-1]url = 'https://api.weixin.qq.com/cgi-bin/media/upload?access_token={}&type={}'files = {'media':(filename,img)}res = requests.post(url.format(self.accessToken,type),files=files)return json.loads(res.content.decode('utf-8'))['media_id']

5.自定义菜单

自定义菜单能够帮助公众号丰富界面，让用户更好更快地理解公众号的功能。开启自定义菜单后，公众号界面如图所示：

menu.py ：自定义菜单增删改查

# -*- coding: utf-8 -*-
# filename: menu.py
import urllib.requestclass Menu(object):def __init__(self):passdef create(self, postData, accessToken):postUrl = "https://api.weixin.qq.com/cgi-bin/menu/create?access_token=%s" % accessToken# if isinstance(postData, 'unicode'):postData = postData.encode('utf-8')urlResp = urllib.request.urlopen(url=postUrl, data=postData)print(urlResp.read())def query(self, accessToken):postUrl = "https://api.weixin.qq.com/cgi-bin/menu/get?access_token=%s" % accessTokenurlResp = urllib.request.urlopen(url=postUrl)print(urlResp.read())def delete(self, accessToken):postUrl = "https://api.weixin.qq.com/cgi-bin/menu/delete?access_token=%s" % accessTokenurlResp = urllib.request.urlopen(url=postUrl)print(urlResp.read())#获取自定义菜单配置接口def get_current_selfmenu_info(self, accessToken):postUrl = "https://api.weixin.qq.com/cgi-bin/get_current_selfmenu_info?access_token=%s" % accessTokenurlResp = urllib.request.urlopen(url=postUrl)print(urlResp.read())if __name__ == '__main__':myMenu = Menu()postJson = """{"button":[{"name": "小迪互动","sub_button":[{"type": "click","name": "图片文字识别","key": "ocr"},{"type": "click","name": "语音聊天","key": "speech"},{"type": "click","name": "文字聊天","key": "text"}]},{"name": "精选栏目","sub_button":[{"type": "view","name": "往期文章","url": "https://mp.weixin.qq.com/mp/profile_ext?action=home&__biz=MzU4ODcxOTE2OA==&scene=124#wechat_redirect"}]},{"name": "关于我们","sub_button":[{"type": "click","name": "DepthsData","key": "depth"},{"type": "click","name": "RPA","key": "rpa"}]}]}"""with open('access_token.txt','r',encoding='utf-8') as f:accessToken = f.read()myMenu.create(postJson, accessToken)

运行menu.py一次性创建自定义菜单，有些许延迟。
创建成功后，点击菜单会向公众号发送一个“事件消息”，根据你自定义的click-key，判断该事件属于哪个菜单。
此外，新用户的订阅会发送一个“subscribe”的“事件消息”，可以利用它给新用户打招呼噢。

6. 总结

到此，一个功能还算齐全的公众号开发完毕。
1.补充完整的流程图：

图中红色字体的文件是需要运行的。
2.文件结构:

3.剩余代码
handle.py

# -*- coding: utf-8 -*-
# filename: handle.py
from receive import parse_xml
from reply import event_reply,text_reply,ocr_reply
from media import Media
from helpfunc import speech2text,text2speechclass MsgHandler(object):"""针对type不同，转交给不同的处理函数。直接处理即可"""def __init__(self, msg):self.msg = parse_xml(msg)self.time = int(time.time())self.temp_text = """<xml><ToUserName><![CDATA[{}]]></ToUserName><FromUserName><![CDATA[{}]]></FromUserName><CreateTime>{}</CreateTime><MsgType><![CDATA[text]]></MsgType><Content><![CDATA[{}]]></Content></xml>"""self.temp_image = """<xml><ToUserName><![CDATA[{}]]></ToUserName><FromUserName><![CDATA[{}]]></FromUserName><CreateTime>{}</CreateTime><MsgType><![CDATA[image]]></MsgType><Image><MediaId><![CDATA[{}]]></MediaId></Image></xml>"""self.temp_voice = """<xml><ToUserName><![CDATA[{}]]></ToUserName><FromUserName><![CDATA[{}]]></FromUserName><CreateTime>{}</CreateTime><MsgType><![CDATA[voice]]></MsgType><Voice><MediaId><![CDATA[{}]]></MediaId></Voice></xml>"""def dispatch(self):self.result = ""  # 统一的公众号出口数据if self.msg.MsgType == "text":self.result = self.textHandle()elif self.msg.MsgType== "voice":self.result = self.voiceHandle()elif self.msg.MsgType == 'image':self.result = self.imageHandle()elif self.msg.MsgType == 'video':self.result = self.videoHandle()elif self.msg.MsgType == 'shortvideo':self.result = self.shortVideoHandle()elif self.msg.MsgType == 'location':self.result = self.locationHandle()elif self.msg.MsgType == 'link':self.result = self.linkHandle()elif self.msg.MsgType == 'event':self.result = self.eventHandle()return self.resultdef textHandle(self):result = text_reply(self.msg.Content,self.msg.FromUserName)response = self.temp_text.format(self.msg.FromUserName, self.msg.ToUserName, self.time, result)return responsedef voiceHandle(self):# template = self.temp_voicemediaid = self.msg.MediaIdfilepath = r'voice/' + str(time.time()) + '.mp3'Media().get(mediaid, filepath)#下载素材text = speech2text(filepath)result = text_reply(text,self.msg.FromUserName)voice_path = text2speech(result)r_mediaid = Media()._add('voice', voice_path)#上传素材response = self.temp_voice.format(self.msg.FromUserName, self.msg.ToUserName, self.time, r_mediaid)return responsedef imageHandle(self):mediaid = self.msg.MediaIdstyle = get_state(self.msg.FromUserName)if style == 'img-ocr':filepath = r'pic/' + str(time.time()) + '.jpg'Media().get(mediaid, filepath)try:result = ocr_reply(imgpath=filepath)except:result = '图上好像没有字吧'respnse = self.temp_text.format(self.msg.FromUserName, self.msg.ToUserName, self.time, result)else:respnse = self.temp_image.format(self.msg.FromUserName, self.msg.ToUserName, self.time, mediaid)return respnsedef videoHandle(self):return self.temp_text.format(self.msg.FromUserName, self.msg.ToUserName, self.time, '不支持的消息类型！')def shortVideoHandle(self):return self.temp_text.format(self.msg.FromUserName, self.msg.ToUserName, self.time, '不支持的消息类型！')def locationHandle(self):return self.temp_text.format(self.msg.FromUserName, self.msg.ToUserName, self.time, '不支持的消息类型！')def linkHandle(self):return self.temp_text.format(self.msg.FromUserName, self.msg.ToUserName, self.time, '不支持的消息类型！')def eventHandle(self):result = event_reply(self.msg,self.msg.FromUserName)response = self.temp_text.format(self.msg.FromUserName, self.msg.ToUserName, self.time, result)return response

receive.py

# -*- coding: utf-8 -*-
# filename: receive.py
import xml.etree.ElementTree as ETdef parse_xml(web_data):if len(web_data) == 0:return NonexmlData = ET.fromstring(web_data)msg_type = xmlData.find('MsgType').textprint(msg_type)if msg_type == 'event':event_type = xmlData.find('Event').textif event_type == 'CLICK':return Click(xmlData)elif event_type in ('subscribe', 'unsubscribe'):return Subscribe(xmlData)#elif event_type == 'VIEW':#return View(xmlData)#elif event_type == 'LOCATION':#return LocationEvent(xmlData)#elif event_type == 'SCAN':#return Scan(xmlData)elif msg_type == 'text':return TextMsg(xmlData)elif msg_type == 'image':return ImageMsg(xmlData)elif msg_type == 'voice':return VoiceMsg(xmlData)else:return Msg(xmlData)class Msg(object):def __init__(self, xmlData):self.ToUserName = xmlData.find('ToUserName').textself.FromUserName = xmlData.find('FromUserName').textself.CreateTime = xmlData.find('CreateTime').textself.MsgType = xmlData.find('MsgType').textclass Click(Msg):def __init__(self, xmlData):Msg.__init__(self, xmlData)self.Event = xmlData.find('Event').textself.Eventkey = xmlData.find('EventKey').textclass Subscribe(Msg):def __init__(self, xmlData):Msg.__init__(self, xmlData)self.Event = xmlData.find('Event').textclass TextMsg(Msg):def __init__(self, xmlData):Msg.__init__(self, xmlData)self.Content = xmlData.find('Content').textclass ImageMsg(Msg):def __init__(self, xmlData):Msg.__init__(self, xmlData)self.PicUrl = xmlData.find("PicUrl").textself.MediaId = xmlData.find("MediaId").textclass VoiceMsg(Msg):def __init__(self, xmlData):Msg.__init__(self, xmlData)self.MediaId = xmlData.find("MediaId").text

reply.py

放心吧，这个我是不会放出来的，毕竟我是一个“称职”的员工。

好了，以上就是用python开发微信公众平台的全部了，觉得好看点个赞再走吧！