700字范文,内容丰富有趣,生活中的好帮手!
700字范文 > 封装Python脚本:使用企业微信机器人发送消息至企业微信

封装Python脚本:使用企业微信机器人发送消息至企业微信

时间:2019-11-10 21:35:15

相关推荐

封装Python脚本:使用企业微信机器人发送消息至企业微信

官方文档地址:https://developer.work./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 : /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_urlself.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 Trueelse:logger.error(f"通过企业微信发送文本消息失败:{response.text}")return Falsedef 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 Trueelse:logger.error(f"通过企业微信发送md消息失败:{response.text}")return Falsedef 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 Trueelse:logger.error(f"通过企业微信发送图片失败:{response.text}")return Falsedef 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 Trueelse:logger.error(f"通过企业微信发送图文失败:{response.text}")return Falsedef 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./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_idelse:logger.error(f"上传文件失败:{response.text}")return Falsedef 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 Trueelse:logger.error(f"通过企业微信发送文件消息失败:{response.text}")return Falseif __name__ == '__main__':webhook_url = "https://qyapi./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": "","picurl": "http://res./node/ww/wwopenmng/images/independent/doc/test_pic_msg1.png"}]bot.send_text_picture(articles=articles)filepath = r"xxxxxxx\apiautotest-report--05-11 14_57_18.html"bot.send_file(media_id=bot.upload_file(filepath))

本内容不代表本网观点和政治立场,如有侵犯你的权益请联系我们处理。
网友评论
网友评论仅供其表达个人看法,并不表明网站立场。