【企业微信-消息推送】模板卡片消息-Python代码-Toy模板网

这篇具有很好参考价值的文章主要介绍了【企业微信-消息推送】模板卡片消息-Python代码。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

浏览消息推送的时候，在模板卡片部分停滞很久；其中主要涉及url的回调工作，不太熟悉。官网链接-模板卡片
官方特殊说明：

特殊说明

仅有按钮交互型、投票选择型、多项选择型以及填写了action_menu字段的文本通知型、图文展示型的卡片支持回调更新或通过接口更新卡片。

支持回调更新的卡片可支持用户点击触发交互事件，需要开发者设置的回调接口来处理回调事件，回调协议可见文档模板卡片事件推送，注意没有配置回调接口的应用不可发送支持回调的卡片。

开发者的服务收到回调事件后，需要根据协议返回相应的数据以更新卡片，对应的协议见文档更新模版卡片消息。

此接口发送支持回调更新的卡片消息之后，返回的参数里会带上response_code，可使用response_code调用更新模版卡片消息接口，response_code 24小时内有效，且只能调用一次接口。

目前解决第二条，配置回调接口支持发送模板卡片

配置回调接口

官网链接-回调服务
通过回调服务可以自定义消息，如识别关键词，动态获取消息的状态等。

在使用代码开发前，首先需要在相应的应用中配置回调接口服务，定义URL, Token, EncodingAESKey。

这里的URL就是之后访问的回调地址

步骤：

打开企业微信的管理者后台，进入应用管理，选择消息推送的目标程序
选择下面收发消息，设置API接收
进行配置

但是当点击保存的时候，需要进行回调验证；设置的URL接口需要支持GET、POST两种请求服务，GET用于进行验证，POST用来之后收发消息（提供给其他需要回调接口的服务）
因此，需要先定义一套接口满足GET，才能保存配置；这里可以参照官网提供的第三方库完成验证，其中包括一系列加密算法可以参考官网链接-加密参数说明

回调接口的GET请求验证

首先下载第三方加密库加密库链接；选择合适的语言去github进行下载

配置django项目，定义回调接口：http://ip:port/jyg/vertifyinfo，企业管理员在保存回调配置信息时，企业微信会发送一条验证消息到填写的URL，即在给定的url后面添加timestamp, nonce, echostr, msg_signature用于验证

def versityInfo(request):
	sToken = "xxx"
	sEncodingAESKey = "xxx"   # 刚才配置的，写在这里，可以将配置信息单独抽出
	sCorpID = "xxx"  # corpID
	if request.method == 'GET':  # GEt为验证
		wxcpt=WXBizMsgCrypt(sToken,sEncodingAESKey,sCorpID)
		# 提取请求中的参数
		sVerifyMsgSig = request.GET.get("msg_signature")
		sVerifyTimeStamp=request.GET.get("timestamp")
		sVerifyNonce=request.GET.get("nonce")
		sVerifyEchoStr=request.GET.get("echostr")
		# 调用第三方库进行验证
		ret,sEchoStr=wxcpt.VerifyURL(sVerifyMsgSig, sVerifyTimeStamp,sVerifyNonce,sVerifyEchoStr)
		if(ret!=0):
			print("ERR: VerifyURL ret: " + str(ret))
			sys.exit(1)
	# 返回验证解密后的消息
	return HttpResponse(sEchoStr)

例如VerifyURL为第三方库，里面一共提供了三个方法，详细可以浏览官网。

代码完成可以使用接口测试工具进行测试：官网链接-接口测试工具
【企业微信-消息推送】模板卡片消息-Python代码
然后返回点击保存即可完成配置。

发送模板卡片消息

请求方式：GET（HTTPS）
请求地址：https://qyapi.weixin.qq.com/cgi-bin/externalcontact/get_follow_user_list?access_token=ACCESS_TOKEN
请求参数：

{
    "touser" : "JiuXiaTian",
    "msgtype" : "template_card",
    "agentid" : 1000033,
    "template_card" : {
        "card_type" : "text_notice",
        "source" : {
            "desc": "企业微信",
            "desc_color": 1
        },
        "action_menu": {
            "desc": "卡片副交互辅助文本说明",
            "action_list": [
                {"text": "接受推送", "key": "A"},
                {"text": "不再推送", "key": "B"}
            ]
        },
        "task_id": "task_id",
        "main_title" : {
            "title" : "欢迎使用企业微信",
            "desc" : "您的好友正在邀请您加入企业微信"
        },
        "quote_area": {
            "type": 1,
            "url": "https://work.weixin.qq.com",
            "title": "企业微信的引用样式",
            "quote_text": "企业微信真好用呀真好用"
        },
        "emphasis_content": {
            "title": "100",
            "desc": "核心数据"
        },
        "sub_title_text" : "下载企业微信还能抢红包！",
        "horizontal_content_list" : [
            {
                "keyname": "邀请人",
                "value": "张三"
            },
            {
                "type": 1,
                "keyname": "企业微信官网",
                "value": "点击访问",
                "url": "https://work.weixin.qq.com"
            }
        ],
        "card_action": {
            "type": 1,
            "url": "https://work.weixin.qq.com"
        }
    },
    "enable_id_trans": 0,
    "enable_duplicate_check": 0,
    "duplicate_check_interval": 1800
}

