zabbix 使用微信发送报警信息

说明

本脚本是利用微信的公众平台进行信息推送的发送报警信息的.
可以通过这个链接 微信平台 了解, 可以注册团队,就没有时间限制了.

代码文件

• 需要安装requests模块,当初嫌麻烦,就直接使用的requests.
• 脚本中最后,需要修改成实际的信息.

#!/usr/bin/python
# -*- coding: utf-8 -*-
# Author: Ycan
"""
    微信报警
"""
 
import urllib3
import urllib
import sys
import json
 
 
try:
    import requests
except:
    print 'request `requests` module'
    sys.exit()
 
if sys.getdefaultencoding() != 'utf8':
    reload(sys)
    sys.setdefaultencoding('utf8')
 
requests.packages.urllib3.disable_warnings()
 
 
class WeiXinExcept(Exception):
    def __init__(self, code=400, message=u'Request Error', *args, **kwargs):
        self.code = code
        self.message = message
        self.data = {'code': self.code, 'message': self.message}
        super(WeiXinExcept, self).__init__(*args, **kwargs)
 
    def __str__(self):
        return repr(self.data)
 
 
class WeiXin(object):
    def __init__(self, corpid='', corpsecret=''):
        """
        微信接口
        :param corpid: CorpID是企业号的标识，每个企业号拥有一个唯一的CorpID
        :param corpsecret: Secret是管理组凭证密钥
        """
        self.corpid = corpid
        self.corpsecret = corpsecret
        self.urls = dict(
            token='https://qyapi.weixin.qq.com/cgi-bin/gettoken',
            send_msg='https://qyapi.weixin.qq.com/cgi-bin/message/send',
            department_list='https://qyapi.weixin.qq.com/cgi-bin/department/list',
            user_list='https://qyapi.weixin.qq.com/cgi-bin/user/simplelist',
        )
        self.headers = {"Content-type": "application/x-www-form-urlencoded", "Accept": "text/plain"}
        if not corpid or not corpsecret:
            raise WeiXinExcept(message=u'need CorpID and Secret')
 
        self.token = self.get_token()
 
 
    def get_token(self):
        """
        :return: token
        """
        token_url = self.urls.get('token')
        params = dict(corpid=self.corpid, corpsecret=self.corpsecret)
        _token = requests.get(token_url, params=params)
 
        if _token.ok and _token.json().has_key('access_token'):
            return _token.json()['access_token']
        else:
            raise WeiXinExcept(message=u'get token error!')
 
 
    def get_department(self, department_name=None):
        """
        :param department_name: 部门的名称
        :return:
        """
        url = self.urls.get('department_list')
        params = dict(access_token=self.token)
        r = requests.get(url, params=params)
        if self._check_request_status(r):
            dept_list = r.json().get('department', None)
            if dept_list and department_name is not None:
                return filter(lambda x: x['name'] == department_name.decode('utf8'), dept_list)
            else:
                return dept_list
        else:
            return None
 
 
    def get_department_members(self, department_id, fetch_child=1, status=1):
        """
        :param department_id: 部门ID
        :param fetch_child: 1/0：是否递归获取子部门下面的成员
        :param status: 0获取全部员工，1获取已关注成员列表，2获取禁用成员列表，4获取未关注成员列表。status可叠加
        :return: 返回部门成员的列表
        """
        params = dict(access_token=self.token, department_id=department_id, fetch_child=fetch_child, status=status)
        r = requests.get(self.urls.get('user_list'), params=params)
        if self._check_request_status(r):
            return r.json().get('userlist', None)
        else:
            return None
 
    def send_msg_to_user(self, users=[], paryts=[], agentid='0', msgtype='text', content='', safe='0'):
        """
        `发送信息到用户`
        :param users: type `string` or list ,发送的用户ID,或者用户ID列表
        :param paryts: type `string` or list ,发送的用户组ID,或者用户组ID列表
        :param agentid: 企业应用的id，整型。可在应用的设置页面查看
        :param msgtype: 消息类型，此时固定为：text
        :param content: 消息内容
        :param safe: 表示是否是保密消息，0表示否，1表示是，默认0
        `发送的数据`
        {
           "touser": "UserID1|UserID2|UserID3",
           "toparty": " PartyID1 | PartyID2 ",
           "totag": " TagID1 | TagID2 ",
           "msgtype": "text",
           "agentid": "1",
           "text": {
               "content": "Holiday Request For Pony(http://xxxxx)"
           },
           "safe":"0"
        }
        :return: True or False
        """
 
        if not (users or paryts):
            raise WeiXinExcept(message=u'users or paryts not null.')
 
        touser = '|'.join(users)
        toparty = '|'.join(paryts)
        _params = dict(
            msgtype=msgtype,
            agentid=agentid,
            text=dict(content=content),
            safe=safe
        )
 
        if touser: _params['touser'] = touser
        if toparty: _params['toparty'] = toparty
        data = json.dumps(_params, ensure_ascii=False).encode('utf8')
        url = self.urls.get('send_msg') + '?access_token=%s' % urllib.quote(self.token)
        r = requests.post(url, data=data, headers=self.headers)
        if self._check_request_status(r):
            return True
 
        return r.json() if r.json() else r.raise_for_status()
 
 
    def _check_request_status(self, rq):
        """
        :param request: 请求
        :return: 如果情况没有错误,返回True, 否则返回False
        """
        try:
            if rq.ok and rq.json().get('errcode') == 0:
                return True
            else:
                return False
        except:
            return False
 
 
