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

本文主要是介绍python发送企业微信群webhook消息（文本、文件），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

import datetime
import os
import time
from copy import copyimport requests
from loguru import logger
from urllib3 import encode_multipart_formdataclass WeiXin_Robot:def __init__(self,url: str = ""):# 测试cartest_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"self.url = url or test_urlself.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_strdata = {"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_idelse: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=}")breakelse: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)passif __name__ == "__main__":pass