封装Python脚本：使用企业微信机器人发送消息至企业微信_机器人的text/markdown类型消息支持在content中使用<@userid>扩展语法来@群

官方文档地址：https://developer.work.weixin.qq.com/document/path/91770#%E5%A6%82%E4%BD%95%E4%BD%BF%E7%94%A8%E7%BE%A4%E6%9C%BA%E5%99%A8%E4%BA%BA

一、获取自定义机器人webhook

可以通过如下步骤设置企业微信机器人：

1. 首先建立或者进入某个群聊
2. 进入群聊设置页面， 点击“群机器人>添加”可添加一个机器人成功
3. 添加成功后，复制并保留其webhook地址。

二、python封装脚本

# -*- coding: utf-8 -*-
# @Time    : 2023/5/11 15:01
# @Author  : chenyinhua
# @File    : webchat_handle.py
# @Software: PyCharm
# @Desc: 企业微信机器人
import os

from requests import request
from loguru import logger
import base64
import hashlib
import re


class WechatBot:
    """
    企业微信机器人
    当前自定义机器人支持文本（text）、markdown（markdown）、图片（image）、图文（news）, 文件（file）五种消息类型。
    机器人的text/markdown类型消息支持在content中使用<@userid>扩展语法来@群成员
    """

    def __init__(self, webhook_url):
        """
        :param webhook_url: 机器人的WebHook_url
        """
        self.webhook_url = webhook_url
        self.headers = {
            "Content-Type": "application/json",
            "Charset": "UTF-8"
        }

    def send_text(self, content, mentioned_list=[], mentioned_mobile_list=[]):
        """
        发送文本消息
        :param content: 文本内容，最长不超过2048个字节，必须是utf8编码
        :param mentioned_list: userid的列表，提醒群中的指定成员(@某个成员)，@all表示提醒所有人，如果开发者获取不到userid，可以使用mentioned_mobile_list
        :param mentioned_mobile_list: 手机号列表，提醒手机号对应的群成员(@某个成员)，@all表示提醒所有人
        """
        payload = {
            "msgtype": "text",
            "text": {
                "content": content,
                "mentioned_list": mentioned_list,
                "mentioned_mobile_list": mentioned_mobile_list
            }
        }
        response = request(url=self.webhook_url, method="POST", json=payload, headers=self.headers)
        if response.json().get("errcode") == 0:
            logger.debug(f"通过企业微信发送文本消息成功：{response.json()}")
            return True
        else:
            logger.error(f"通过企业微信发送文本消息失败：{response.text}")
            return False

    def send_markdown(self, content):
        """
        发送markdown消息
        目前支持的markdown语法是如下的子集：
            1. 标题 （支持1至6级标题，注意#与文字中间要有空格）
            2. 加粗
            3. 链接
            4. 行内代码段（暂不支持跨行）
            5. 引用
            6. 字体颜色(只支持3种内置颜色), 绿色（color="info"），灰色（color="comment"），橙红色（color="warning"）
        :param content: markdown内容，最长不超过4096个字节，必须是utf8编码
        """
        payload = {
            "msgtype": "markdown",
            "markdown": {
                "content": content
            }
        }
        response = request(url=self.webhook_url, method="POST", json=payload, headers=self.headers)
        if response.json().get("errcode") == 0:
            logger.debug(f"通过企业微信发送md消息成功：{response.json()}")
            return True
        else:
            logger.error(f"通过企业微信发送md消息失败：{response.text}")
            return False

    def send_picture(self, image_path):
        """
        发送图片消息
        :param image_path: 图片的绝对路径
        """
        with open(image_path, "rb") as f:
            image_data = f.read()
        payload = {
            "msgtype": "image",
            "image": {
                "base64": base64.b64encode(image_data).decode("utf-8"),  # # 将图片数据转换成Base64编码格式
                "md5": hashlib.md5(image_data).hexdigest()  # # 计算图片的MD5值
            }
        }
        response = request(url=self.webhook_url, method="POST", json=payload, headers=self.headers)
        if response.json().get("errcode") == 0:
            logger.debug(f"通过企业微信发送图片消息成功：{response.json()}")
            return True
        else:
            logger.error(f"通过企业微信发送图片失败：{response.text}")
            return False

    def send_text_picture(self, articles: list):
        """
        发送图文消息
        :param articles: 图文消息，一个图文消息支持1到8条图文, 包括如下字段
            1. title: 标题，不超过128个字节，超过会自动截断
            2. description: 非必填，描述，不超过512个字节，超过会自动截断
            3. url: 点击后跳转的链接。
            4. picurl: 非必填，图文消息的图片链接，支持JPG、PNG格式，较好的效果为大图 1068*455，小图150*150。
        """
        payload = {
            "msgtype": "news",
            "news": {
                "articles": [
                ]
            }
        }
        for article in articles:
            payload["news"]["articles"].append(
                {
                    "title": article.get("title"),
                    "description": article.get("description", ""),
                    "url": article.get("url"),
                    "picurl": article.get("picurl", "")
                }
            )
        response = request(url=self.webhook_url, method="POST", json=payload, headers=self.headers)
        if response.json().get("errcode") == 0:
            logger.debug(f"通过企业微信发送图文消息成功：{response.json()}")
            return True
        else:
            logger.error(f"通过企业微信发送图文失败：{response.text}")
            return False

    def upload_file(self, file_path):
        """
        上传文件到企业微信服务器(要求文件大小在5B~20M之间)
        注意：素材上传得到media_id，该media_id仅三天内有效；media_id只能是对应上传文件的机器人可以使用
        :param file_path: 文件绝对路径
        """
        token_regex = r"key=([\w-]+)"
        match = re.search(token_regex, self.webhook_url)
        token = match.group(1)
        url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/upload_media?key={token}&type=file"
        headers = {
            "Content-Type": "multipart/form-data;"
        }
        with open(file_path, "rb") as f:
            files = {"media": (os.path.basename(file_path), f.read())}
        response = request(url=url, method="POST", files=files, headers=headers)
        if response.json().get("errcode") == 0:
            media_id = response.json().get("media_id")
            logger.debug(f"上传文件成功，media_id= {media_id}")
            return media_id
        else:
            logger.error(f"上传文件失败：{response.text}")
            return False

    def send_file(self, media_id):
        """
        发送文件
        :param media_id: 文件id，通过下文的文件上传接口获取
        """
        payload = {
            "msgtype": "file",
            "file": {
                "media_id": media_id,
            }
        }
        response = request(url=self.webhook_url, method="POST", json=payload, headers=self.headers)
        if response.json().get("errcode") == 0:
            logger.debug(f"通过企业微信发送文件消息成功：{response.json()}")
            return True
        else:
            logger.error(f"通过企业微信发送文件消息失败：{response.text}")
            return False


if __name__ == '__main__':
    webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=***********"
    bot = WechatBot(webhook_url)
    bot.send_text(content="hello1", mentioned_list=["@all"])
    bot.send_text(content="hello2", mentioned_list=["@all"], mentioned_mobile_list=["18774970063"])
    md = "实时新增用户反馈<font color=\"warning\">132例</font>，请相关同事注意。\n>类型:<font color=\"comment\">用户反馈</font>>普通用户反馈:<font color=\"comment\">117例</font>>VIP用户反馈:<font color=\"comment\">15例</font>"
    bot.send_markdown(content=md)
    bot.send_picture(image_path=r"xxxxxx.png")
    articles = [
        {
            "title": "中秋节礼品领取",
            "description": "今年中秋节公司有豪礼相送",
            "url": "www.qq.com",
            "picurl": "http://res.mail.qq.com/node/ww/wwopenmng/images/independent/doc/test_pic_msg1.png"
        }
    ]
    bot.send_text_picture(articles=articles)
    filepath = r"xxxxxxx\apiautotest-report-2023-05-11 14_57_18.html"
    bot.send_file(media_id=bot.upload_file(filepath))

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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200