def sendMsg(users=[], content=''):
    weixin = WeiXin(corpid=CorpID, corpsecret=Secret)
    department_list = weixin.get_department(department_name=department_name)
    if department_list:
        dept_id = department_list[0]['id']
 
        user_list = weixin.get_department_members(dept_id)
        if user_list:
            user_list_id = [x['userid'] for x in user_list]
 
            # 筛选只关注了企业帐号的 userid
            send_user_id = set(users).intersection(set(user_list_id))
 
            out_user_id = set(users).difference(set(user_list_id))
 
            if send_user_id:
                result = weixin.send_msg_to_user(users=send_user_id, content=content)
                if result is not True:
                    print result
 
                if out_user_id:
                    print(u'指定的接收人[%s]的ID没有关注企业帐号.' % ','.join(out_user_id))
            else:
                raise WeiXinExcept(message=u'指定的接收人[%s]的ID没有关注企业帐号.' % ','.join(user_list))
        else:
            raise WeiXinExcept(message=u'部门[%s]下,没有可用的成员列表' % department_name)
    else:
        raise WeiXinExcept(message=u'不能获取到部门[%s]列表,请检查.' % department_name)
 
 
if __name__ == '__main__':
    # 开发者的信息
	# 需要修改成实际的帐号以及agentid信息
    CorpID = 'xx'
    Secret = 'xx'
    # 发送的user在哪个部门
    department_name = 'YunWei'
    # 应用ID
    agentid = 0
 
    if len(sys.argv) < 3:
        help = """
            usge: %s <userid,userid...> <subject> [content]
            """ % sys.argv[0]
        print help
        sys.exit()
 
    userid_list = sys.argv[1].split(',')
    content = '\n'.join(sys.argv[2:])
    sendMsg(users=userid_list, content=content)

测试

907310009为你添加用户时指定的帐号.

# python weixin.py 907310009 "你好,测试微信报警"

怎样查看agentid 和 Secret 等信息

下面用图说明吧,太详细的在后台自己可以琢磨琢磨

后续工作 TODO

• 增加信息发送频率的限制
  这个应用的场景是,在有大量报警的前提下,为了不在短时间范围内一下收到太多短信轰炸,做一个限制多少分钟内发送指定的接收人不超过多少条信息.
  这个功能其实已经用shell处理好了,只是python这段代码没有事件修改.