1.调用api向企业微信(通过CORPID标识)的自建应用程序(通过Secret, AgentID标识)发送文本、图片或者文件消息;
2.创建实例时传入以下参数:
(1) touser–>str,接收消息者的标识(已在企业微信的通讯录中添加,添加后微信后台会自动分配标识),多个用户使用 “|” 隔开(如"zhdb|zhj|wqq"),所有人("@all");
(2) corp_id–>str,企业ID,申请企业微信时获得;
(3) secret, agent_id–>str, 创建企业应用时获得.
3.本实例定义了三种消息格式的发送(text, image, file), 使用实例如下:
chat = CorpWechat(touser, corp_id, secret, agent_id)# 发送text文本消息chat.send_message(msg_type='text', contents="Format message str")# 发送image图片消息(本地图片)chat.send_message(msg_type='image', file_obj=open(image_path, 'rb'))# 发送file文件消息(本地文件)chat.send_message(msg_type='file', file_obj=open(file_path, 'rb'))
另外也经常有这种情况,调用其它图片生成库在线绘制的图形可先存入到二进制缓存文件中,再将该对象作为file_obj参数传入,
例如用发送用matplotlib所生成的图片:
import matplotlib.pyplot as pltimport numpy as npfrom io import BytesIOx = np.arange(50)plt.plot(x, x**2)buffer = BytesIO() # 创建缓存文件plt.savefig(buffer) # 将生成的图片存入缓存文件data = buffer.getvalue() # 读取成可以传入file_obj的数据格式chat.send_message(msg_type='image', file_obj=data)
再有就是发送调用其它文件生成库生成的文件的话,可先创建临时文件夹,然后保存到该文件夹下,发送时再读取,
例如用reportlab生成PDF文件的发送如下:
–>脚本目录下创建temp文件夹;
from reportlab.pdfgen import canvaspdf_path = 'temp\\test.pdf'c = canvas.Canvas(pdf_path)c.drawString(50, 50, "This is a test pdf file!")c.save() # 生成文件到本地chat.send_message(msg_type='file', file_obj=open(pdf_path, 'rb'))
CorpWechat类创建脚本如下:
# -*- coding: utf-8 -*-import requestsimport jsonclass CorpWechat:def __init__(self, touser, corp_id, secret, agent_id):self.base_url = "https://qyapi./cgi-bin"self.touser = touserself.corp_id = corp_idself.secret = secretself.agent_id = agent_idself.token = self._get_token()def _get_token(self):arg_url = '/gettoken?corpid={}&corpsecret={}'.format(self.corp_id, self.secret)url = self.base_url + arg_urlr = requests.get(url)js = json.loads(r.text)try:return js['access_token']except KeyError: # 成功返回后就会获得'access_token'字段,否则报错KeyErrorraise KeyError("Get access-token failed.")def _get_media_id(self, msg_type, file_obj):arg_url = "/media/upload?access_token={}&type={}".format(self.token, msg_type)url = self.base_url + arg_urldata = {"media": file_obj}r = requests.post(url=url, files=data)js = r.json()try:return js['media_id']except KeyError: # 成功返回后就会获得'media_id'字段,否则报错KeyErrorraise KeyError("Get media_id failed.")def _gen_msg(self, msg_type, contents, file_obj):base_string = '''{"touser": self.touser, "msgtype": msg_type, "agentid": self.agent_id, msg_type: {'%s': '%s'},"safe": 0}'''if msg_type == 'text':values = base_string % ('content', contents)else:media_id = self._get_media_id(msg_type, file_obj)values = base_string % ('media_id', media_id)data = eval(values)js = json.dumps(data)to_bytes = bytes(js, 'utf-8')return to_bytesdef send_message(self, msg_type, contents='', file_obj=None):post_msg = self._gen_msg(msg_type, contents, file_obj)arg_url = '/message/send?access_token={}'.format(self.token)url = self.base_url + arg_urlrequests.post(url, data=post_msg)