Python调用zabbix api 接口
#!/usr/bin/env python# -*- coding: utf-8 -*-
import json
import urllib2
import sys
class zabbixtools:
def __init__(self):
self.url = "http://172.16.4.10/zabbix/api_jsonrpc.php"
self.header = {"Content-Type": "application/json"}
self.authID = self.user_login()
def user_login(self):
data = json.dumps(
{
"jsonrpc": "2.0",
"method": "user.login",
"params": {
"user": "luojl",
"password": "luojl"
},
"id": 0
}
)
request = urllib2.Request(self.url,data)
for key in self.header:
request.add_header(key,self.header)
try:
result = urllib2.urlopen(request)
except URLError as e:
print "Auth Failed, Please Check Your Name And Password:",e.code
else:
response = json.loads(result.read())
result.close()
authID = response['result']
return authID
def get_data(self,data,hostip=""):
request = urllib2.Request(self.url,data)
for key in self.header:
request.add_header(key,self.header)
try:
result = urllib2.urlopen(request)
except URLError as e:
if hasattr(e, 'reason'):
print 'We failed to reach a server.'
print 'Reason: ', e.reason
elif hasattr(e, 'code'):
print 'The server could not fulfill the request.'
print 'Error code: ', e.code
return 0
else:
response = json.loads(result.read())
result.close()
return response
def get_hostlist(self):
data = json.dumps(
{
"jsonrpc":"2.0",
"method":"host.get",
"params":{
"output":["hostid","name","status","host"],
"filter":{"host":""}
},
"auth":self.authID,
"id":2
}
)
request = urllib2.Request(self.url,data)
for key in self.header:
request.add_header(key,self.header)
try:
result = urllib2.urlopen(request)
except URLError as e:
if hasattr(e, 'reason'):
print 'We failed to reach a server.'
print 'Reason: ', e.reason
elif hasattr(e, 'code'):
print 'The server could not fulfill the request.'
print 'Error code: ', e.code
else:
response = json.loads(result.read())
result.close()
print "Number Of Hosts: ", len(response['result'])
for host in response['result']:
print "Host ID:",host['hostid'],"Host Name:",host['name']
def host_get(self,hostip):
data = json.dumps(
{
"jsonrpc": "2.0",
"method": "host.get",
"params": {
"output": ["hostid","name","status","host"],
"filter": {"host": }
},
"auth": self.authID,
"id": 1
}
)
res = self.get_data(data)['result']
if (res !=0) and (len(res) != 0):
host = res
if host['status'] == '1':
print "\t","\033.ljust(15),'\t',"\033.encode('UTF-8'),'\t',"\033[1;31;40m%s\033[0m" % u'未在监控状态'.encode('UTF-8')
return host['hostid']
elif host['status'] == '0':
print "\t","\033.ljust(15),'\t',"\033.encode('UTF-8'),'\t',"\033[1;32;40m%s\033[0m" % u'在监控状态'.encode('UTF-8')
return host['hostid']
else:
print '\t',"\033[1;31;40m%s\033[0m" % "Get Host Error or cannot find this host,please check !"
return 0
def abc(self):
hostip = raw_input("\033[1;35;40m%s\033[0m" % 'Enter Your Check Host:Host_ip :')
a = self.host_get(hostip)
print a
def hostgroup_get(self):
data = json.dumps(
{
"jsonrpc": "2.0",
"method": "hostgroup.get",
"params": {
"output": "extend"
},
"auth": self.authID,
"id": 3,
}
)
res = self.get_data(data)
if 'result' in res.keys():
res = res['result']
if (res !=0) or (len(res) !=0):
print "\033[1;32;40m%s\033[0m" % "Number Of Group: ", "\033[1;31;40m%d\033[0m" % len(res)
for host in res:
print "\t","HostGroup_id:",host['groupid'],"\t","HostGroup_Name:",host['name'].encode('UTF-8')
else:
print "Get HostGroup Error,please check !"
def template_get(self):
data = json.dumps(
{
"jsonrpc": "2.0",
"method": "template.get",
"params": {
"output": "extend",
},
"auth": self.authID,
"id": 4,
}
)
res = self.get_data(data)
if 'result' in res.keys():
res = res['result']
if (res !=0) or (len(res) !=0):
print "\033[1;32;40m%s\033[0m" % "Number Of Template: ", "\033[1;31;40m%d\033[0m" % len(res)
for host in res:
print "\t","Template_id:",host['templateid'],"\t","Template_Name:",host['name'].encode('UTF-8')
else:
print "Get Template Error,please check !"
if __name__ == "__main__":
test = zabbixtools()
argv = sys.argv.strip()
if argv == 'hostlist':
test.get_hostlist()
elif argv == 'grouplist':
test.hostgroup_get()
elif argv == 'templist':
test.template_get()
else:
test.abc()
页:
[1]