您现在的位置：程式師世界 >> 編程語言 > >> 更多編程語言 >> Python

【企業微信-消息推送】模板卡片消息-Python代碼

編輯：Python

文章目錄

- @[TOC](文章目錄)
配置回調接口
回調接口的GET請求驗證
發送模板卡片消息
被動響應消息
完整測試Demo

浏覽消息推送的時候，在模板卡片部分停滯很久；其中主要涉及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為第三方庫，裡面一共提供了三個方法，詳細可以浏覽官網。

代碼完成可以使用接口測試工具進行測試：官網鏈接-接口測試工具

然後返回點擊保存即可完成配置。

發送模板卡片消息

請求方式：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
}

運行結果：

被動響應消息

在企業微信中可以根據發送的消息定制回調消息，效果圖如下：

被動接收消息走的回調接口的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

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