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

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

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

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

以下是一些有用的文档

• 开发前必读 - 文档 - 企业微信开发者中心 (qq.com)

一切功能的前提

获取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.验证企业微信应用服务器有效性,路径都是根目录'/'

@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.接收消息

# 接受消息模版
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招聘写的.

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