zabbix本地监控程序

此脚本常驻系统右下角，当有告警会弹出通知。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import json
import requests
import time
import threading
from datetime import datetime
from plyer import notification
from pyzabbix import ZabbixAPI
import sys
import os
import pystray
from PIL import Image

class ZabbixAlertMonitor:
    def __init__(self, zabbix_url, username, password):
        self.zabbix_url = zabbix_url
        self.username = username
        self.password = password
        self.zapi = None
        self.last_alert_time = None
        self.monitor_thread = None  # 监控线程
        self.is_running = False     # 运行状态标志（控制线程和托盘）
        self.tray_icon = None       # 托盘图标实例

    def connect_zabbix(self):
        """连接到Zabbix API"""
        try:
            self.zapi = ZabbixAPI(self.zabbix_url)
            self.zapi.login(self.username, self.password)
            print(f"成功连接到Zabbix服务器: {self.zabbix_url}")
            return True
        except Exception as e:
            print(f"连接Zabbix失败: {e}")
            return False

    def get_latest_alerts(self, limit=10):
        """获取最新的告警信息（仅严重级）"""
        try:
            # 过滤严重级（priority=4）且触发状态（value=1）的告警
            triggers = self.zapi.trigger.get(
                output=["description", "priority", "lastchange"],
                filter={"value": 1, "priority": 4},  # 只保留严重级告警
                sortfield="lastchange",
                sortorder="DESC",
                limit=limit,
                selectHosts=["host"],
                selectItems=["name", "key_"]
            )
            
            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_name = trigger['hosts'][0]['host'] if trigger['hosts'] else "未知主机"
                    
                    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"获取告警失败: {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_desktop_notification(self, alert):
        """在桌面右下角显示通知"""
        try:
            notification.notify(
                title=f"⚠️ Zabbix告警 - {alert['priority']}",
                message=f"主机: {alert['host']}\n描述: {alert['description']}\n时间: {alert['time']}",
                timeout=15,  # 显示15秒
                app_name="Zabbix告警监控",
				app_icon="_internal/zabbix.ico"
            )
            print(f"已显示告警通知: {alert['description']}")
        except Exception as e:
            print(f"显示通知失败: {e}")
    def monitor_alerts_loop(self, check_interval=10):
        """告警监控循环（运行在独立线程）"""
        print("开始监控Zabbix告警...")
        print(f"检查间隔: {check_interval}秒")
        
        while self.is_running:
            try:
                # 检查Zabbix连接是否有效，无效则重连
                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_desktop_notification(alert)
                        time.sleep(2)  # 避免通知密集
                else:
                    print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - 未发现新告警")
                
                time.sleep(check_interval)
                
            except Exception as e:
                print(f"监控过程中出现错误: {e}")
                time.sleep(30)  # 错误后延迟重连

    def start_monitor(self, check_interval=10):
        """启动监控（独立线程运行，避免阻塞托盘）"""
        self.is_running = True
        self.monitor_thread = threading.Thread(
            target=self.monitor_alerts_loop,
            args=(check_interval,),
            daemon=True  # 线程随主线程退出
        )
        self.monitor_thread.start()

    def stop_monitor(self):
        """停止监控"""
        self.is_running = False
        if self.monitor_thread and self.monitor_thread.is_alive():
            self.monitor_thread.join(timeout=5)
        print("监控已停止")

    def setup_tray(self):
        """设置系统托盘菜单"""
        # 加载图标
        icon_path = "_internal/zabbix.ico"  # 可自行放置图标文件在脚本目录
        icon = Image.open(icon_path)

        # 托盘菜单（右键菜单）
        menu = pystray.Menu(
            pystray.MenuItem(
                "退出监控",
                self.exit_app,  # 点击退出触发的函数
                default=False
            )
        )

        # 创建托盘图标
        self.tray_icon = pystray.Icon(
            name="ZabbixAlertMonitor",
            icon=icon,
            title="Zabbix告警监控",  # 鼠标悬浮时显示的提示
            menu=menu
        )

    def exit_app(self, icon, item):
        """退出应用（停止监控+关闭托盘）"""
        print("正在退出应用...")
        self.stop_monitor()  # 停止监控线程
        icon.stop()          # 关闭托盘图标
        sys.exit(0)          # 退出程序

    def run(self, check_interval=10):
        """启动应用（连接Zabbix+启动监控+显示托盘）"""
        # 1. 连接Zabbix
        if not self.connect_zabbix():
            print("无法连接到Zabbix服务器，请检查配置")
            sys.exit(1)
        
        # 2. 启动监控线程
        self.start_monitor(check_interval)
        
        # 3. 设置并运行托盘（阻塞主线程，保持程序常驻）
        self.setup_tray()
        self.tray_icon.run()

def main():
    # Zabbix服务器配置
    ZABBIX_CONFIG = {
        'url': 'http://*.*.*.*/api_jsonrpc.php',
        'username': 'Admin', 
        'password': 'zabbix'
    }
    
    # 创建监控器实例
    monitor = ZabbixAlertMonitor(
        ZABBIX_CONFIG['url'],
        ZABBIX_CONFIG['username'],
        ZABBIX_CONFIG['password']
    )
    
    # 运行应用（含托盘和监控）
    monitor.run(check_interval=10)

if __name__ == "__main__":
    main()

pyinstaller打包：

pyinstaller -w -i zabbix.ico --add-data "zabbix.ico;." --hidden-import=plyer.platforms.win.notification zabbix.py