Python利用Zabbix API定时报告存在报警的机器

简介:

由于手上的zabbix服务器有几个,有时会没注意报警短信,又懒得登陆上去看看zabbix,所以用Python利用zabbix的API获取有问题的机器每天早上发到我邮箱,foxmail又是我天天都用的东西,这样我就可以方便地看到所有有问题的机器了

zabbix_sendmail.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
#!/usr/bin/python
#coding:utf-8
import  smtplib
from  email.mime.text  import  MIMEText
import  sys
#mail_to_list = ['lihuipeng@xxx.com',]
mail_host  =  'smtp.163.com'
mail_user  =  'testest'      #发件人邮箱帐号
mail_pass  =  '123456'       #发件人邮箱密码
mail_postfix  =  '163.com'
def  send_mail(to_list,subject,content):
     me  =  mail_user + "<" + mail_user + "@" + mail_postfix + ">"
     msg  =  MIMEText(content)
     msg[ 'Subject' =  subject
     msg[ 'From' =  me
     msg[ 'to' =  ";" .join(to_list)
     #msg['to'] = to_list
                                                                                             
     try :
         =  smtplib.SMTP()
         s.connect(mail_host)
         s.login(mail_user,mail_pass)
         s.sendmail(me,to_list,msg.as_string())
         s.close()
         return  True
     except  Exception,e:
         print  str (e)
         return  False
                                                                                             
if  __name__  = =  "__main__" :
     print  sys.argv[ 1 ]
     print  sys.argv[ 2 ]
     print  sys.argv[ 3 ]
     send_mail(sys.argv[ 1 ], sys.argv[ 2 ], sys.argv[ 3 ])

这个是我用来发邮件的脚本,原理就是连接到163邮箱发送邮件,这里我是注册了一个帐号专门用来发邮件的。


zabbix_report.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
#!/usr/bin/python
#coding:utf-8
import  json
import  urllib2
from  urllib2  import  URLError
import  sys
import  zabbix_sendmail
#接收人
mailtolist  =  [ 'test@163.com' ,]
#格式:zabbix地址,zabbix帐号,zabbix密码,邮件标题
zabbix_addresses = [ 'http://test1.zabbix.com,admin,123456,test1' , 'http://test2.zabbix.com,admin,123456,test2' , 'http://test3.zabbix.com,admin,123456,test3' ]
class  ZabbixTools:
     def  __init__( self ,address,username,password):
                                                                       
         self .address  =  address
         self .username  =  username
         self .password  =  password
                                                                       
         self .url  =  '%s/api_jsonrpc.php'  %  self .address
         self .header  =  { "Content-Type" : "application/json" }
                                                                       
                                                                       
                                                                       
     def  user_login( self ):
         data  =  json.dumps({
                            "jsonrpc" "2.0" ,
                            "method" "user.login" ,
                            "params" : {
                                       "user" self .username,
                                       "password" self .password
                                       },
                            "id" 0
                            })
                                                                       
         request  =  urllib2.Request( self .url, data)
         for  key  in  self .header:
             request.add_header(key,  self .header[key])
                                                                   
         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()
             #print response['result']
             self .authID  =  response[ 'result' ]
             return  self .authID
                                                                           
     def  trigger_get( self ):
         data  =  json.dumps({
                            "jsonrpc" : "2.0" ,
                            "method" : "trigger.get" ,
                            "params" : {
                                       "output" : [
                                                 "triggerid" ,
                                                 "description" ,
                                                 "priority"
                                                 ],
                                       "filter" : {
                                                  "value" 1
                                                  },
                                       "expandData" : "hostname" ,
                                       "sortfield" "priority" ,
                                       "sortorder" "DESC"
                                     },
                            "auth" self .user_login(),
                            "id" : 1              
         })
                                                                       
         request  =  urllib2.Request( self .url, data)
         for  key  in  self .header:
             request.add_header(key,  self .header[key])
                                                                       
         try :
             result  =  urllib2.urlopen(request)
         except  URLError as e:
             print  "Error as " , e
         else :
             response  =  json.loads(result.read())
             result.close()
             issues  =  response[ 'result' ]
             content  =  ''
             if  issues:
                 for  line  in  issues:
                     content  =  content  +  "%s:%s\r\n"  %  (line[ 'host' ],line[ 'description' ])
             return  content
                                                                           
if  __name__  = =  "__main__" :
     for  zabbix_addres  in  zabbix_addresses:
         address,username,password,subject  =  zabbix_addres.split( ',' )
         =  ZabbixTools(address = address, username = username, password = password)
         content  =  z.trigger_get()
         zabbix_sendmail.send_mail(mailtolist, subject, content)
     print  "Done!"

这里就是利用API查询信息,然后通过上面的发邮件脚本把信息发出来,效果是这样的:

145315337.png

再添加一个crontab就OK~

本文转自运维笔记博客51CTO博客,原文链接http://blog.51cto.com/lihuipeng/1306261如需转载请自行联系原作者


lihuipeng

相关文章
|
2天前
|
机器学习/深度学习 算法 数据挖掘
机器学习--K近邻算法,以及python中通过Scikit-learn库实现K近邻算法API使用技巧
机器学习--K近邻算法,以及python中通过Scikit-learn库实现K近邻算法API使用技巧
|
5天前
|
缓存 前端开发 API
toapi,一个强大的 Python Web API库!
toapi,一个强大的 Python Web API库!
20 5
|
7天前
|
API Python
[AIGC] Python列表([])和字典({})常用API介绍
[AIGC] Python列表([])和字典({})常用API介绍
|
7天前
|
API Python
[AIGC] Python字符串常用API介绍
[AIGC] Python字符串常用API介绍
|
7天前
|
API Python
[AIGC] 使用Python刷LeetCode:常用API及技巧指南
[AIGC] 使用Python刷LeetCode:常用API及技巧指南
|
7天前
|
测试技术 API 网络架构
Python的api自动化测试 编写测试用例
【4月更文挑战第18天】使用Python进行API自动化测试,可以结合`requests`库发送HTTP请求和`unittest`(或`pytest`)编写测试用例。以下示例: 1. 安装必要库:`pip install requests unittest` 2. 创建`test_api.py`,导入库,定义基础URL。 3. 创建继承自`unittest.TestCase`的测试类,包含`setUp`和`tearDown`方法。 4. 编写测试用例,如`test_get_users`,检查响应状态码和内容。 5. 运行测试:`python -m unittest test_api.py`
13 2
|
7天前
|
JSON 测试技术 API
Python的Api自动化测试使用HTTP客户端库发送请求
【4月更文挑战第18天】在Python中进行HTTP请求和API自动化测试有多个库可选:1) `requests`是最流行的选择,支持多种请求方法和内置JSON解析;2) `http.client`是标准库的一部分,适合需要低级别控制的用户;3) `urllib`提供URL操作,适用于复杂请求;4) `httpx`拥有类似`requests`的API,提供现代特性和异步支持。根据具体需求选择,如多数情况`requests`已足够。
13 3
|
7天前
|
Java 测试技术 API
Python的api自动测试选择合适的测试框架
【4月更文挑战第18天】在Python API自动测试中,选择合适的框架至关重要。常见的测试工具有unittest(集成度高,适合基础测试)、pytest(功能强大,支持插件扩展和高级功能)、requests-mock(用于HTTP请求模拟和断言)、rest-assured(针对RESTful API的简洁测试)以及allure-pytest(生成美观的测试报告)。选择时要考虑项目需求、团队熟悉度和社区支持。确保遵循良好测试实践,编写清晰、全面的测试用例。
10 2
|
8天前
|
人工智能 机器人 API
【Python+微信】【企业微信开发入坑指北】3. 如何利用企业微信API给微信群推送消息
【Python+微信】【企业微信开发入坑指北】3. 如何利用企业微信API给微信群推送消息
11 0
|
8天前
|
缓存 人工智能 API
【Python+微信】【企业微信开发入坑指北】2. 如何利用企业微信API主动给用户发应用消息
【Python+微信】【企业微信开发入坑指北】2. 如何利用企业微信API主动给用户发应用消息
9 0

推荐镜像

更多