python发送企业微信群webhook消息（文本、文件）

import datetime
import os
import time
from copy import copy

import requests
from loguru import logger
from urllib3 import encode_multipart_formdata


class WeiXin_Robot:
    def __init__(
            self,
            url: str = ""
    ):

        # 测试car
        test_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"

        self.url = url or test_url
        self.mHeader = {'Content-Type': 'application/json; charset=UTF-8'}


    def send_cars_msg(self, car_dict):
        logger.info(f"{car_dict=}")
        cars = f"  ##通知\n" \
               f"一一一一一一一一一一一一一一\n"
        for vin, info in car_dict.items():
            info_str = f"{vin}：{info}\n"
            cars += info_str
        data = {
            "msgtype": "markdown",
            "markdown": {
                'content': cars
            }
        }
        self.send(data=data)

    def send_error_msg(self, info):
        data = {
            "msgtype": "text",
            "text": {
                "content":
                    f"企业微信异常提醒\n"
                    f"问题描述：{info.get('desc')}\n"
                    f"名称：{info.get('name')}\n"
                    f"id：{info.get('media_id')}\n",
                "mentioned_mobile_list": ["18817957261"]
            }
        }
        self.send(data=data)

    def send(self, data):
        try:
            info = requests.post(url=self.url, json=data, headers=self.mHeader)
            if info.json().get("errcode") == 0 and info.json().get("errmsg") == "ok":
                logger.info(f"企业微信通知发送成功，msg={info.json()}")
            else:
                logger.warning(f"企业微信通知发送异常,{info.json()=}")
        except Exception as e:
            logger.warning("企业微信通知发送异常")
            logger.warning(e)
            pass

    # file_path: e.g /root/data/test_file.xlsx
    # 如果D:\\windows\\ 下面file_name的split需要调整一下
    # upload_file 是为了生成 media_id， 供消息使用
    def upload_file(self, file_path):
        try:
            key = self.url.split('=')[1]
            wx_upload_url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/upload_media?key={key}&type=file"
            file_name = file_path.split("/")[-1]
            with open(file_path, 'rb') as f:
                length = os.path.getsize(file_path)
                data = f.read()
            print(file_name)
            headers = {"Content-Type": "application/octet-stream"}
            params = {
                "name": 'media',
                "filename": file_name,
                "filelength": length,
            }
            file_data = copy(params)
            file_data['file'] = (file_path.split('/')[-1:][0], data)
            encode_data = encode_multipart_formdata(file_data)
            file_data = encode_data[0]
            headers['Content-Type'] = encode_data[1]
            r = requests.post(wx_upload_url, data=file_data, headers=headers)
            print(r.text)
            if r.json().get("errcode") == 0 and r.json().get("errmsg") == "ok":
                logger.info(f"上传文件到企业微信成功，msg={r.json()},{file_path=}")
                media_id = r.json()['media_id']
                return media_id
            else:
                logger.warning(f"上传文件到企业微信异常,{r.json()=},{file_path=}")

        except Exception as e:
            logger.warning("上传文件到企业微信失败")
            logger.warning(e)
            pass

    # media_id 通过上一步上传的方法获得
    def send_file(self, file_path=None, media_id=""):
        """企业微信发送文件"""
        if not media_id:
            media_id = self.upload_file(file_path=file_path)
            time.sleep(1)
        try:
            # headers = {"Content-Type": "text/plain"}
            data = {
                "msgtype": "file",
                "file": {
                    "media_id": media_id
                }
            }
            for i in range(3):
                r = requests.post(url=self.url, json=data, headers=self.mHeader)
                if r.json().get("errcode") == 0 and r.json().get("errmsg") == "ok":
                    logger.info(f"企业微信发送文件成功，msg={r.json()},{file_path=},{media_id=}")
                    break
                else:
                    logger.warning(f"企业微信发送文件异常,{r.json()=},{file_path=}")
                    logger.warning('3s后再次尝试发送文件...')
                    time.sleep(3)
                    if i == 2:
                        self.send_error_msg(
                            info={
                                'desc': f'企业微信发送文件异常',
                                "name": file_path,
                                'media_id': media_id,
                            })
            # print(r.text)
        except Exception as e:
            logger.warning("企业微信发送文件异常")
            logger.warning(e)
            pass


if __name__ == "__main__":
    pass

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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