运行结果：
【企业微信-消息推送】模板卡片消息-Python代码

被动响应消息

在企业微信中可以根据发送的消息定制回调消息，效果图如下：
【企业微信-消息推送】模板卡片消息-Python代码
被动接收消息走的回调接口的POST方法，官网调用接口如下：
请求方式：POST
请求地址：http://api.3dept.com/?msg_signature=ASDFQWEXZCVAQFASDFASDFSS&timestamp=13500001234&nonce=123412323
请求体：当用户提交消息后，会在请求体中封装成一个${xml}格式的数据

可以使用 if request.method 的方式进行路由分发，但是观察参数：GET验证中包含sVerifyEchoStr，而POST中存放的时候请求体数据，也可以根据这个点就行判断，如：

# 公共参数
sVerifyMsgSig = request.GET.get("msg_signature")
sVerifyTimeStamp = request.GET.get("timestamp")
sVerifyNonce = request.GET.get("nonce")
sVerifyEchoStr = request.GET.get("echostr")
if sVerifyEchoStr:
    pass
 else:  # 接收消息
 	pass

准备工作，定义一个自定义的消息体，去处理信息解密，加密等其他情况：

# 自定义消息处理
class MessageHandler:
    def __init__(self, baseData):
        self.wxcpt = baseData['wxcpt']
        self.sVerifyMsgSig = baseData['sVerifyMsgSig']
        self.sVerifyTimeStamp = baseData['sVerifyTimeStamp']
        self.sVerifyNonce = baseData['sVerifyNonce']

# 请求中处理初始化参数
# 数据初始化
sToken = URLRobackSetting['sToken']
sEncodingAESKey = URLRobackSetting['sEncodingAESKey']
sCorpID = URLRobackSetting['sCorpID']
wxcpt = WXBizMsgCrypt(sToken, sEncodingAESKey, sCorpID)

# 公共参数
sVerifyMsgSig = request.GET.get("msg_signature")
sVerifyTimeStamp = request.GET.get("timestamp")
sVerifyNonce = request.GET.get("nonce")
sVerifyEchoStr = request.GET.get("echostr")

baseData = {
   "wxcpt": wxcpt,
   "sVerifyMsgSig": sVerifyMsgSig,
   "sVerifyTimeStamp": sVerifyTimeStamp,
   "sVerifyNonce": sVerifyNonce
}
# 自定义消息处理类
messageHandler = MessageHandler(baseData)

接受消息，解析成明文，调用官方的解密函数：DecryptMsg

def receiveInfo(self, sReqData):
   """"接收消息"""
    try:
        # 解密
        sReqMsgSig = self.sVerifyMsgSig
        sReqTimeStamp = self.sVerifyTimeStamp
        sReqNonce = self.sVerifyNonce

        # 解密函数
        ret, sMsg = self.wxcpt.DecryptMsg(
            sReqData, sReqMsgSig, sReqTimeStamp, sReqNonce)
        print("=>", ret, sMsg)
        # 此时sMsg为明文
        return ret, sMsg

    except Exception as e:
        raise e

解析明文中的内容，然后定制消息：

def responseInfo(self, sEncryptMsg):
    result = xmltodict.parse(sEncryptMsg)
    xml = result["xml"]

    # 解析内容
    content = xml['Content']
    fromUserName = xml['FromUserName']

    # 定制消息
    if content == "test":
        message = "this is success test"
    else:
        message = "please check document"

    return self.customMessage(fromUserName, message)

xmltodict：模块，用于将xml转换为python中的字典

定制的消息按照官方进行数据加密，调用函数：EncryptMsg文章来源地址https://www.toymoban.com/news/detail-499453.html

def customMessage(self, user, content):
   try:
        response_data = "<xml><ToUserName><![CDATA[" + user + \
            "]]></ToUserName><FromUserName><![CDATA[corid]]></FromUserName><CreateTime>1476422779</CreateTime><MsgType><![CDATA[text]]></MsgType><Content><![CDATA[" + content + "]]></Content></xml>"
        # response_data = unicode(response_data).encode('utf8')  # python2处理
        print("data=>", response_data)
        ret, sEncryptMsg = self.wxcpt.EncryptMsg(
            response_data, self.sVerifyNonce, self.sVerifyTimeStamp)
        if(ret != 0):
            raise Exception
        # 返回加密的回复消息
        final_data = {
            "code": 200,
            "sEncryptMsg": sEncryptMsg
        }
        return HttpResponse(json.dumps(final_data))
    except Exception:
        raise Exception

