官方文档地址：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
可以通过如下步骤设置企业微信机器人：
- 首先建立或者进入某个群聊
- 进入群聊设置页面， 点击“群机器人>添加”可添加一个机器人成功
- 添加成功后，复制并保留其webhook地址。
二、python封装脚本
# -*- coding: utf-8 -*-# @Time : 2023/5/11 15:01# @Author : chenyinhua# @File : webchat_handle.py# @Software: PyCharm# @Desc: 企业微信机器人import osfrom requests import requestfrom loguru import loggerimport base64import hashlibimport reclass 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 Falseif __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 = "实时新增用户反馈132例，请相关同事注意。n>类型:用户反馈>普通用户反馈:117例>VIP用户反馈:15例" bot.send_markdown(content=md) bot.send_picture(image_path=r"xxxxxx.png") articles = [ { "title": "中秋节礼品领取", "description": "今年中秋节公司有豪礼相送", "url": "www.qq.com", "picurl": "http://www.quxuehao.com/file/tupian/20230516/test_pic_msg1.png" } ] bot.send_text_picture(articles=articles) filepath = r"xxxxxxxapiautotest-report-2023-05-11 14_57_18.html" bot.send_file(media_id=bot.upload_file(filepath))
