环境：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
原文地址：https://www.cnblogs.com/feigerlan/p/9054699.html
时间： 2025-01-13 11:15:44