完整测试Demo

@csrf_exempt
def versityInfo(request):
    sToken = URLRobackSetting['sToken']
    sEncodingAESKey = URLRobackSetting['sEncodingAESKey']
    sCorpID = URLRobackSetting['sCorpID']
    try:
        wxcpt = WXBizMsgCrypt(sToken, sEncodingAESKey, sCorpID)

        # 公共参数
        sVerifyMsgSig = request.GET.get("msg_signature")
        sVerifyTimeStamp = request.GET.get("timestamp")
        sVerifyNonce = request.GET.get("nonce")
        sVerifyEchoStr = request.GET.get("echostr")

        baseData = {
            "wxcpt": wxcpt,
            "sVerifyMsgSig": sVerifyMsgSig,
            "sVerifyTimeStamp": sVerifyTimeStamp,
            "sVerifyNonce": sVerifyNonce
        }
        # 自定义消息处理类
        messageHandler = MessageHandler(baseData)

        if sVerifyEchoStr:
            print("=>", sVerifyMsgSig, sVerifyTimeStamp, sVerifyEchoStr)
            ret, sEchoStr = wxcpt.VerifyURL(
                sVerifyMsgSig, sVerifyTimeStamp, sVerifyNonce, sVerifyEchoStr)
            if(ret != 0):
                print("ERR: VerifyURL ret: " + str(ret))
                final_data = {
                    "code": 400,
                    "sEncryptMsg": "验证未通过"
                }
                return HttpResponse(json.dumps(final_data))
            return HttpResponse(sEchoStr)
        else:  # 接收消息
            # 提取参数
            # unicode(string).encode("utf-8")
            sReqData = request.body
            sReqMsgSig = sVerifyMsgSig
            sReqTimeStamp = sVerifyTimeStamp
            sReqNonce = sVerifyNonce
            print("=>", sReqData, type(sReqMsgSig), sReqTimeStamp, sReqNonce)

            # 接收数据
            ret, sEncryptMsg = messageHandler.receiveInfo(sReqData)
            if(ret != 0):
                print("ERR: EncryptMsg ret: ", str(ret))
                final_data = {
                    "code": 400,
                    "sEncryptMsg": "参数错误"
                }
                return HttpResponse(json.dumps(final_data))

            # 解析明文数据，然后发送消息
            return messageHandler.responseInfo(sEncryptMsg)
    except Exception as e:
        print("system error:", e)
        final_data = {
            "code": 10056,
            "sEncryptMsg": "系统当前维修。。。"
        }
        response = HttpResponse(json.dumps(final_data))
        return response


# 自定义消息体
class MessageHandler:
    def __init__(self, baseData):
        self.wxcpt = baseData['wxcpt']
        self.sVerifyMsgSig = baseData['sVerifyMsgSig']
        self.sVerifyTimeStamp = baseData['sVerifyTimeStamp']
        self.sVerifyNonce = baseData['sVerifyNonce']

    def receiveInfo(self, sReqData):
        """"接收消息"""
        try:
            # 解密
            sReqMsgSig = self.sVerifyMsgSig
            sReqTimeStamp = self.sVerifyTimeStamp
            sReqNonce = self.sVerifyNonce

            # 解密函数
            ret, sMsg = self.wxcpt.DecryptMsg(
                sReqData, sReqMsgSig, sReqTimeStamp, sReqNonce)
            print("=>", ret, sMsg)
            # 此时sMsg为明文
            return ret, sMsg

        except Exception as e:
            raise e

    def responseInfo(self, sEncryptMsg):
        result = xmltodict.parse(sEncryptMsg)
        xml = result["xml"]

        # 解析内容
        content = xml['Content']
        fromUserName = xml['FromUserName']

        # 定制消息
        if content == "test":
            message = "this is success test"
        else:
            message = "please check document"

        return self.customMessage(fromUserName, message)

    def customMessage(self, user, content):
        try:
            response_data = "<xml><ToUserName><![CDATA[" + user + \
                "]]></ToUserName><FromUserName><![CDATA[wx7fc55822f12887ad]]></FromUserName><CreateTime>1476422779</CreateTime><MsgType><![CDATA[text]]></MsgType><Content><![CDATA[" + content + "]]></Content></xml>"
            # response_data = unicode(response_data).encode('utf8')  # python2处理
            print("data=>", response_data)
            ret, sEncryptMsg = self.wxcpt.EncryptMsg(
                response_data, self.sVerifyNonce, self.sVerifyTimeStamp)
            if(ret != 0):
                raise Exception
            # 返回加密的回复消息
            final_data = {
                "code": 200,
                "sEncryptMsg": sEncryptMsg
            }
            return HttpResponse(json.dumps(final_data))
        except Exception:
            raise Exception