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
| import json import requests import sys import logging import ast
API_BASE_URL = "https://api.dingtalk.com/v1.0" APP_KEY = "*************" APP_SECRET = "************************" ROBOT_CODE = "********************"
def get_access_token(): url = f"{API_BASE_URL}/oauth2/accessToken" headers = {"Content-Type": "application/json"} data = {"appKey": APP_KEY, "appSecret": APP_SECRET}
try: response = requests.post(url, json=data, headers=headers) response.raise_for_status() json_response = response.json() return json_response.get("accessToken") except requests.exceptions.RequestException as e: logging.error(f"Error getting access token: {e}") return None
def send_to_message(access_token, user_ids, msg_param_data): url = f"{API_BASE_URL}/robot/oToMessages/batchSend" headers = {"Content-Type": "application/json", "x-acs-dingtalk-access-token": access_token} data = {"robotCode": ROBOT_CODE, "userIds": user_ids, "msgKey": "officialMarkdownMsg", "msgParam": msg_param_data}
try: response = requests.post(url, json=data, headers=headers) response.raise_for_status() print(response.text) except requests.exceptions.RequestException as e: logging.error(f"Error sending message: {e}")
def get_user_id(access_token, phone): url = f"{API_BASE_URL}/topapi/v2/user/getbymobile?access_token={access_token}" headers = {"Content-Type": "application/json"} data = {"mobile": phone}
try: response = requests.post(url, json=data, headers=headers) response.raise_for_status() json_response = response.json() return json_response.get("result", {}).get("userid") except requests.exceptions.RequestException as e: logging.error(f"Error getting user ID: {e}") return None
def get_trigger_color(severity): color_dict = { "Not classified": "#808080", "Information": "#008000", "Warning": "#FFFF00", "Average": "#FFA500", "High": "#FF0000", "Disaster": "#8B0000", "default_color": "#808080" } return color_dict.get(severity, color_dict["default_color"])
def main(): access_token = get_access_token()
if not access_token: logging.error("Failed to obtain access token. Exiting.") sys.exit(1)
phone_id = sys.argv[1]
user_id = get_user_id(access_token, phone_id)
if not user_id: logging.error("Failed to obtain user ID. Exiting.") sys.exit(1)
user_ids = [user_id]
subject = sys.argv[2] message = sys.argv[3]
message_data = json.loads(message)
message_dictionary = ast.literal_eval(message) HOSTNAME = ': '.join(('> 告警主机',message_dictionary["HOSTNAME"])) HOSTIP = ': '.join(('告警地址',message_dictionary["HOSTIP"])) ITEMNAME = ': '.join(('监控项目',message_dictionary["ITEMNAME"])) ITEMLASTVALUE = ': '.join(('监控取值',message_dictionary["ITEMLASTVALUE"])) TRIGGERSEVERITY = ': '.join(('告警等级',message_dictionary["TRIGGERSEVERITY"])) TRIGGERSTATUS = ': '.join(('当前状态',message_dictionary["TRIGGERSTATUS"])) TRIGGERNAME = ': '.join(('告警信息',message_dictionary["TRIGGERNAME"])) EVENTTIME = ': '.join(('告警时间',message_dictionary["EVENTTIME"])) EVENTAGE = ': '.join(('持续时间',message_dictionary["EVENTAGE"])) EVENTID = ': '.join(('事件ID',message_dictionary["EVENTID"])) sendtext = '\n\n >'.join((HOSTNAME,HOSTIP,ITEMNAME,ITEMLASTVALUE,TRIGGERSEVERITY,TRIGGERSTATUS,TRIGGERNAME,EVENTTIME,EVENTAGE,EVENTID))
trigger_severity = message_data.get('TRIGGERSEVERITY', None) rgb_code = get_trigger_color(trigger_severity)
msg_data = { "title": "zabbix告警消息", "text": ''.join(('<font color=' + rgb_code + ' ' + 'face="黑体">' + subject + '</font>' + '\n\n' + sendtext + '\n\n' '#### [点击转到ZABBIX](https://zabbix.wenet.com.cn/)' )) }
msg_data_string = json.dumps(msg_data)
send_to_message(access_token, user_ids, msg_data_string)
logging.info(f"user_id: {user_id}") logging.info(f"subject: {subject}") logging.info(f"message: {message}")
if __name__ == "__main__": main()
|