目录一、前提条件二、脚本逻辑三、部分代码四、实现效果五、获取完整代码一、前提条件1、搭建有zabbix_watch 项目。2、运维人员无钉钉、飞书、企业微信的后台管理权限无法通过企业的办公软件绑定zabbix_watch。3、搭建有本地用的Openclaw且创建有响应的QQBOT二、脚本逻辑1、通过网页获取到了zabbix_watch的数据传输连接以及token首先绑定对应的API。# Zabbix API ZABBIX_URL http://zabbix_watch服务器IP:端口/api/modules/host-details #(zabbix_watch发送监控数据的连接 ZABBIX_TOKEN 连接服务器的TOKEN通过网页可以获取2、绑定对应的openclaw可通过咨询龙虾获取对应的API。# OpenClaw Gateway改成 Windows 主机内网 IPcmd 里 ipconfig 查看 GATEWAY_HOST 127.0.0.1 GATEWAY_PORT 18789 GATEWAY_TOKEN Openclaw的连接TOKEN3、设置QQ私聊的会话key# QQ 私聊会话 key QQ_SESSION_KEY QQ私聊会话的key4、监控条件设置# 监控设置 INTERVAL_SECONDS 60 # 轮询间隔秒 DISK_THRESHOLD 98.0 # 告警阈值百分比 # 弹窗设置 POPUP_EVERY 60 # 每 N 次监测弹窗一次60 次 x 60秒 ≈ 1小时 POPUP_TIMEOUT 60 # 弹窗自动关闭秒数0 不自动关闭需手动关 # 报告日志文件每次监测的完整状态都追加写入这里 REPORT_FILE os.path.join(os.path.dirname(os.path.abspath(__file__)), monitor_report.txt) REPORT_MAX_SIZE 5 * 1024 * 1024 # 超过 5MB 自动轮转为 monitor_report.old.txt5、设置后台静默运行# 系统日志只写文件控制台完全静默 LOG_FILE os.path.join(os.path.dirname(os.path.abspath(__file__)), monitor.log) logging.basicConfig( levellogging.INFO, format[%(asctime)s] %(levelname)s %(message)s, datefmt%Y-%m-%d %H:%M:%S, handlers[logging.FileHandler(LOG_FILE, encodingutf-8)] ) log logging.getLogger(zabbix_monitor)6、检测相关数据。三、部分代码def send_qq_alert(message): 通过 OpenClaw Gateway 发送 QQ 消息 url fhttp://{GATEWAY_HOST}:{GATEWAY_PORT}/tools/invoke headers { Authorization: fBearer {GATEWAY_TOKEN}, Content-Type: application/json } payload { tool: sessions_send, args: {sessionKey: QQ_SESSION_KEY, message: message} } try: resp requests.post(url, headersheaders, jsonpayload, timeout10) if resp.status_code 200: log.info(✅ QQ 消息已发送: %s, message.replace(\n, | )) with _stats_lock: _stats[alerts_sent] 1 _stats[last_alert_time] now() return True log.warning(QQ 发送失败 HTTP %s: %s, resp.status_code, resp.text[:200]) return False except Exception as e: log.warning(QQ 发送异常: %s, e) return Falsedef main(): log.info( * 52) log.info( Zabbix 磁盘监控已启动后台静默模式) log.info(Zabbix : %s, ZABBIX_URL) log.info(Gateway: http://%s:%s, GATEWAY_HOST, GATEWAY_PORT) log.info(间隔 %ss | 阈值 %s%% | 每 %s 次弹窗, INTERVAL_SECONDS, DISK_THRESHOLD, POPUP_EVERY) log.info(报告日志: %s, REPORT_FILE) log.info( * 52) cycle 0 token_expired_notified False while True: cycle 1 with _stats_lock: _stats[cycles] cycle hosts, token_expired fetch_hosts() # Token 过期发 QQ 通知后退出 if token_expired: log.error(Token 已过期发送通知后退出) write_report_failure(ZABBIX_TOKEN 已过期) if not token_expired_notified: send_qq_alert(⚠️ Zabbix 磁盘监控\nZABBIX_TOKEN 已过期请尽快更换) token_expired_notified True show_status_popup(build_status_summary(None, [])) log.info( 脚本已停止Token 过期) break # 数据获取失败失败也记录到报告 txt if hosts is None: log.warning(第 %s 次监测失败等待下次轮询, cycle) write_report_failure(Zabbix API 请求失败详见 monitor.log) time.sleep(INTERVAL_SECONDS) continue # 数据正常写入完整状态报告 token_expired_notified False with _stats_lock: _stats[last_report_time] now() alerts [] for h in hosts: visible h.get(visible_name, h.get(name, N/A)) name h.get(name, N/A) ip h.get(ip, N/A) disks parse_disks(h.get(disk, )) over [d for d in disks if d[usage] DISK_THRESHOLD] if over: alerts.append((visible, name, ip, over)) # 完整监控状态 → 写入 txt每次监测都记录 write_report(hosts, alerts) # 告警 → 发 QQ if alerts: log.warning(检测到 %s 台主机磁盘告警, len(alerts)) for visible, name, ip, over in alerts: disk_detail , .join(f{d[partition]}: {d[usage]}% for d in over) msg (f 磁盘告警\n服务器: {visible} ({name})\nIP: {ip}\n磁盘: {disk_detail}) send_qq_alert(msg) time.sleep(1) # 避免连续发送太快 # 每 N 次监测弹窗展示状态 if cycle % POPUP_EVERY 0: log.info(第 %s 次监测完成弹出状态窗口, cycle) show_status_popup(build_status_summary(hosts, alerts)) time.sleep(INTERVAL_SECONDS)四、实现效果五、获取完整代码点赞即可获取。
)
