网络工程师的Python3 企业微信机器人调用

简介

企业微信的机器人可以发送文字、图片、Markdown、图文、文件、语音等消息。
下面的脚本将其封装为一个class，方便调用。

官方文档：https://developer.work.weixin.qq.com/document/path/91770

PS：首先要获取一个webhook，例如：https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=693a91f6-7xxx-4bc4-97a0-0ec2sifa5aaa

import os

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


class WechatBot: # 定义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_voice(self, media_id):
        """
        发送图片消息
        :param voice_path: 语音文件的绝对路径
        """
        payload = {
            "msgtype": "voice",
            "voice": {
                "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
    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 upload_voice_file(self, vocie_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=voice"
        headers = {
            "Content-Type": "multipart/form-data;"
        }
        with open(vocie_file_path, "rb") as f:
            files = {"media": (os.path.basename(vocie_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=xxxx"
    bot = WechatBot(webhook_url)
    ## 定义机器人访问地址


    bot.send_text(content="企业机器人发送文字测试", mentioned_list=["@all"])
    ## 只发送文本消息


    bot.send_text(content="企业机器人发送文字测试 AT测试", mentioned_list=["@all"], mentioned_mobile_list=["18774970063"])

    ## 只发送文本消息，@所有人。可指定@手机号和@userid

    md = ("MarkDown测试<font color=\"warning\">颜色</font>，BBQ。\n"
          ">类型:<font color=\"comment\">引用格式测试</font>>普通用户反馈:<font color=\"comment\">117例</font>>VIP用户反馈:<font color=\"comment\">15例</font>")
    bot.send_markdown(content=md)

    ## 发送markdown消息，支持部分markdown语法,支持多颜色

    bot.send_picture(image_path=r"49f32e7dad281833ca8c4.png")


    ## 发送图片消息，图片路径为绝对路径。因通过Base64编码上传，文件不能超过20M。

    articles = [
        {
            "title": "Songxwn博客订阅",
            "description": "写一些计算机网络、Linux、虚拟化、Python自动化相关的学习笔记。",
            "url": "https://songxwn.com",
            "picurl": "https://img.songxwn.com/file/ed2c48db6dccbcbfe29fa.jpg"
        }
    ]
    bot.send_text_picture(articles=articles)

    ## 发送图文消息，支持多图文消息，每个图文消息支持title、description、url、picurl字段


    filepath = r"main.py"
    bot.send_file(media_id=bot.upload_file(filepath))

    ## 上传文件并发送文件消息，文件路径为绝对路径。因通过Base64编码上传，文件不能超过20M。

    voice_file_path = r"玛卡巴卡呀_1.amr"

    bot.send_voice(media_id=bot.upload_voice_file(voice_file_path))

    ## 发送语音消息，语音路径为绝对路径。语音(voice)：文件大小不超过2M，播放长度不超过60s，仅支持AMR格式

# PS：企业微信机器人API文档：https://developer.work.weixin.qq.com/document/path/91770
# 企业微信机器人有发送频率限制，为每分钟20次。

代码参考：https://blog.csdn.net/FloraCHY/article/details/130624354