- 大家好，我是同学小张，日常分享AI知识和实战案例
- 欢迎 点赞 + 关注 👏，持续学习，持续干货输出。
- 一起交流💬，一起进步💪。
- 微信公众号也可搜【同学小张】 🙏
本站文章一览：
上篇文章我们将企业微信自建应用与我们自己的服务器连接起来了，下面进行第二步，接收用户消息处理，并回复。
类比公众号开发的这篇文章，打通数据链路：【超详细！】Python微信公众号开发（2）
还是采用Flask框架。
0. 接收与被动回复消息的格式
0.1 消息格式（以文本消息为例）
看下官方文档：https://developer.work.weixin.qq.com/document/path/96466
0.1.1 接收消息格式
0.1.2 回复消息格式
接收到的消息格式为xml格式，回复内容的格式也是xml格式。
这里接收到的xml格式的内容和发送出去的xml格式的内容都是需要加密的，所以在接收到之后和回复消息之前都应该进行解密和加密处理。
该功能的限制，与微信公众号一样，也是只有5s的响应时间，总共重试三次：
了解了基本内容之后，我们开始编写代码
1. 编写代码
1.1 入口的处理
@app.route('/company_wechat', methods=['GET', 'POST']) def company_wechat(): if request.method == 'GET': return signature(request, 0) else: print("收到请求......") return handle_user_message(request, 0)
如果是GET请求，就去校验是否是微信消息。
如果是POST请求，就进行相应处理并给出回复。
1.2 解密接收到的消息内容
上面提到了，企业微信服务转发过来的消息是经过加密的，需要我们自己解密。
原始发过来的消息示例如下：
解密代码：
user_message = request.data msg_signature = request.args.get('msg_signature', '') timestamp = request.args.get('timestamp', '') nonce = request.args.get('nonce', '') ret, sMsg = qy_api[i].DecryptMsg(user_message.decode('utf-8'), msg_signature, timestamp, nonce) decrypt_data = {} for node in list(fromstring(sMsg.decode('utf-8'))): decrypt_data[node.tag] = node.text print(decrypt_data)
解密完之后的消息内容：
1.3 对消息内容进行处理，并封装回复的消息体
得到明文的消息内容之后，我们就可以对其进行相应的处理，并给出回复了。
示例代码如下，如果用户发过来一个“我帅吗”，就给用户回复一个“帅得一逼”
# 处理文本消息 if decrypt_data.get('Content', '') == '我帅吗': sRespData = """<xml> <ToUserName>{to_username}</ToUserName> <FromUserName>{from_username}</FromUserName> <CreateTime>{create_time}</CreateTime> <MsgType>text</MsgType> <Content>{content}</Content> </xml> """.format(to_username=decrypt_data['ToUserName'], from_username=decrypt_data['FromUserName'], create_time=decrypt_data['CreateTime'], content="帅得一逼",)
最后，封装完之后不要忘了加密：
ret, send_msg = qy_api[i].EncryptMsg(sReplyMsg=sRespData, sNonce=nonce) if ret == 0: return send_msg else: print(send_msg)
1.4 运行效果
成功！
2. 完整代码
参考：https://gist.github.com/GentleCP/5d02f4e84b8c8905bcf67643223cd499
#!/usr/bin/env python # -*- coding: utf-8 -*- from flask import Flask, request, make_response from company_wechat_verify.WXBizMsgCrypt3 import WXBizMsgCrypt from xml.etree.ElementTree import fromstring app = Flask(__name__) import xml.etree.ElementTree as ET def printXML(xml_content): # 创建XML元素 element = ET.XML(xml_content) # 使用indent()函数进行格式化打印 ET.indent(element) print(ET.tostring(element, encoding='unicode')) qy_api = [ WXBizMsgCrypt("xxx", "xxx", "xxx"), ] #对应接受消息回调模式中的token，EncodingAESKey 和 企业信息中的企业id # 开启消息接受模式时验证接口连通性 def signature(request, i): msg_signature = request.args.get('msg_signature', '') timestamp = request.args.get('timestamp', '') nonce = request.args.get('nonce', '') echo_str = request.args.get('echostr', '') ret,sEchoStr=qy_api[i].VerifyURL(msg_signature, timestamp, nonce, echo_str) if (ret != 0): print("ERR: VerifyURL ret: " + str(ret)) return("failed") else: return(sEchoStr) def handle_user_message(request, i): ''' 接收用户消息，可进行被动响应 ''' user_message = request.data printXML(user_message) msg_signature = request.args.get('msg_signature', '') timestamp = request.args.get('timestamp', '') nonce = request.args.get('nonce', '') ret, sMsg = qy_api[i].DecryptMsg(user_message.decode('utf-8'), msg_signature, timestamp, nonce) decrypt_data = {} for node in list(fromstring(sMsg.decode('utf-8'))): decrypt_data[node.tag] = node.text # 解析后得到的decrypt_data: {"ToUserName":"企业号", "FromUserName":"发送者用户名", "CreateTime":"发送时间", "Content":"用户发送的内容", "MsgId":"唯一id，需要针对此id做出响应", "AagentID": "应用id"} # 用户应根据Content的内容自定义要做出的行为，包括响应返回数据，如下例子，如果发送的是123，就返回hello world print(decrypt_data) # 处理文本消息 if decrypt_data.get('Content', '') == '我帅吗': sRespData = """<xml> <ToUserName>{to_username}</ToUserName> <FromUserName>{from_username}</FromUserName> <CreateTime>{create_time}</CreateTime> <MsgType>text</MsgType> <Content>{content}</Content> </xml> """.format(to_username=decrypt_data['ToUserName'], from_username=decrypt_data['FromUserName'], create_time=decrypt_data['CreateTime'], content="帅得一逼",) ret, send_msg = qy_api[i].EncryptMsg(sReplyMsg=sRespData, sNonce=nonce) if ret == 0: return send_msg else: print(send_msg) @app.route('/company_wechat', methods=['GET', 'POST']) def company_wechat(): if request.method == 'GET': return signature(request, 0) else: print("收到请求......") return handle_user_message(request, 0) if __name__ == '__main__': app.run(host='0.0.0.0', port=80, debug=True)
3. 遇到的坑
其实也不能算是坑，应该算是惊吓。
刚开始没看具体发来的消息是什么，只是在收到请求的时候打了一行日志，结果… 发现一直在收到请求，着实吓到我了，不得不关了服务… 。
原本以为这是回复失败的尝试，但说明文档中说了只会重试3次，这里这么多请求，都是什么？
它其实是 Location 类型的消息，基本1s一次。
有这个消息的原因，是我在API接收消息是，勾选了 上报地理位置，并且在手机上，开启了位置权限。
如果觉得本文对你有帮助，麻烦点个赞和关注呗 ~~~
- 大家好，我是 同学小张，日常分享AI知识和实战案例
- 欢迎 点赞 + 关注 👏，持续学习，持续干货输出。
- 一起交流💬，一起进步💪。
- 微信公众号也可搜【同学小张】 🙏
本站文章一览：
