企业微信应用消息通知与被动回调回复

最近用python写了一个抖音监控的软件,使用了企业微信的机器人进行通知.具体实现就是爬虫和轮询监听视频列表.

在写企业微信应用回调的时候,发现官方文档写的让我感觉不太舒服,网上能搜索到的资料也不太多,所以在这里把自己的理解记录一下

一切的前提是你会注册企业微信并且创建应用

以下是一些有用的文档

一切功能的前提

获取access_token - 接口文档 - 企业微信开发者中心 (qq.com)

最基础的主动发送通知

应用推送消息 - 文档 - 企业微信开发者中心 (qq.com)

该文档记录了所有类型的回复模板,还是比较简单的,根据官方给的请求示例构建一下参数就可以.

GentleCP/corpwechatbot: 企业微信的python封装接口,简易上手,开箱即用,一行代码实现消息推送)

这是一个封装好的Python库,主动消息发送起来很方便.

回调功能

难点在回调功能,涉及到加解密以及要求回复的格式是XML

接受消息需要加解密参数,官方说明如下:

加解密方案说明 - 文档 - 企业微信开发者中心 (qq.com)

这里推荐使用官方人员提供的测试案例的工具类进行加解密

加解密库下载与返回码 - 文档 - 企业微信开发者中心 (qq.com)

这里提供了几个常见语言的加解密库,可以直接使用.

当然,如果你和我一样是小白的话,我相信你看到目前还是一头雾水,解密然后呢,这里我推荐阅读官方提供的案例

weworkapi_python/callback at master · sbzhu/weworkapi_python (github.com)

下面我将讲述如何快速开发接收消息并回复的功能,我使用fastApi进行举例

1.验证企业微信应用服务器有效性,路径都是根目录'/'

image-20240622214549493
@app.get("/")
async def Verify(msg_signature: str, timestamp: str, nonce: str, echostr: str):
  sVerifyMsgSig = msg_signature
  sVerifyTimeStamp = timestamp
  sVerifyNonce = nonce
  sVerifyEchoStr = echostr
  # 使用官方解密库确认有效性
  ret, sReplyEchoStr = wxcpt.VerifyURL(sVerifyMsgSig, sVerifyTimeStamp,sVerifyNonce,sVerifyEchoStr)
  if( ret!=0 ):
      print("ERR: DecryptMsg ret: " + str(ret))
      sys.exit(1)
  return int(sReplyEchoStr)

2.接收消息

image-20240622214737308
image-20240622214924546
# 接受消息模版
Recived_Temp = """<xml>
  <ToUserName><![CDATA[%(ToUserName)s]]></ToUserName>
  <AgentID><![CDATA[%(AgentID)s]]></AgentID>
  <Encrypt><![CDATA[%(Encrypt)s]]></Encrypt>
</xml>"""

@app.post("/")
async def main(msg_signature: str, timestamp: str, nonce: str, q: str = None, item: Item = Depends(XmlBody(Item))):
   Recived_dict = {
       'ToUserName': item.ToUserName,
       'AgentID': item.AgentID,
       'Encrypt': item.Encrypt,
          }
   ReqData = Recived_Temp % Recived_dict
   ret,sMsg=wxcpt.DecryptMsg(sPostData=ReqData, sMsgSignature=msg_signature, sTimeStamp=timestamp, sNonce=nonce)
   if( ret!=0 ):
       print("ERR: DecryptMsg ret: " + str(ret))
       sys.exit(1)
   xml_tree = ET.fromstring(sMsg)
   MsgType = xml_tree.find("MsgType").text
   # msg类型
   if MsgType == "text":
       print("text")
       content_recived = xml_tree.find("Content").text
       FromUserName = xml_tree.find("FromUserName").text
       ToUserName = xml_tree.find("ToUserName").text
       sRespData = func.handle_msg(to_user_id = ToUserName, recived_msg = content_recived,from_user_id=FromUserName,time=timestamp)
       ret,sEncryptMsg=wxcpt.EncryptMsg(sReplyMsg = sRespData, sNonce = nonce, timestamp = timestamp)
       return sEncryptMsg
   # 点击事件
   elif MsgType == "event":
       print("event")
       Event = xml_tree.find("Event").text
       if Event == "click":
           print("CLICK")
           EventKey = xml_tree.find("EventKey").text
           print(EventKey)
           FromUserName = xml_tree.find("FromUserName").text
           ToUserName = xml_tree.find("ToUserName").text
           sRespData = func.handle_msg(to_user_id = ToUserName, recived_msg = EventKey,from_user_id=FromUserName,time=timestamp)
           ret,sEncryptMsg=wxcpt.EncryptMsg(sReplyMsg = sRespData, sNonce = nonce, timestamp = timestamp)
           return sEncryptMsg

这里通过解析发送消息XML模板获得数据,MsgType代表了数据的类型.func.handle_msg是我用来构造回复XML格式的方法,消息类型的判断也可以放在func中,具体怎么抽象方法还看你们.

handle_msg方法 (这里以我只发送click事件和文字消息举例)

Send_vedio_Temp="""
<xml>
  <ToUserName>%(ToUserName)s</ToUserName>
  <FromUserName>%(FromUserName)s</FromUserName>
  <CreateTime>%(timestamp)s</CreateTime>
  <MsgType>video</MsgType>
  <Video>
      <MediaId>%(MediaId)s</MediaId>
      <Title>%(Title)s</Title>
      <Description>%(Description)s</Description>
  </Video>
</xml>
"""

#发送消息模版
Send_Temp = """<xml>
  <ToUserName>%(ToUserName)s</ToUserName>
  <FromUserName>%(FromUserName)s</FromUserName>
  <CreateTime>%(timestamp)s</CreateTime>
  <MsgType>text</MsgType>
  <Content>%(content)s</Content>
</xml>"""

def handle_msg(to_user_id: str, recived_msg: str,from_user_id:str,time):
   name = to_user_id
   if '/get' in recived_msg:
       Send_dict = {
           "ToUserName": to_user_id,
           "FromUserName": from_user_id,
           "timestamp": time,
           "content": "ababab",
      }
       return Send_Temp % Send_dict
   elif '/video' in recived_msg  or 'random_video' in recived_msg:
       download_video_thread = threading.Thread(target=download_video)
       download_video_thread.start()
       download_video_thread.join()
       parent_directory_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
       media_id =uploadfile(parent_directory_path,'video.mp4')
      send_dic={
               "ToUserName": to_user_id,
               "FromUserName": from_user_id,
               "timestamp": time,
               "MediaId": media_id,
               "Title": title,
               "Description": tags
      }
       return Send_vedio_Temp % send_dic

思路就是解析消息内容,根据类型和关键字构造返回消息内容

下面这个文档可以查看消息格式类型

被动回复消息格式 - 文档 - 企业微信开发者中心 (qq.com)

最近做了什么

小组作业

最近一星期都在写小组作业,Springboot3+Vue3写一个招聘网站,一带四.界面模仿BOSS招聘写的.

写着写着突然感觉自己更想写一些底层的东西.希望自己未来有能力实现吧

image-20240626184932182

Angel,请你不要放开我的手