【Python】获取企业微信汇报数据，机器人推送源码教程

hack78

事情是这样的， 企业微信汇报数据可以发送到群聊，实在是不方便，而且需要手动点击才能看到，效果如下

经过一番改造以后的效果。


原理，通过企业微信汇报数据接口：https://work.weixin.qq.com/api/doc/90000/90135/93394
和企业微信机器人接口https://work.weixin.qq.com/api/doc/90000/90136/91770

来实现以上的功能；话不多说，直接上代码；

[Python] 纯文本查看 复制代码

?

001 002 003 004 005 006 007 008 009 010 011 012 013 014 015 016 017 018 019 020 021 022 023 024 025 026 027 028 029 030 031 032 033 034 035 036 037 038 039 040 041 042 043 044 045 046 047 048 049 050 051 052 053 054 055 056 057 058 059 060 061 062 063 064 065 066 067 068 069 070 071 072 073 074 075 076 077 078 079 080 081 082 083 084 085 086 087 088 089 090 091 092 093 094 095 096 097 098 099 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152

# coding=utf-8   import time import requests   class WX:     def __init__(self):         self.corpid = ""#企业ID         self.corpsecret = ""#汇报应用Secret         self.access_url = " https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={}&corpsecret={}".format(             self.corpid, self.corpsecret         )           self.webhook = ""#群Webhook地址     def GET_ACCESS_Token(self):         response = requests.get(self.access_url).json()         print(response)         access_token = response['access_token']         return access_token       def get_record_list(self, cursor=0):         starttime = str(time.mktime(time.strptime(str(time.strftime("%Y-%m-%d") + " 00:00:00"), '%Y-%m-%d %H:%M:%S')))[                     :-2]         endtime = str(time.mktime(time.strptime(str(time.strftime("%Y-%m-%d") + " 23:59:59"), '%Y-%m-%d %H:%M:%S')))[                   :-2]         print(int((time.time()-60)))         url = "https://qyapi.weixin.qq.com/cgi-bin/oa/journal/get_record_list?access_token={}".format(self.GET_ACCESS_Token())         data = {             #"starttime": int(time.time()), #正式使用的时间戳             "starttime": int((time.time()-60)), #测试使用的时间戳             #"endtime": int(time.time()+86400), #正式使用的时间戳             "endtime": int((time.time()+60)),  #测试使用的时间戳             "cursor": cursor,             "limit": 100,             "filters": [                 {                     "key": "template_id",                     "value": "3TjzDqVEqLJFs6GrC7T5WNXiU1LiGmLY1guJ7hs4"#订单号，日报的                 }             ]         }         response = requests.post(url=url, json=data).json()         # print(response)         # print(len(response['journaluuid_list']))         print(response)         result = [[response['next_cursor']], response['journaluuid_list']]         print('1111',result)         return result       def GET_details(self, journaluuid):         edg = 1         user_list = [["张三", "******"], ["李四", '********'], ["王五", '18*****191']]#这个地方无法获取用户名，所以做了一个手动列表,前面用户名，后面是企业微信用户ID         url = "https://qyapi.weixin.qq.com/cgi-bin/oa/journal/get_record_detail?access_token={}".format(str(self.GET_ACCESS_Token()))         data = {             "journaluuid": journaluuid,         }         response = requests.post(url=url, json=data).json()['info']         print(response)           template_name = response['template_name']         report_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(response['report_time']))         submitter = response['submitter']['userid']         for user in user_list:             if submitter in user:                 submitter = user[0]           receiverss = []         number = 0         while True:             try:                 receivers = response['receivers'][number]['userid']                 for user in user_list:                     if receivers in user:                         receiver = user[0]                         receiverss.append(receiver)                 number += 1             except Exception:                 break         try:             receivers = ','.join(receiverss)             Product = response['apply_data']['contents'][0]['value']['text']             Phone = response['apply_data']['contents'][1]['value']['text']             Customer_Name = response['apply_data']['contents'][2]['value']['text']             Customer_Company = response['apply_data']['contents'][3]['value']['text']             Province = response['apply_data']['contents'][4]['value']['text']             Company_category = response['apply_data']['contents'][5]['value']['selector']['options'][0]['value'][0][                 'text']             MarkDownText = " ### %s\r %s\r\n>   ### 今日工作：\n<font color=\"warning\"> %s\r</font>\n"  "> ### 明日计划：\n <font color=\"info\"> %s\r</font>\n  提交时间：%s" %(                     submitter,template_name, Product, Phone,  report_time                            )             print(MarkDownText)             return MarkDownText,edg         except Exception as e:             print(e)             edg = 0             receivers = ','.join(receiverss)             Product = response['apply_data']['contents'][0]['value']['text']             Phone = response['apply_data']['contents'][1]['value']['text']             Customer_Name = response['apply_data']['contents'][2]['value']['text']             MarkDownText = " ### %s\r %s\r\n>   ### 今日工作：\n<font color=\"warning\"> %s\r</font>\n"  "> ### 明日计划：\n <font color=\"info\"> %s\r</font>\n  提交时间：%s" %(                     submitter,template_name, Product, Phone,  report_time                            )             print(MarkDownText)             return MarkDownText,edg     def Run(self):         try:             next_cursor = [0]             record_list = self.get_record_list(next_cursor[0])             print(record_list)             next_cursor = record_list[0]             record_list = record_list[1]             details_list = record_list[-1]             "模板不一样所以暂时把单号固定"             # details = "8WBtd3jNAAfiJtjHRHm8x64LiRx4oJFLX92kvvVnQ4SRaFbqsEYp4Jji981BwAexjP"             Text, edg = self.GET_details(journaluuid=details_list)             headers = {"Content-Type": "text/plain"}             data = {                 "msgtype": "markdown",                 "markdown": {                     "content": Text,                 }             }             if edg == 0:                 response = requests.post(url=self.webhook, headers=headers, json=data).text                 print("pooo")                 print(response)             else:                 url = ''#机器人地址                 response = requests.post(url=url, headers=headers, json=data).text                 print("pooo2")                 print(response)         except:             print("还没更新")         # for details in details_list:         #     Text = self.GET_details(journaluuid=details)         #     headers = {"Content-Type": "text/plain"}         #     data = {         #         "msgtype": "markdown",         #         "markdown": {         #             "content": Text,         #         }         #     }         #     response = requests.post(url=self.webhook, headers=headers, json=data).text         #     print("pooo")         #     print(response)     if __name__ == '__main__':     while 1:         app = WX()         app.Run()         time.sleep(60)

最后
放在环境里执行，一分钟会自动获取一次数据进行实时推送到机器人群里；

tyler998

感谢分享，支持原创

diorlin

虽然用不上，了解一下也不错

yangming0825

大神，获取企业微信汇报中的模板那，方便帮忙解答下嘛？我这没有发送成功

hack78

yangming0825 发表于 2022-5-31 10:54
大神，获取企业微信汇报中的模板那，方便帮忙解答下嘛？我这没有发送成功

创建好汇报模版，然后获取模版id

zmshan2008

正想试试这个怎么弄，在学习中。。。

yangming0825

大神，我只想要一个汇报模板里的其中一个字段，咋能获取到啊，看企微那边好像每个人提交的汇报内容顺序不一样，这种情况怎么搞哇，麻烦大神帮解答一下