python发送企业微信群webhook消息（文本、文件）-Toy模板网

这篇具有很好参考价值的文章主要介绍了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}：<font color=\"info\">{info}</font>\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