python开发微信企业应用-报警程序

为了熟悉python,记使用python开发一个简单的微信报警后台的过程,主要完成了微信企业应用中进行文字推送。

  1. 使用urllib3发送REST(POST,GET)请求;
  2. 使用Apscheduler定时任务;
  3. 使用pyinstaller打包应用;
  4. 使用logging产生日志;
  5. 使用configparser读取ini配置文件中的属性与参数;
  6. 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!!!')
  1. 使用urllib3的GET方法,定义了一个获取微信token的方法getToken,需要传输企业微信应用的两个参数corpidcorpsecret

  2. 使用json模块对数据进行解析;

  3. 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('消息发送失败!')
  1. 常见的获取python对象的Attribute(属性)的方法:

    方法一:

    [item['userid'] for item in USERLIST]   
    

    方法二:

    attrList = map(lambda x: x.attr, objectList)
    
  2. 使用join方法,组合字符串;

  3. 加入`try…except’防止异常抛出;

参考文章:

Python——深入理解urllib、urllib2及requests(requests不建议使用?)

如何使用urllib3发送POST带参请求

Python–urllib3库详解1

Python中的json解析

python解析url返回的json格式数据

定时任务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()
  1. 使用Blocking阻塞的方式创建了scheduler;

  2. minHourJob加入定时任务,设置每5秒执行一次;

  3. 使用datetime.timedelta对时间进行运算,strftime格式化时间;

参考文章:

Python定时任务的实现方式

Python下APScheduler的快速指南

python 实现日期加1天或减少一天

加入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)

参考文章:

python logging模块使用教程

使用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

参考文章:

PYINSTALLER打包PYTHON脚本的一些心得

pip国内镜像源的配置

修改PyPI源

使用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.

  1. windows配置变量LD_LIBRARY_PATH;

  2. linux或者Mac,新建lib目录,然后把.dylib的动态库文件使用ln -s链接过去。

参考文章:

Python ConfigParser模块常用方法示例

Python 读取写入配置文件 —— ConfigParser

cx_Oracle 6 Installation

cx_Oracle - Oracle client on Mac OS X - RPATH issue

-------------【本文完】一路前行感谢有你【本文完】-------------
  • 本文作者: var author='wayearn'
  • 文章标题: python开发微信企业应用-报警程序
  • 创建时间: 2017-12-20 09:52:26
  • 本文链接: https://www.wayearn.com/python开发微信报警程序/
  • 版权声明: 本博客所有文章除特别声明外,均采用 BY-NC-SA 许可协议。转载请注明出处!
0%