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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
| #!/usr/bin/env python3
# -*- coding: utf-8 -*-
import time
import threading
import webbrowser
import winreg
from datetime import datetime
from pyzabbix import ZabbixAPI
import sys
import os
import pystray
from PIL import Image
from win10toast import ToastNotifier
class ZabbixAlertMonitor:
def __init__(self, zabbix_url, username, password, web_url=None):
# Zabbix配置
self.zabbix_url = zabbix_url
self.web_url = web_url
self.username = username
self.password = password
self.zapi = None
# 监控状态
self.last_alert_time = None
self.is_running = False
self.monitor_thread = None
# 通知和托盘
self.toaster = ToastNotifier()
self.tray_icon = None
self.icon_path = self._get_absolute_path("_internal/zabbix.ico") # 可自行放置图标文件在脚本目录
self.notification_icon_path = self.icon_path if os.path.exists(self.icon_path) else None
# 自启动配置
self.app_name = "ZabbixAlertMonitor"
self.reg_path = r"Software\Microsoft\Windows\CurrentVersion\Run"
def _get_absolute_path(self, relative_path):
"""将相对路径转换为绝对路径（解决开机自启路径问题）"""
if getattr(sys, 'frozen', False):
# 打包成EXE后的情况（EXE所在目录）
base_path = os.path.dirname(sys.executable)
else:
# 脚本运行的情况（脚本所在目录）
base_path = os.path.dirname(os.path.abspath(__file__))
# 拼接绝对路径
absolute_path = os.path.join(base_path, relative_path)
return absolute_path
def connect_zabbix(self):
"""连接Zabbix API"""
try:
self.zapi = ZabbixAPI(self.zabbix_url)
self.zapi.session.verify = False
self.zapi.login(self.username, self.password)
print(f"成功连接Zabbix服务器")
return True
except Exception as e:
print(f"Zabbix连接失败: {str(e)}")
return False
def get_latest_alerts(self, limit=10):
"""获取最新严重告警"""
try:
if not self.zapi or not self.zapi.auth:
return []
triggers = self.zapi.trigger.get(
output=["description", "priority", "lastchange", "status"],
filter={"value": 1, "priority": 4, "status": 0},
sortfield="lastchange",
sortorder="DESC",
limit=limit,
selectHosts=["host", "name"],
selectItems=["name", "key_"],
expandDescription=True
)
alerts = []
for trigger in triggers:
alert_time = datetime.fromtimestamp(int(trigger['lastchange']))
# 初始化最后告警时间
if self.last_alert_time is None:
self.last_alert_time = alert_time
continue
# 只保留新告警
if alert_time > self.last_alert_time:
host = trigger['hosts'][0] if trigger['hosts'] else {}
host_name = host.get('name', host.get('host', "未知主机"))
alert_info = {
'description': trigger['description'],
'priority': self.get_priority_text(trigger['priority']),
'host': host_name,
'time': alert_time.strftime("%Y-%m-%d %H:%M:%S"),
'timestamp': alert_time
}
alerts.append(alert_info)
self.last_alert_time = alert_time
return alerts
except Exception as e:
print(f"获取告警失败: {str(e)}")
return []
def get_priority_text(self, priority_code):
"""优先级代码转文本"""
priority_map = {
'0': '未分类', '1': '信息', '2': '警告',
'3': '一般严重', '4': '严重', '5': '灾难'
}
return priority_map.get(str(priority_code), '未知')
def show_notification(self, alert):
"""显示桌面通知（修复阻塞问题）"""
try:
# 由于设置了 threaded=True，通知会在后台线程显示，不需要等待
self.toaster.show_toast(
title=f"Zabbix告警 - {alert['priority']}",
msg=f"主机: {alert['host']}\n描述: {alert['description']}\n时间: {alert['time']}",
duration=10,
icon_path=self.notification_icon_path,
threaded=True # 关键：后台线程显示通知，不阻塞主线程
)
print(f"已触发告警通知: {alert['description']}")
except Exception as e:
print(f"通知显示失败: {str(e)}")
def monitor_loop(self, check_interval=10):
"""监控循环"""
print(f"开始监控（间隔{check_interval}秒）")
while self.is_running:
try:
# 检查连接
if not self.zapi or not self.zapi.auth:
print("重连Zabbix...")
self.connect_zabbix()
time.sleep(5)
continue
# 获取告警
alerts = self.get_latest_alerts()
if alerts:
print(f"发现{len(alerts)}个新告警")
for alert in alerts:
self.show_notification(alert)
time.sleep(0.5) # 缩短间隔，避免多个通知排队阻塞
else:
print(f"{datetime.now().strftime('%H:%M:%S')} - 无新告警")
time.sleep(check_interval)
except Exception as e:
print(f"监控错误: {str(e)}")
time.sleep(30)
# ------------------------------
# 开机自启功能
# ------------------------------
def is_auto_start(self):
"""检查是否已启用自启"""
try:
key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, self.reg_path, 0, winreg.KEY_READ)
winreg.QueryValueEx(key, self.app_name)
winreg.CloseKey(key)
return True
except FileNotFoundError:
return False
except Exception as e:
print(f"检查自启失败: {str(e)}")
return False
def set_auto_start(self, enable=True):
"""设置开机自启"""
try:
if enable:
# 启用自启：添加注册表项
exe_path = sys.executable
script_path = os.path.abspath(__file__)
# 构建启动命令
start_cmd = f'"{exe_path}"'
key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, self.reg_path, 0, winreg.KEY_WRITE)
winreg.SetValueEx(key, self.app_name, 0, winreg.REG_SZ, start_cmd)
winreg.CloseKey(key)
print(f"已启用开机自启")
return True
else:
# 禁用自启：删除注册表项
key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, self.reg_path, 0, winreg.KEY_WRITE)
winreg.DeleteValue(key, self.app_name)
winreg.CloseKey(key)
print(f"已禁用开机自启")
return True
except PermissionError:
print("权限不足！请以管理员身份运行")
self.show_notification({
'priority': '警告', 'host': '本地',
'description': '设置自启失败：需要管理员权限',
'time': datetime.now().strftime("%Y-%m-%d %H:%M:%S")
})
return False
except Exception as e:
print(f"设置自启失败: {str(e)}")
return False
# ------------------------------
# 托盘菜单功能
# ------------------------------
def open_zabbix_web(self):
"""打开Zabbix网页"""
try:
print(f"打开Zabbix网页: {self.web_url}")
webbrowser.open_new(self.web_url)
except Exception as e:
print(f"打开网页失败: {str(e)}")
def toggle_auto_start(self):
"""切换自启状态"""
self.set_auto_start(not self.is_auto_start())
self.update_tray_menu()
def exit_app(self):
"""退出应用"""
print("\n 退出应用...")
self.is_running = False
if self.tray_icon:
self.tray_icon.stop()
# 停止toaster的后台线程（避免残留）
self.toaster.stop()
try:
sys.exit(0)
except:
pass
def create_tray_menu(self):
"""创建托盘菜单"""
# 自启状态标记
if self.is_auto_start():
auto_end_mark = "[已开启]"
else:
auto_end_mark = "[已关闭]"
# 创建菜单（使用更简单的事件绑定）
return pystray.Menu(
pystray.MenuItem(
"打开Zabbix网页",
lambda icon, item: self.open_zabbix_web(),
default=True # 左键点击触发
),
pystray.MenuItem(
f"开机自启 {auto_end_mark}",
lambda icon, item: self.toggle_auto_start()
),
pystray.MenuItem(
"退出监控",
lambda icon, item: self.exit_app()
)
)
def update_tray_menu(self):
"""更新托盘菜单"""
if self.tray_icon:
self.tray_icon.menu = self.create_tray_menu()
def init_tray(self):
"""初始化托盘图标"""
icon_path = self.notification_icon_path
# 图标文件不存在时的降级处理
if not os.path.exists(icon_path):
print(f"警告：图标文件 {icon_path} 不存在，使用默认图标")
# 创建一个简单的默认图标（红色圆形）
icon = Image.new('RGB', (64, 64), color='red')
else:
icon = Image.open(icon_path)
# 创建托盘图标
self.tray_icon = pystray.Icon(
name=self.app_name,
icon=icon,
title="Zabbix告警监控",
menu=self.create_tray_menu()
)
# ------------------------------
# 主控制方法
# ------------------------------
def start_monitor(self, check_interval=10):
"""启动监控线程"""
self.is_running = True
self.monitor_thread = threading.Thread(
target=self.monitor_loop,
args=(check_interval,),
daemon=True
)
self.monitor_thread.start()
def run(self, check_interval=10):
"""启动应用"""
# 1. 连接Zabbix
if not self.connect_zabbix():
print("连接Zabbix失败，程序即将退出")
time.sleep(5)
sys.exit(1)
# 2. 启动监控
self.start_monitor(check_interval)
# 3. 初始化并运行托盘（阻塞主线程）
self.init_tray()
print("托盘图标已启动，右键操作菜单，左键打开网页")
self.tray_icon.run()
def main():
# 创建实例并运行
monitor = ZabbixAlertMonitor(
zabbix_url=ZABBIX_CONFIG['url'],
username=ZABBIX_CONFIG['username'],
password=ZABBIX_CONFIG['password'],
web_url=ZABBIX_CONFIG['web_url']
)
monitor.run(check_interval=10)
if __name__ == "__main__":
# Zabbix配置（请根据实际情况修改）
ZABBIX_CONFIG = {
'url': 'http://*.*.*.*/api_jsonrpc.php',
'username': 'Admin',
'password': 'zabbix',
'web_url': 'http://*.*.*.*/'
}
try:
main()
except KeyboardInterrupt:
print("\n 程序被中断")
sys.exit(0)
except Exception as e:
print(f"\n 程序异常: {str(e)}")
time.sleep(5)
sys.exit(1)
|