python开发微信企业应用-报警程序
为了熟悉python，记使用python开发一个简单的微信报警后台的过程，主要完成了微信企业应用中进行文字推送。
- 使用urllib3发送REST（POST，GET）请求；
- 使用Apscheduler定时任务；
- 使用pyinstaller打包应用；
- 使用logging产生日志；
- 使用configparser读取ini配置文件中的属性与参数；
- cx_oracle常见问题；
使用urllib模块
import urllib3
import json
http = urllib3.PoolManager()
def getToken(corpid, corpsecret):
url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={a}&corpsecret={b}'.format(a=corpid, b=corpsecret)
try:
r = http.request('GET', url)
data = json.loads(r.data)
if data['errmsg'] == 'ok':
return data['access_token']
except:
print('Get token is failed, check the network!!!')
使用urllib3的GET方法，定义了一个获取微信token的方法
getToken，需要传输企业微信应用的两个参数corpid，corpsecret；使用json模块对数据进行解析；
format的常见用法，参见： python format 用法详解，Python format 格式化函数
下面是，获取发送对象（user）和发送接口；
def getApp(token, agentid):
url = 'https://qyapi.weixin.qq.com/cgi-bin/agent/get?access_token={a}&agentid={b}'.format(
a=token,
b=agentid,
)
try:
r = http.request('GET', url)
data = json.loads(r.data)
if data['errmsg'] == 'ok':
USERLIST = data['allow_userinfos']['user']
return [item['userid'] for item in USERLIST]
except:
print('Get App user is failed!!!')
def sendMsg(postData):
token = getToken(corpid, corpsecret)
toUser = '|'.join(getApp(token, cp['rest']['agentid']))
# print(toUser)
dataTomsg = {
"touser": toUser,
"toparty": "",
"totag": "",
"msgtype": "text",
"agentid": cp['rest']['agentid'],
"text": {
"content": postData
},
"safe": 0
}
url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={}'.format(token)
try:
r = http.request("POST",
url,
body=json.dumps(dataTomsg),
headers={'Content-Type': 'application/json'})
data = json.loads(r.data)
if data['errmsg'] == 'ok':
USERLIST = data['invaliduser']
return USERLIST.split('|')
except:
print('消息发送失败！')
常见的获取python对象的Attribute（属性）的方法：
方法一：
[item['userid'] for item in USERLIST]方法二：
attrList = map(lambda x: x.attr, objectList)使用`join`方法，组合字符串；加入`try…except’防止异常抛出；
参考文章：
Python——深入理解urllib、urllib2及requests（requests不建议使用？）
定时任务Apscheduler介绍
from apscheduler.schedulers.background import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
import datetime
scheduler = BlockingScheduler()
def minHourJob():
today = datetime.datetime.now()
fromTime = today + datetime.timedelta(hours=-1)
startTime = today.strftime('%Y-%m-%d %H:00:00')
# 使用cx_oracle读取oracle数据库中的数据，传递时间参数
data = orcl.getMinHours(startTime)
if len(data) > 0:
list = data[0]
msg = "分钟数据完整率[按小时统计]：{a}至{b}，收到数据量：{c}，应有数据量：{d}，数据完整率：{e}".format(
a=fromTime.strftime('%Y-%m-%d %H:00:00'),
b=startTime,
c=list[0],
d=list[1],
e=list[2]
)
if isSendCheck('minHourNum', list):
auth.sendMsg(msg)
trigger = CronTrigger(day_of_week='*', hour='*', minute='*', second='*/5')
scheduler.add_job(minHourJob, trigger, id='job_min_hours')
scheduler.start()
使用Blocking阻塞的方式创建了scheduler；
把
minHourJob加入定时任务，设置每5秒执行一次；使用
datetime.timedelta对时间进行运算，strftime格式化时间；
参考文章：
加入logging模块：
import logging
# log part
log = logging.getLogger('定时任务')
log.setLevel(logging.INFO) # DEBUG
fmt = logging.Formatter('[%(levelname)s]: %(name)s-%(message)s')
log_path = "../debug.log"
h = logging.FileHandler(log_path)
# h = logging.StreamHandler()
h.setFormatter(fmt)
log.addHandler(h)
参考文章：
使用pyinstaller打包windows应用
因为考虑到方便在windows上运行，所以使用pyinstaller打包应用；
需要一台windows机器，如果之前是在Linux上开发，那么需要导出requirements.txt（依赖包）
pip install pipreqs
pipreqs [options] <path>
简单的安装pyinstaller之后，就可以使用该命令打包应用了。
pip install pyinstaller
pyinstaller src/app.py
使用pip命令安装包的时候，可以使用-i命令，指定中国大陆的源，这样安装起来会比较快。
pip install pyinstaller -i https://pypi.douban.com/simple
参考文章：
使用configparser读取配置文件
有的时候，打包好的应用中有一些可变的因素，比如数据库连接、定时器的触发条件，需要把配置写在外面。
import configparser, os
import cx_Oracle
path = os.path.abspath('..')
cp = configparser.ConfigParser()
# ORCL连接
user = cp['db']['user']
passwd = cp['db']['passwd']
ip = cp['db']['server']
port = cp['db']['port']
SID = cp['db']['sid']
dsn_tns = cx_Oracle.makedsn(ip, port, SID)
def execute(sql):
con = cx_Oracle.connect(user, passwd, dsn_tns)
cur = con.cursor()
cur.execute(sql)
res = cur.fetchall()
con.commit()
cur.close()
con.close()
return res
使用cx_oracle比较麻烦的是需要下载动态库，Linux与windows是一样的，需要在oracle官网下载client，如果提示如下错：
cx_Oracle.DatabaseError: DPI-1047: Oracle Client library cannot be loaded: dlopen(libclntsh.dylib, 1): image not found. See https://oracle.github.io/odpi/doc/installation.html for help
说明可能配置没有配置好，官方的原话：
Add Oracle 12.2, 12.1 or 11.2 client libraries to your operating system library search path such as PATH on Windows or LD_LIBRARY_PATH on Linux. On macOS move the files to ~/lib or /usr/local/lib.
windows配置变量
LD_LIBRARY_PATH;linux或者Mac，新建lib目录，然后把
.dylib的动态库文件使用ln -s链接过去。
参考文章：