准备工作：
到企业微信官网，注册一个企业；登录企业微信后台，创建一个“自建”应用， 获取企业ID、agentid、secret这3个必要的参数；在企业微信的通讯录中，创建多个测试账号；在手机端安装“企业微信”APP，使用测试账号登录到企业微信，准备接收消息。
程序代码：
企业微信提供API开发接口，通过HTTPS的GET、POST方法与企业微信后台进行交互，完成获取令牌、发送数据、获取数据的操作。
Python代码主要使用requests库，将企业微信API进行简单封装，模拟https的GET、POST操作，向指定的用户发送企业微信消息。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import time
import requests
import json
class WeChat:
def __init__(self):
self.CORPID = 'ww2e1234567895498f5498f' #企业ID，在管理后台获取
self.CORPSECRET = 'xy11234567898hk_ecJ123456789DhKy4_1y12345OI'#自建应用的Secret，每个自建应用里都有单独的secret
self.AGENTID = '1000002' #应用ID，在后台应用中获取
self.TOUSER = "maomao|dingding" # 接收者用户名,多个用户用|分割
def _get_access_token(self):
url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken'
values = {'corpid': self.CORPID,
'corpsecret': self.CORPSECRET,
}
req = requests.post(url, params=values)
data = json.loads(req.text)
return data["access_token"]
def get_access_token(self):
try:
with open('./tmp/access_token.conf', 'r') as f:
t, access_token = f.read().split()
except:
with open('./tmp/access_token.conf', 'w') as f:
access_token = self._get_access_token()
cur_time = time.time()
f.write('\t'.join([str(cur_time), access_token]))
return access_token
else:
cur_time = time.time()
if 0 < cur_time - float(t) < 7260:
return access_token
else:
with open('./tmp/access_token.conf', 'w') as f:
access_token = self._get_access_token()
f.write('\t'.join([str(cur_time), access_token]))
return access_token
def send_data(self, message):
send_url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=' + self.get_access_token()
send_values = {
"touser": self.TOUSER,
"msgtype": "text",
"agentid": self.AGENTID,
"text": {
"content": message
},
"safe": "0"
}
send_msges=(bytes(json.dumps(send_values), 'utf-8'))
respone = requests.post(send_url, send_msges)
respone = respone.json() #当返回的数据是json串的时候直接用.json即可将respone转换成字典
return respone["errmsg"]
if __name__ == '__main__':
wx = WeChat()
wx.send_data("这是程序发送的第1条消息！\n Python程序调用企业微信API,从自建应用“告警测试应用”发送给管理员的消息！")
wx.send_data("这是程序发送的第2条消息！")
https://blog.csdn.net/CSDN_duomaomao/article/details/90193422
#!/usr/bin/python
# -*- coding: utf-8 -*-
import time
import requests
import json,os
class WeChat_SMS:
def __init__(self):
self.CORPID = 'XXXX'#企业ID， 登陆企业微信，在我的企业-->企业信息里查看
self.CORPSECRET = 'XXXXX'#自建应用，每个自建应用里都有单独的secret
self.AGENTID = 'XXXX' #应用代码
self.TOUSER = "XXX"# 接收者用户名,@all 全体成员
self.TOPARY = "2" #部门ID
def _get_access_token(self):
url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken'
values = {'corpid': self.CORPID,'corpsecret': self.CORPSECRET,}
req = requests.post(url, params=values)
data = json.loads(req.text)
#print (data)
return data["access_token"]
def get_access_token(self):
try:
with open('access_token.conf', 'r') as f:
t, access_token = f.read().split()
except:
with open('access_token.conf', 'w') as f:
access_token = self._get_access_token()
cur_time = time.time()
f.write('\t'.join([str(cur_time), access_token]))
return access_token
else:
cur_time = time.time()
if 0 < cur_time - float(t) < 7200:#token的有效时间7200s
return access_token
else:
with open('access_token.conf', 'w') as f:
access_token = self._get_access_token()
f.write('\t'.join([str(cur_time), access_token]))
return access_token
def send_data(self, msg):
send_url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=' + self.get_access_token()
send_values = {
"touser": self.TOUSER,
#"toparty": self.TOPARY, #设置给部门发送
"msgtype": "text",
"agentid": self.AGENTID,
"text": {
"content": msg
},
"safe": "0"
}
send_msges=(bytes(json.dumps(send_values), 'utf-8'))
respone = requests.post(send_url, send_msges)
respone = respone.json()#当返回的数据是json串的时候直接用.json即可将respone转换成字典
#print (respone["errmsg"])
return respone["errmsg"]
if __name__ == '__main__':
wx = WeChat_SMS()
wx.send_data(msg="服务崩了，你还在这里吟诗作对？")
#以下是添加对日志的监控
# srcfile = u"G:/123.txt"
# file = open(srcfile)
# file.seek(0, os.SEEK_END)
# while 1:
# where = file.tell()
# line = file.readline()
# if not line:
# time.sleep(1)
# file.seek(where)
# else:
# print(line)
# wx.send_data(msg=line)
https://blog.csdn.net/liyyzz33/article/details/86080936