zabbix本地监控程序

此脚本常驻系统右下角，当有告警会弹出通知，点击图标会打开监控网页。

#!/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)

pyinstaller打包：

pyinstaller -w -i zabbix.ico --add-data "zabbix.ico;." zabbix.py