通过itchat调用zabbixAPI实现微信确认zabbix告警
Posted feigerlan
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了通过itchat调用zabbixAPI实现微信确认zabbix告警相关的知识，希望对你有一定的参考价值。
环境：python3.6.1 ,pip 9.0.1
1，通过itchat,实现微信的扫码登录，和关键定识别
#coding=utf-8 import urllib.request, urllib.error, urllib.parse import sys import json import argparse from login import * from function import * ######################################################################## import itchat import time import re import requests import threading import shutil import os ‘‘‘ reload(sys) sys.setdefaultencoding(‘utf-8‘) ‘‘‘ self_name=‘PH‘ @itchat.msg_register(itchat.content.TEXT) def cmd(revmsg): msg = revmsg[‘Text‘] if msg == "注销登录": quit() print("已注销") # elif msg == "检索事件": # getevents() elif str.isdigit("".join(re.findall(r‘查询事件(\d+)‘,msg))): print(msg) eventid = "".join(re.findall(r‘查询事件(\d+)‘,msg)) print(eventid) itchat.send(‘识别到事件id，开始查询事件‘+eventid,toUserName = revmsg[‘FromUserName‘]) getevent(eventid,revmsg[‘FromUserName‘]) elif str.isdigit("".join(re.findall(r‘确认事件(\d+)‘,msg))): eventid = "".join(re.findall(r‘确认事件(\d+)‘,msg)) #messages = "".join(re.findall(r‘确认消息<(.+)>‘,msg)) #有确认消息，不关闭问题 if "".join(re.findall(r‘(确认消息)<.+>‘,msg)) == "确认消息" and "".join(re.findall(r‘【关闭】‘,msg)) == "": messages = "".join(re.findall(r‘确认消息<(.+)>‘,msg)) itchat.send("识别到事件id，开始确认事件"+eventid+"，确认消息为:"+messages,toUserName = revmsg[‘FromUserName‘]) ackevent(eventid,revmsg[‘FromUserName‘],messages) #无确认消息，关闭问题 elif "".join(re.findall(r‘(确认消息)<.+>‘,msg)) == "" and "".join(re.findall(r‘【关闭】‘,msg)) == "【关闭】": itchat.send("识别到事件id，开始确认事件"+eventid+"，无确认消息。关闭问题。",toUserName = revmsg[‘FromUserName‘]) ackevent(eventid,revmsg[‘FromUserName‘],action=1) #有确认消息，关闭问题 elif "".join(re.findall(r‘(确认消息)<.+>‘,msg)) == "确认消息" and "".join(re.findall(r‘【关闭】‘,msg)) == "【关闭】": messages = "".join(re.findall(r‘确认消息<(.+)>‘,msg)) itchat.send("识别到事件id，开始确认事件"+eventid+"，确认消息为:"+messages+"。关闭问题。",toUserName = revmsg[‘FromUserName‘]) ackevent(eventid,revmsg[‘FromUserName‘],messages,1) #无确认消息，不关闭问题。 else: print("识别到事件id:"+eventid) itchat.send("识别到事件id，开始确认事件:"+eventid,toUserName = revmsg[‘FromUserName‘]) ackevent(eventid,revmsg[‘FromUserName‘]) elif msg == "查询告警": gettrigger(revmsg[‘FromUserName‘]) print("查询问题触发器") # elif msg == "告警": # gaojing() # print("执行告警") else: welcome = "您好，欢迎使用zabbix微信告警系统！你可以回复关键字\n(查询告警、确认事件、查询事件等)实现功能\n如：确认事件12345,确认消息<已解决>【关闭】（确认消息以“<>”为分隔符）\n如：查询事件1234\n更多功能正在研发中，敬请期待！" itchat.send(welcome,toUserName = revmsg[‘FromUserName‘]) ####################################### ####################################### def quit(): itchat.logout() ####################################### def gaojing(fromuser): users = itchat.search_friends(name=u‘Feiger‘) userName = users[0][‘UserName‘] #登陆zabbix获取auth auth = authenticate(url, username, password) #状态0是启用监控，1是禁用监控 status=1 #定义操作ip hostip=‘192.168.15.139‘ #通过hostip获取zabbix hostids=ipgetHostsid(hostip,url,auth) hostid=hostids[0][‘hostid‘] alerts=actionidgetalert(url,auth) xiaoxi=str(alerts[0]) itchat.send(xiaoxi,toUserName = userName) def gettrigger(fromuser): #获取当前触发器 auth = authenticate(url, username, password) trigerIDs = gettrigetID(url,auth) #获取触发器列表 for trigerIDresault in trigerIDs: #遍历触发器 eventlist = triggergetevents(trigerIDresault[‘triggerid‘],url,auth) #获取事件列表 #for event in eventlist: #遍历事件 # if event[‘acknowledged‘] == ‘0‘ and event[‘value‘] == ‘0‘: # print(event) # print(event[‘eventid‘]) # else: # continue xiaoxi=str(trigerIDs) itchat.send(xiaoxi,toUserName = fromuser) def ackevent(eventid,fromuser,message="已确认(微信默认消息)",action=0): #确认事件 auth = authenticate(url, username, password) #执行事件确认操作,并返回已确认事件的对象 eventobj=eventackknowledge(eventid,url,auth,message,action) #调取确认事件的方法 if ‘eventids‘ in eventobj: #eventobj这事件ID的字典 itchat.send(str(eventobj[‘eventids‘])+"事件确认成功",toUserName = fromuser) else: itchat.send("确认事件失败："+eventobj,toUserName = fromuser) def getevent(eventid,fromuser): auth = authenticate(url, username, password) event=eventget(eventid,url,auth) #查询事件 itchat.send(str(event),toUserName = fromuser) def getevents(): #通过时间获取事件 auth = authenticate(url, username, password) users = itchat.search_friends(name=u‘Feiger‘) userName = users[0][‘UserName‘] events = timegetevents("1523030400","1523116800",url,auth) print(events) #xiaoxi=str(events[0]) #itchat.send(xiaoxi,toUserName = userName) ####################################### if __name__ == ‘__main__‘: itchat.auto_login(enableCmdQR=2) itchat.run()
2，zabbix登录逻辑
#!/usr/bin/env python #coding:utf-8 import urllib.request, urllib.error, urllib.parse import json #定义URL账户密码 url = ‘http://192.168.165.179:8027/api_jsonrpc.php‘ username = ‘admin‘ password = ‘zabbix‘ #定义通过HTTP方式访问API地址的函数，后面每次请求API的各个方法都会调用这个函数 def requestJson(url,values): data = json.dumps(values) data=bytes(data,‘utf8‘) req = urllib.request.Request(url, data, {‘Content-Type‘: ‘application/json-rpc‘}) response = urllib.request.urlopen(req, data) output = json.loads(response.read()) print(output) try: message = output[‘result‘] except: message = output[‘error‘][‘data‘] print(message) return message #API接口认证的函数，登录成功会返回一个Token def authenticate(url, username, password): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: ‘user.login‘, ‘params‘: { ‘user‘: username, ‘password‘: password }, ‘id‘: ‘0‘ } idvalue = requestJson(url,values) return idvalue
3，zabbixAPI，此段代码也是网上找到的，自己看了Zabbix的官方文档添加了些许功能。理论上来说可以通过微信触发关键字的方式实现一切对zabbix的操作。
#coding=utf-8 import sys import argparse import urllib.request, urllib.error, urllib.parse import json from login import * #定义检索问题状态下所有触发器的ID，名称和严重性，并按降序对其进行严重性排序。 def gettrigetID(url,auth): values = { "jsonrpc" : "2.0" , "method" : "trigger.get" , "params" : { "output" : [ "triggerid" , "description" , "priority" ] , "filter" : { "value" : 1 } , "sortfield" : "priority" , "sortorder" : "DESC" } , "auth" : auth , "id" : 1 } output = requestJson(url,values) return output #触发器检索事件,返回一系列以事件为元素的列表。 def triggergetevents(triggerID,url,auth): values = { "jsonrpc": "2.0", "method": "event.get", "params": { "output": ["eventid","objectid"], "select_acknowledges":"extend", #"output": "extend", "objectids": triggerID, "sortfield": ["clock", "eventid"], "sortorder": "DESC" }, "auth": auth, "id": 1 } output = requestJson(url,values) return output #按时间检索事件 def timegetevents(time_from,time_till,url,auth): values = { "jsonrpc": "2.0", "method": "event.get", "params": { "output": "extend", "time_from": time_from, "time_till": time_till, "sortfield": ["clock", "eventid"], "sortorder": "desc" }, "auth": auth, "id": 1 } output = requestJson(url,values) return output #通过eventID查询事件 def eventget(eventid,url,auth): values = { "jsonrpc": "2.0", "method": "event.get", "params": { "output": ["eventid","acknowledged","objectid"], "select_acknowledges":["eventid","message","action","alias"], "eventids": eventid, }, "auth": auth, "id": 1 } output = requestJson(url,values) return output #定义确认事件方法 def eventackknowledge(eventid,url,auth,message="已确认(微信默认确认消息)",action=0): values = { ‘jsonrpc‘ : ‘2.0‘, ‘method‘ : ‘event.acknowledge‘, ‘params‘ : { ‘eventids‘ : eventid, ‘message‘ : message, ‘action‘ : action }, ‘auth‘ : auth, ‘id‘ : 1 } print("调用了eventacknowledge方法,正在确认事件：%s"%(eventid)) output = requestJson(url,values) if ‘eventids‘ in output: print("确认事件成功") else: print("确认事件失败") return output #通过动作ID检索警报 def actionidgetalert(url,auth): values = { "jsonrpc": "2.0", "method": "alert.get", "params": { "output": "extend", "actionids": ‘7‘ }, "auth":auth, "id": 1 } output = requestJson(url,values) return output #定义更新action函数 def mediatypeupdate(mediatypeid,status,auth): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: ‘mediatype.update‘, ‘params‘: { "mediatypeid": mediatypeid, "status": status }, ‘auth‘: auth, ‘id‘: ‘1‘ } output = requestJson(url,values) #定义读取状态函数 def triggerget(auth): values = {‘jsonrpc‘: ‘2.0‘, "method":"trigger.get", "params": { "output": [ "triggerid", "description", "priority" ], "filter": { "value": 1 }, "expandData":"hostname", "sortfield": "priority", "sortorder": "DESC" }, ‘auth‘: auth, ‘id‘: ‘2‘ } output = requestJson(url,values) return output #定义通过ip获取主机id的函数 def ipgetHostsid(ip,url,auth): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: ‘host.get‘, ‘params‘: { ‘output‘: [ "host" ], ‘filter‘: { ‘ip‘: ip }, }, ‘auth‘: auth, ‘id‘: ‘3‘ } output = requestJson(url,values) return output #定义通过主机id获取开启关闭监控函数 def idupdatehost(status,hostid,url,auth): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: ‘host.update‘, ‘params‘: { "hostid": hostid, "status": status }, ‘auth‘: auth, ‘id‘: ‘4‘ } output = requestJson(url,values) return output #定义通过项目hostid获取itemid函数 def getHostsitemsid(hostid,itemsname,url,auth): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: "item.get", "params": { "output": ["itemids"], "hostids": hostid, "filter":{ "key_": itemsname, }, }, ‘auth‘: auth, ‘id‘: ‘5‘ } output = requestJson(url,values) if len(output)==0: return output else: return output[0][‘itemid‘] #定义通过项目id获取监控项目最近值信息的函数 def getHostsitemsvalue(itemid,url,auth): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: "history.get", "params": { "output": "extend", "history":3, "itemids":itemid, "sortfield": "clock", "sortorder": "DESC", "limit":1, }, ‘auth‘: auth, ‘id‘: ‘6‘ } output = requestJson(url,values) if len(output)==0: return output else: return output[0]["value"] #定义更新读取状态action函数 def mediatypeget(mediatypeid,auth): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: ‘mediatype.get‘, ‘params‘: { "output": "extend", "filter": { "mediatypeid":mediatypeid, }, }, ‘auth‘: auth, ‘id‘: ‘7‘ } output = requestJson(url,values) if len(output)==0: return output else: return output[0][‘status‘] #定义maintenance维修模式host函数 def maintenancecreate(maintenancename,active_since,active_till,hostid,auth): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: ‘maintenance.create‘, ‘params‘: { "name": maintenancename, "active_since": active_since, "active_till": active_till, "hostids": [ hostid ], "timeperiods": [ { "timeperiod_type": 0, "every": 1, "dayofweek": 64, "start_time": 64800, "period": 3600 } ] }, ‘auth‘: auth, ‘id‘: ‘8‘ } output = requestJson(url,values) #定义通过模糊获取关闭主机信息函数 def disabledhostget(url,auth): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: ‘host.get‘, "params": { "output": ["host"], ‘selectInterfaces‘: [ "ip" ], "filter": { "status":1 } }, ‘auth‘: auth, ‘id‘: ‘9‘ } output = requestJson(url,values) return output #定义maintenance维修模式group函数 def maintenancecreategroup(maintenancename,active_since,active_till,groupid,auth): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: ‘maintenance.create‘, ‘params‘: { "name": maintenancename, "active_since": active_since, "active_till": active_till, "groupids": [ groupid ], "timeperiods": [ { "timeperiod_type": 0, "every": 1, "dayofweek": 64, "start_time": 64800, "period": 3600 } ] }, ‘auth‘: auth, ‘id‘: ‘10‘ } output = requestJson(url,values) #定义通过host groups named 获取groupid def groupnameGroupid(groupname,auth): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: ‘hostgroup.get‘, "params": { "output": "extend", "filter": { "name": [ groupname ] } }, ‘auth‘: auth, ‘id‘: ‘11‘ } output = requestJson(url,values) return output #定义模糊查询维护主机 def maintenanceget(url,auth): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: ‘maintenance.get‘, "params": { "output": "extend", }, ‘auth‘: auth, ‘id‘: ‘12‘ } output = requestJson(url,values) return output #定义批量恢复处于维护主机 def maintenancedelete(maintenanceid,url,auth): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: ‘maintenance.delete‘, "params": [ maintenanceid ], ‘auth‘: auth, ‘id‘: ‘13‘ } output = requestJson(url,values) return output #定义通过hostid获取graphid的函数 def getgraphid(hostid,graphname,url,auth): values = {‘jsonrpc‘: ‘2.0‘, ‘method‘: ‘graph.get‘, ‘params‘: { "output": "name", "hostids": hostid, "sortfield": "name", ‘filter‘: { "name": graphname }, }, ‘auth‘: auth, ‘id‘: ‘14‘ } output = requestJson(url,values) return output
09:57:03
以上是关于通过itchat调用zabbixAPI实现微信确认zabbix告警的主要内容，如果未能解决你的问题，请参考以下文章