基于python环境yum install python3pip3 install requests告警配置文件在cd /usr/lib/zabbix/alertscripts/ cd /usr/local/share/zabbix/alertscripts[rootlocalhost alertscripts]# lsdingding.py#!/usr/bin/env python3 # -*- coding: utf-8 -*- import requests import json import sys import os import datetime webhook https://oapi.dingtalk.com/robot/send?access_token自己的token11ca4841903f166d24e853502730985133d267ae0f1f8ea6c95d545edef4d768 log_path /usr/lib/zabbix/alertscripts/log/dingding.log if len(sys.argv) ! 4: print(参数错误python3 dingding.py 手机号 标题 内容) sys.exit(1) user str(sys.argv[1]).strip() subject sys.argv[2] text sys.argv[3] data { msgtype: text, text: { content: f{subject}{text} }, at: { atMobiles: [], isAtAll: False } } headers {Content-Type: application/json} try: x requests.post(urlwebhook, datajson.dumps(data), headersheaders, timeout10) res x.json() print(钉钉接口返回, res) except Exception as e: res {errcode: -999, errmsg: f请求异常:{str(e)}} print(请求异常, e) log_dir os.path.dirname(log_path) if not os.path.exists(log_dir): os.makedirs(log_dir) with open(log_path, a, encodingutf-8) as f: f.write(\n -- * 30) now str(datetime.datetime.now()) if res[errcode] 0: f.write(f\n{now} 推送成功 内容:{subject}{text}\n)测试[rootlocalhost alertscripts]# ls dingding.py [rootlocalhost alertscripts]# python3 /usr/local/share/zabbix/alertscripts/dingding.py 11111111111 告警 来自Zabbix的测试消息 钉钉接口返回 {errcode: 0, errmsg: ok}配置完成钉钉加签设置#!/usr/bin/env python3 # -*- coding: utf-8 -*- import requests import json import sys import os import datetime import time import hmac import hashlib import base64 # 钉钉机器人配置 webhook https://oapi.dingtalk.com/robot/send?access_token自己的token11ca4841903f166d24e853502730985133d267ae0f1f8ea6c95d545edef4d768 secret 自己secSEC111c25df8a37e56253eded6c7f51ddc96e414acd27258ac16cc96192d1769d7c1 log_path /usr/local/share/zabbix/alertscripts/log # 参数校验 if len(sys.argv) ! 4: print(参数错误python3 dingding.py 手机号 标题 内容) sys.exit(1) user str(sys.argv[1]).strip() subject sys.argv[2] text sys.argv[3] # 消息内容如需关键词可拼接 content f告警 {subject}{text} data { msgtype: text, text: { content: f{subject}{text} }, at: { atMobiles: [user], isAtAll: False } } headers {Content-Type: application/json} # 加签核心逻辑 timestamp str(round(time.time() * 1000)) sign_str f{timestamp}\n{secret} hmac_obj hmac.new(secret.encode(utf-8), sign_str.encode(utf-8), digestmodhashlib.sha256) sign base64.b64encode(hmac_obj.digest()).decode(utf-8) # 拼接带签名、时间戳的完整请求地址 send_url f{webhook}timestamp{timestamp}sign{sign} try: x requests.post(urlsend_url, datajson.dumps(data), headersheaders, timeout10) res x.json() print(钉钉接口返回, res) except Exception as e: res {errcode: -999, errmsg: f请求异常:{str(e)}} print(请求异常, e) # 自动创建日志目录写入日志 log_dir os.path.dirname(log_path) if not os.path.exists(log_dir): os.makedirs(log_dir) with open(log_path, a, encodingutf-8) as f: f.write(\n -- * 30) now str(datetime.datetime.now()) if res[errcode] 0: f.write(f\n{now} 推送成功 接收人:{user} 内容:{subject}{text}\n) else: f.write(f\n{now} 推送失败 接收人:{user} 错误信息:{res} 内容:{subject}{text}\n)
