脚本通知说明
信息
脚本通知（Script Notification）是私有化部署环境下的一种自定义通知渠道，通过在部署节点上执行用户提供的 Shell / Python 脚本来发送通知，适用于平台原生渠道（邮件、钉钉、企业微信、Webhook 等）无法覆盖的场景，如对接内部 OA 系统、短信网关、自建 IM 等。
注意
脚本通知仅支持私有化部署环境，公有云 SaaS 版本不提供该能力。
使用前提
- 部署形态为私有化部署
- 已获得部署环境的服务器访问权限，可将脚本文件放置到指定目录
- 熟悉 Shell 或 Python，了解脚本的执行环境（解释器版本、依赖库等）需与平台运行环境兼容
快速上手
第一步：编写通知脚本
按照约定的入参格式编写 Shell 或 Python 脚本，脚本需要读取平台传入的告警上下文参数（如告警标题、等级、状态等），并在脚本内部实现具体的推送逻辑。
第二步：部署脚本文件
将编写好的脚本放置到部署环境约定的脚本目录下，并赋予可执行权限（chmod +x）。
第三步：配置通知渠道
在通知策略的通知渠道中选择脚本通知，指定脚本文件路径。
第四步：测试并保存
建议先在脚本所在服务器上手动执行一次，确认脚本能正常运行、返 回预期结果后再保存策略。
功能说明
脚本执行方式
| 配置项 | 说明 |
|---|---|
| 脚本类型 | 支持Shell 和 Python 两种脚本类型 |
| 脚本路径 | 部署环境服务器上的可执行文件绝对路径，平台仅负责调用，不负责脚本内容的安全性和正确性 脚本路径"/data/br/conf/smartalert_service/script/alert" |
| 传入参数 | 平台将告警上下文（如标题、等级、状态、触发时间等）以参数形式传递给脚本 |
注意
脚本由用户自行编写和维护，平台不对脚本内部逻辑的正确性、安全性负责。建议脚本内部妥善处理异常，避免因未捕获异常导致进程非正常退出。
脚本示例数据
脚本内的json_data示例数据
{
// 【通知正文】按告警规则 notify.template 渲染后的纯文本（模板变量已替换）
"notifyContent": "告警等级：致命\n告警 ID：260609150027760001\n告警时间：2026-06-09 15:00:00\n告警状态：告警中\n服务：one 'framework' jar-自定义2\n服务识别名称：one_framework.jar",
// 【告警唯一标识】对应 Alert.alertId
"alertId": "260609150027760001",
// 【告警名称】对应 Alert.alertName
"alertName": "测试脚本通知",
// 【告警来源】1=系统检测 2=Prometheus 3=Zabbix 4=阿里云 5=华为云 6=天翼云 7=腾讯云 8=第三方
"sourceType": 1,
// 【分组维度信息】评估 target + groupBy；实体维度会带 {key}_name
"groupInfo": {
"serviceDetectedName": "one_framework.jar",
"serviceId_name": "one 'framework' jar-自定义2",
"serviceId": "848903"
},
// 【规则 ID】
"ruleId": 70657,
// 【规则 内容 MD5】无关联时为空串
"ruleMd5": "",
// 【规则名称】
"ruleName": "测试脚本通知",
// 【规则类型】1=阈值 2=突变 3=AI 4=组合 5=日志 6=事件 7=AI自适应 8=AI预测
"ruleType": 1,
// 【规则条件配置】规则表 condition_config 解析结果
"conditionConfig": {
"method": 0,
"exprs": [{
"name": "a",
"catalogId": "302611,302622,302625",
"path": ["错误", "302625"],
"metricId": "one.service.service.error.errorCount",
"attrKey": null,
"agg": "sum",
"show": 1,
"filters": [],
"logic": "and",
"groups": [{
"type": "entity",
"key": "serviceId",
"modelKey": "service",
"dimensionKey": "service",
"dataType": "number"
}, {
"type": "normal",
"key": "serviceDetectedName",
"modelKey": "service",
"dimensionKey": "",
"dataType": "text"
}]
}],
"expr": "",
"formula": "",
"timeRange": "5m",
"timeRange2": "",
"funcType": "",
"trigger": {
"operator": "greater",
"levels": [
{ "level": 6, "value": "3" },
{ "level": 5, "value": "" },
{ "level": 4, "value": "" }
],
"normal": 3,
"duration": 1,
"level": ""
},
"nodataFlag": 0,
"nodataLevel": 0,
"delayFlag": 1,
"delay": "1m",
"timeShift": "",
"rules": "",
"ruleExpr": "",
"algorithmAdvanced": ""
},
// 【告警状态】0=已恢复 1=告警中 2=无数据 3=已关闭 4=已关闭(手动)
"status": 1,
// 【告警等级】2=提醒 3=一般 4=警告 5=严重 6=致命
"level": 6,
// 【时间戳】Unix 毫秒；未发生时 recoverTime/closeTime 常为 ""
"createTime": 1780988400000,
"startTime": 1780988400000,
"recoverTime": "",
"closeTime": "",
"updateTime": 1780988400000,
// 【租户与资源域】
"accountId": 20001207,
"envId": "default",
"resourceZoneId": 1,
// 【关联实体】dimensionKey → entityId 列表
"relatedEntity": {
"service": [848903]
},
// 【所属系统】Alert.systemId → entity.system；[{id, name}]
"systems": [
{ "id": 7410748171534319, "name": "333333333" },
{ "id": 6472793535397552, "name": "mftest" }
],
// 【所属应用】Alert.appId → entity.app；结构与 systems 相同
"apps": [
{ "id": 7410748171, "name": "订单中 心" },
{ "id": 6472793535, "name": "支付网关" }
],
// 【用户自定义 labels】
"labels": "",
// 【附加信息】告警创建时写入
"annotations": {
"metric": "错误次数",
"dimEntityMap": {
"service": { "serviceId": 848903 }
},
"alertObjects": [
"服务:one 'framework' jar-自定义2",
"服务识别名称:one_framework.jar"
],
"modelKey": "service"
},
// 【通知策略名称】发送时注入
"notifyStrategyName": "脚本通知",
// 【通知策略类型】0=重复 1=升级；首次告警常为 ""
"strategyType": "",
// 【升级类型】仅 strategyType=1：0=未关闭 1=未恢复
"upgradeType": "",
// 【通知模板原文】规则 notify 配置 JSON 字符串
"template": "{\"content\":\"\"}",
// 【通知模板 ID】对应 Alert.templateId；未配置时常为 ""
"templateId": "",
// 【实体维度映射】dimensionKey → groupKey → entityId
"dimEntityMap": {
"service": {
"serviceId": 848903
}
},
// 【指标展示名】
"metric": "错误次数",
// 【指标触发值】
"value": 14.0,
// 【实体属性快照】发送时查 Titan is_show=1 属性；modelKey → { attrKey: value }
"entityAttr": {
"service": {
"detectedName": "one_framework.jar",
"instanceId": 848903,
"lifeCycle": 1,
"customizedName": "one 'framework' jar-自定义2"
}
},
// 【实体标签快照】modelKey → ["tagKey:tagValue", ...]；空时序列化为 ""
"tagsAttr": {
"service": ["aaaaaaaaaa:aaa"]
},
// 【通知触发原因】1=告警 2=恢复 3=关闭 4=重复 5=升级
"triggerReason": 1,
// 【异常类型】1=可用性 2=无数据 3=错误 4=缓慢 5=吞吐 6=资源 7=信息 8=其他
"exceptionType": 3
}
shell脚本示例
#!/bin/bash
# 脚本严格校验 任一步骤执行错误 脚本直接停止 避免带着错误数据往下执行
set -euo pipefail
# 日志路径统一配置
LOG_DIR="/data/br/conf/smartalert_service/script"
mkdir -p "${LOG_DIR}"
RAW_PARAM_LOG="${LOG_DIR}/param_raw.log"
DECODED_JSON_LOG="${LOG_DIR}/param_decoded.log"
# zh_CN=中文，其他值=英文
LANG_CODE="zh_CN"
# 检查jq是否安装
check_jq() {
if ! command -v jq &> /dev/null; then
echo "jq未安装，开始自动安装..."
if [ -f /etc/redhat-release ]; then
yum install -y jq &> /dev/null
elif [ -f /etc/debian_version ]; then
apt update &> /dev/null && apt install -y jq &> /dev/null
else
echo "无法自动安装jq，请手动安装后重试！"
exit 255
fi
fi
}
# 根据 status 编码与中英文语言参数返回展示文案
translate_status() {
local status_code="$1"
local lang_code="$2"
case "${status_code}" in
0)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "已恢复"; else echo "Resolved"; fi
;;
1)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "告警中"; else echo "Alerting"; fi
;;
2)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "无数据"; else echo "No data"; fi
;;
3)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "已关闭"; else echo "Closed"; fi
;;
4)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "已关闭(手动)"; else echo "Closed (Manual)"; fi
;;
*)
echo "${status_code}"
;;
esac
}
# 根据 level 编码与中英文语言参数返回展示文案
translate_level() {
local level_code="$1"
local lang_code="$2"
case "${level_code}" in
2)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "提醒"; else echo "remindful"; fi
;;
3)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "一般"; else echo "general"; fi
;;
4)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "警告"; else echo "warning"; fi
;;
5)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "严重"; else echo "serious"; fi
;;
6)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "致命"; else echo "fatal"; fi
;;
*)
echo "${level_code}"
;;
esac
}
# 根据 exceptionType 编码与中英文语言参数返回展示文案
translate_exception_type() {
local type_code="$1"
local lang_code="$2"
case "${type_code}" in
1)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "可用性"; else echo "usability"; fi
;;
2)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "无数据"; else echo "nodata"; fi
;;
3)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "错误"; else echo "error"; fi
;;
4)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "缓慢"; else echo "slow"; fi
;;
5)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "吞吐"; else echo "throughput"; fi
;;
6)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "资源"; else echo "resource"; fi
;;
7)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "通知"; else echo "info"; fi
;;
8)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "其他"; else echo "other"; fi
;;
*)
echo "${type_code}"
;;
esac
}
# 根据 triggerReason 编码与中英文语言参数返回展示文案
translate_trigger_reason() {
local reason_code="$1"
local lang_code="$2"
case "${reason_code}" in
1)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "告警通知"; else echo "TRIGGERED"; fi
;;
2)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "恢复通知"; else echo "RECOVERED"; fi
;;
3)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "关闭通知"; else echo "CLOSED"; fi
;;
4)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "重复通知"; else echo "RE_TRIGGERED"; fi
;;
5)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "升级通知"; else echo "ESCALATED"; fi
;;
*)
echo "${reason_code}"
;;
esac
}
# 将毫秒时间戳格式化为 YYYY-MM-DD HH:MM:SS，已是该格式则原样返回
format_timestamp() {
local raw_value="$1"
if [[ -z "${raw_value}" || "${raw_value}" == "null" ]]; then
echo ""
return
fi
if [[ "${raw_value}" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2} ]]; then
echo "${raw_value}"
return
fi
if [[ "${raw_value}" =~ ^[0-9]+(\.[0-9]+)?$ ]]; then
local timestamp="${raw_value%%.*}"
if (( timestamp >= 1000000000000 )); then
timestamp=$((timestamp / 1000))
fi
date -d "@${timestamp}" "+%Y-%m-%d %H:%M:%S" 2>/dev/null || echo "${raw_value}"
return
fi
echo "${raw_value}"
}
# 追加告警消息行，空字段也展示标签
append_alert_line() {
local label="$1"
local value="$2"
if [[ "${value}" == "null" ]]; then
value=""
fi
FINAL_ALERT_MSG="${FINAL_ALERT_MSG}${label}: ${value}"$'\n'
}
# 从 conditionConfig.trigger.levels 中按告警 level 解析触发阈值
resolve_alert_threshold() {
local decoded_json="$1"
local raw_level="$2"
printf '%s' "${decoded_json}" | jq -r --arg level "${raw_level}" '
(.conditionConfig
| if type == "string" then (try fromjson catch null) else . end) as $conditionConfig
| if $conditionConfig == null
or ($conditionConfig.trigger // null) == null
or ($conditionConfig.trigger.levels // null) == null
then ""
else (
($level | if . == "" or . == "null" then null else tonumber end) as $levelNumber
| if $levelNumber == null then ""
else (
$conditionConfig.trigger.levels[]
| select(.level == $levelNumber)
| .value
| tostring
)
end
)
end
' | head -n 1
}
# 1. 前置检查
check_jq
# 2. 参数校验
json_data="$1"
echo "原始入参: ${json_data}" >> "${RAW_PARAM_LOG}"
if [[ -z "${json_data}" ]]; then
printf "%s" "参数不能为空"
exit 255
fi
# 3. Base64解码（处理可能的换行/空格）
json_data=$(printf "%s" "${json_data}" | base64 -d 2>/dev/null)
if [[ -z "${json_data}" ]]; then
printf "%s" "Base64解码失败或解码后为空"
exit 255
fi
echo "解码后JSON: ${json_data}" >> "${DECODED_JSON_LOG}"
# 4. 提取原始 status / level / exceptionType / triggerReason 编码
RAW_STATUS=$(printf '%s' "${json_data}" | jq -r '.status // ""')
RAW_LEVEL=$(printf '%s' "${json_data}" | jq -r '.level // ""')
RAW_EXCEPTION_TYPE=$(printf '%s' "${json_data}" | jq -r '.exceptionType // ""')
RAW_TRIGGER_REASON=$(printf '%s' "${json_data}" | jq -r '.triggerReason // ""')
# 5. 按语言参数翻译 status / level / exceptionType / triggerReason
ALERT_STATUS=$(translate_status "${RAW_STATUS}" "${LANG_CODE}")
ALERT_LEVEL=$(translate_level "${RAW_LEVEL}" "${LANG_CODE}")
ALERT_EXCEPTION_TYPE=$(translate_exception_type "${RAW_EXCEPTION_TYPE}" "${LANG_CODE}")
ALERT_TRIGGER_REASON=$(translate_trigger_reason "${RAW_TRIGGER_REASON}" "${LANG_CODE}")
# 6. 从 conditionConfig.trigger.levels 解析当前告警触发阈值
ALERT_THRESHOLD=$(resolve_alert_threshold "${json_data}" "${RAW_LEVEL}")
# 7. 提取其他字段（缺失时为 ""）
ALERT_ENV=$(printf '%s' "${json_data}" | jq -r '.envId // ""')
ALERT_NAME=$(printf '%s' "${json_data}" | jq -r '.alertName // ""')
ALERT_DURATION=$(printf '%s' "${json_data}" | jq -r '.alertDuration // ""')
RAW_START_TIME=$(printf '%s' "${json_data}" | jq -r '.startTime // ""')
RAW_UPDATE_TIME=$(printf '%s' "${json_data}" | jq -r '.updateTime // ""')
ALERT_START_TIME=$(format_timestamp "${RAW_START_TIME}")
ALERT_UPDATE_TIME=$(format_timestamp "${RAW_UPDATE_TIME}")
ALERT_ID=$(printf '%s' "${json_data}" | jq -r '.alertId // ""')
ALERT_METRIC=$(printf '%s' "${json_data}" | jq -r '.metric // ""')
HOST_IPV4_ADDRESS=$(printf '%s' "${json_data}" | jq -r '.groupInfo.ipv4Address // ""')
HOST_NAME=$(printf '%s' "${json_data}" | jq -r '.groupInfo.hostId_name // ""')
# 打印所有变量值到日志（不论是否有值）
{
echo "========== 解码后JSON变量值 =========="
echo "语言参数: ${LANG_CODE}"
echo "原始状态编码: ${RAW_STATUS}"
echo "原始等级编码: ${RAW_LEVEL}"
echo "原始异常类型编码: ${RAW_EXCEPTION_TYPE}"
echo "原始触发原因编码: ${RAW_TRIGGER_REASON}"
echo "告警状态: ${ALERT_STATUS}"
echo "异常类型: ${ALERT_EXCEPTION_TYPE}"
echo "触发原因: ${ALERT_TRIGGER_REASON}"
echo "环境名称: ${ALERT_ENV}"
echo "告警名称: ${ALERT_NAME}"
echo "主机IPV4地址: ${HOST_IPV4_ADDRESS}"
echo "主机名称: ${HOST_NAME}"
echo "告警持续时长: ${ALERT_DURATION}"
echo "原始发生时间: ${RAW_START_TIME}"
echo "原始更新时间: ${RAW_UPDATE_TIME}"
echo "发生时间: ${ALERT_START_TIME}"
echo "更新时间: ${ALERT_UPDATE_TIME}"
echo "告警等级: ${ALERT_LEVEL}"
echo "触发阈值: ${ALERT_THRESHOLD}"
echo "告警ID: ${ALERT_ID}"
echo "指标: ${ALERT_METRIC}"
} >> "${LOG_DIR}/json_values.log"
# ===================== 【设置一个完整消息变量，供后面使用】 =====================
FINAL_ALERT_MSG="【告警通知】"$'\n'
append_alert_line "告警状态" "${ALERT_STATUS}"
append_alert_line "异常类型" "${ALERT_EXCEPTION_TYPE}"
append_alert_line "触发原因" "${ALERT_TRIGGER_REASON}"
append_alert_line "环境名称" "${ALERT_ENV}"
append_alert_line "告警名称" "${ALERT_NAME}"
append_alert_line "主机IPV4地址" "${HOST_IPV4_ADDRESS}"
append_alert_line "主机名称" "${HOST_NAME}"
append_alert_line "告警持续时长" "${ALERT_DURATION}"
append_alert_line "发生时间" "${ALERT_START_TIME}"
append_alert_line "更新时间" "${ALERT_UPDATE_TIME}"
append_alert_line "告警等级" "${ALERT_LEVEL}"
append_alert_line "触发阈值" "${ALERT_THRESHOLD}"
append_alert_line "告警ID" "${ALERT_ID}"
append_alert_line "指标" "${ALERT_METRIC}"
FINAL_ALERT_MSG="${FINAL_ALERT_MSG%"${FINAL_ALERT_MSG##*[![:space:]]}"}"
# 9. 构建告警内容函数
get_content(){
echo "${FINAL_ALERT_MSG}"
}
# 10. 发送通知
send_notification() {
printf "%s\n" "start send notification"
# 使用 jq 构建 JSON，避免特殊字符导致 curl 请求体异常
curl_payload=$(jq -n \
--arg message "${FINAL_ALERT_MSG}" \
--arg status "${ALERT_STATUS:-}" \
--arg level "${ALERT_LEVEL:-}" \
--arg threshold "${ALERT_THRESHOLD:-}" \
--arg exceptionType "${ALERT_EXCEPTION_TYPE:-}" \
--arg triggerReason "${ALERT_TRIGGER_REASON:-}" \
--arg env "${ALERT_ENV:-}" \
--arg alertName "${ALERT_NAME:-}" \
--arg hostIpv4 "${HOST_IPV4_ADDRESS:-}" \
--arg hostName "${HOST_NAME:-}" \
--arg duration "${ALERT_DURATION:-}" \
--arg startTime "${ALERT_START_TIME:-}" \
--arg updateTime "${ALERT_UPDATE_TIME:-}" \
--arg alertId "${ALERT_ID:-}" \
--arg metric "${ALERT_METRIC:-}" \
--arg lang "${LANG_CODE:-}" \
'{
message: $message,
status: $status,
level: $level,
threshold: $threshold,
exceptionType: $exceptionType,
triggerReason: $triggerReason,
env: $env,
alertName: $alertName,
hostIpv4: $hostIpv4,
hostName: $hostName,
duration: $duration,
startTime: $startTime,
updateTime: $updateTime,
alertId: $alertId,
metric: $metric,
lang: $lang
}')
curl_result=$(curl -s -X POST \
-H "Content-Type: application/json" \
-H "Accept: */*" \
-H "Connection: keep-alive" \
-d "${curl_payload}" \
"https://xxx/test")
if [[ $? -eq 0 ]]; then
printf "%s\n" "notification sent successfully"
else
printf "%s\n" "failed to send notification"
exit 1
fi
printf "%s\n" "end send notification"
}
# 执行核心逻辑
send_notification
get_content
exit 0
python脚本示例
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
告警通知脚本（Python 原生实现，无第三方依赖）。
仅支持 Python 3.5+；若使用 Python 2 执行，将记录错误日志后直接退出。
"""
import base64
import binascii
import datetime
import json
import os
import re
import sys
import urllib.parse
# 日志路径（版本检查前定义，供 Python 2 退出时写日志）
LOG_DIR = "/data/br/conf/smartalert_service/script"
SCRIPT_ERROR_LOG = os.path.join(LOG_DIR, "script_error.log")
if sys.version_info[0] < 3:
_err_msg = "本脚本仅支持 Python 3，当前为 Python 2，请使用 python3 执行"
try:
import codecs
import datetime
if not os.path.isdir(LOG_DIR):
os.makedirs(LOG_DIR)
_ts = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
with codecs.open(SCRIPT_ERROR_LOG, "a", encoding="utf-8") as _f:
_f.write("[{0}] {1}\n".format(_ts, _err_msg))
except Exception:
pass
sys.stderr.write(_err_msg + "\n")
sys.exit(255)
import urllib.error
import urllib.request
RAW_PARAM_LOG = os.path.join(LOG_DIR, "param_raw.log")
DECODED_JSON_LOG = os.path.join(LOG_DIR, "param_decoded.log")
JSON_VALUES_LOG = os.path.join(LOG_DIR, "json_values.log")
# 通知接口地址
#NOTIFY_URL = "https://xxx/test"
NOTIFY_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
# 语言参数（脚本内写死）：zh_CN=中文，其他值=英文
LANG_CODE = "zh_CN"
STATUS_TEXT = {
"zh_CN": {
"0": "已恢复",
"1": "告警中",
"2": "无数据",
"3": "已关闭",
"4": "已关闭(手动)",
},
"en": {
"0": "Resolved",
"1": "Alerting",
"2": "No data",
"3": "Closed",
"4": "Closed (Manual)",
},
}
LEVEL_TEXT = {
"zh_CN": {
"2": "提醒",
"3": "一般",
"4": "警告",
"5": "严重",
"6": "致命",
},
"en": {
"2": "remindful",
"3": "general",
"4": "warning",
"5": "serious",
"6": "fatal",
},
}
EXCEPTION_TYPE_TEXT = {
"zh_CN": {
"1": "可用性",
"2": "无数据",
"3": "错误",
"4": "缓慢",
"5": "吞吐",
"6": "资源",
"7": "通知",
"8": "其他",
},
"en": {
"1": "usability",
"2": "nodata",
"3": "error",
"4": "slow",
"5": "throughput",
"6": "resource",
"7": "info",
"8": "other",
},
}
TRIGGER_REASON_TEXT = {
"zh_CN": {
"1": "告警通知",
"2": "恢复通知",
"3": "关闭通知",
"4": "重复通知",
"5": "升级通知",
},
"en": {
"1": "TRIGGERED",
"2": "RECOVERED",
"3": "CLOSED",
"4": "RE_TRIGGERED",
"5": "ESCALATED",
},
}
def append_log(log_file, message):
"""追加一行日志。"""
if not os.path.isdir(LOG_DIR):
os.makedirs(LOG_DIR)
with open(log_file, "a", encoding="utf-8") as f:
f.write(message + "\n")
def append_log_block(log_file, lines):
"""追加多行日志。"""
if not os.path.isdir(LOG_DIR):
os.makedirs(LOG_DIR)
with open(log_file, "a", encoding="utf-8") as f:
f.write("\n".join(lines) + "\n")
def decode_base64_param(raw_param):
"""Base64 解码入参。"""
cleaned = raw_param.strip()
if not cleaned:
return ""
try:
decoded_bytes = base64.b64decode(cleaned)
except (ValueError, binascii.Error):
return ""
return decoded_bytes.decode("utf-8")
def normalize_json_text(json_text):
"""修复脚本通知 JSON 中常见的非法格式。"""
text = json_text
if "%" in text and (text.startswith("%7B") or "%22" in text):
text = urllib.parse.unquote(text)
text = re.sub(
r'("catalogId"\s*:\s*")((?:[^"\\]|\\.)*)(")',
lambda match: match.group(1) + re.sub(r"[\r\n\t ]+", "", match.group(2)) + match.group(3),
text,
flags=re.DOTALL,
)
return text
def get_json_field(data, *keys):
"""从 JSON 中提取字段，缺失或 null 时返回空字符串。"""
current = data
for key in keys:
if not isinstance(current, dict):
return ""
current = current.get(key)
if current is None:
return ""
return str(current)
def format_timestamp(raw_value):
"""将毫秒时间戳格式化为 YYYY-MM-DD HH:MM:SS，已是该格式则原样返回。"""
if raw_value in ("", "null", None):
return ""
text = str(raw_value).strip()
if not text:
return ""
if re.match(r"^\d{4}-\d{2}-\d{2}", text):
return text
if not re.match(r"^\d+(\.\d+)?$", text):
return text
try:
timestamp = int(float(text))
except (TypeError, ValueError):
return text
if timestamp >= 1000000000000:
timestamp = timestamp // 1000
try:
return datetime.datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S")
except (OverflowError, OSError, ValueError):
return text
def translate_status(status_code, lang_code):
"""根据 status 编码与中英文语言参数返回展示文案。"""
status_key = str(status_code)
lang_key = "zh_CN" if lang_code == "zh_CN" else "en"
return STATUS_TEXT.get(lang_key, {}).get(status_key, status_key)
def translate_level(level_code, lang_code):
"""根据 level 编码与中英文语言参数返回展示文案。"""
level_key = str(level_code)
lang_key = "zh_CN" if lang_code == "zh_CN" else "en"
return LEVEL_TEXT.get(lang_key, {}).get(level_key, level_key)
def translate_exception_type(exception_type_code, lang_code):
"""根据 exceptionType 编码与中英文语言参数返回展示文案。"""
type_key = str(exception_type_code)
lang_key = "zh_CN" if lang_code == "zh_CN" else "en"
return EXCEPTION_TYPE_TEXT.get(lang_key, {}).get(type_key, type_key)
def translate_trigger_reason(trigger_reason_code, lang_code):
"""根据 triggerReason 编码与中英文语言参数返回展示文案。"""
reason_key = str(trigger_reason_code)
lang_key = "zh_CN" if lang_code == "zh_CN" else "en"
return TRIGGER_REASON_TEXT.get(lang_key, {}).get(reason_key, reason_key)
def resolve_alert_threshold(json_data, raw_level):
"""从 conditionConfig.trigger.levels 中按告警 level 解析触发阈值。"""
condition_config = json_data.get("conditionConfig")
if isinstance(condition_config, str):
try:
condition_config = json.loads(condition_config)
except ValueError:
return ""
if not isinstance(condition_config, dict):
return ""
trigger = condition_config.get("trigger")
if not isinstance(trigger, dict):
return ""
levels = trigger.get("levels")
if not isinstance(levels, list):
return ""
if raw_level in ("", "null", None):
return ""
try:
level_number = int(raw_level)
except (TypeError, ValueError):
return ""
for level_item in levels:
if not isinstance(level_item, dict):
continue
if level_item.get("level") == level_number:
value = level_item.get("value")
return "" if value is None else str(value)
return ""
def build_alert_message(field_items):
"""构建完整告警消息文本，空字段也展示标签。"""
lines = ["【告警通知】"]
for label, value in field_items:
if value in ("null", None):
value = ""
lines.append("{0}: {1}".format(label, value))
return "\n".join(lines)
def get_content(final_alert_msg):
"""返回告警内容（供外部调用）。"""
return final_alert_msg
def send_notification(content_text):
"""发送 HTTP 通知。"""
print("start send notification")
wechat_body = {
"msgtype": "markdown",
"markdown": {
"content": content_text[:4096],
},
}
request_body = json.dumps(wechat_body, ensure_ascii=False).encode("utf-8")
request = urllib.request.Request(
NOTIFY_URL,
data=request_body,
headers={
"Content-Type": "application/json",
"Accept": "*/*",
"Connection": "keep-alive",
},
)
try:
response = urllib.request.urlopen(request, timeout=30)
try:
response.read()
finally:
response.close()
print("notification sent successfully")
except urllib.error.URLError as exc:
print("failed to send notification")
sys.stderr.write(str(exc) + "\n")
sys.exit(1)
print("end send notification")
def main():
if len(sys.argv) < 2:
print("参数不能为空", end="")
return 255
raw_param = sys.argv[1]
append_log(RAW_PARAM_LOG, "原始入参: {0}".format(raw_param))
if not raw_param:
print("参数不能为空", end="")
return 255
json_text = decode_base64_param(raw_param)
if not json_text:
print("Base64解码失败或解码后为空", end="")
return 255
append_log(DECODED_JSON_LOG, "解码后JSON: {0}".format(json_text))
json_text = normalize_json_text(json_text)
try:
json_data = json.loads(json_text)
except ValueError:
print("JSON解析失败", end="")
return 255
if not isinstance(json_data, dict):
print("JSON根节点必须是对象", end="")
return 255
raw_status = get_json_field(json_data, "status")
raw_level = get_json_field(json_data, "level")
raw_exception_type = get_json_field(json_data, "exceptionType")
raw_trigger_reason = get_json_field(json_data, "triggerReason")
alert_status = translate_status(raw_status, LANG_CODE)
alert_level = translate_level(raw_level, LANG_CODE)
alert_threshold = resolve_alert_threshold(json_data, raw_level)
alert_exception_type = translate_exception_type(raw_exception_type, LANG_CODE)
alert_trigger_reason = translate_trigger_reason(raw_trigger_reason, LANG_CODE)
alert_env = get_json_field(json_data, "envId")
alert_name = get_json_field(json_data, "alertName")
alert_duration = get_json_field(json_data, "alertDuration")
raw_start_time = get_json_field(json_data, "startTime")
raw_update_time = get_json_field(json_data, "updateTime")
alert_start_time = format_timestamp(raw_start_time)
alert_update_time = format_timestamp(raw_update_time)
alert_id = get_json_field(json_data, "alertId")
alert_metric = get_json_field(json_data, "metric")
host_ipv4_address = get_json_field(json_data, "groupInfo", "ipv4Address")
host_name = get_json_field(json_data, "groupInfo", "hostId_name")
append_log_block(
JSON_VALUES_LOG,
[
"========== 解码后JSON变量值 ==========",
"语言参数: {0}".format(LANG_CODE),
"原始状态编码: {0}".format(raw_status),
"原始等级编码: {0}".format(raw_level),
"原始异常类型编码: {0}".format(raw_exception_type),
"原始触发原因编码: {0}".format(raw_trigger_reason),
"告警状态: {0}".format(alert_status),
"异常类型: {0}".format(alert_exception_type),
"触发原因: {0}".format(alert_trigger_reason),
"环境名称: {0}".format(alert_env),
"告警名称: {0}".format(alert_name),
"主机IPV4地址: {0}".format(host_ipv4_address),
"主机名称: {0}".format(host_name),
"告警持续时长: {0}".format(alert_duration),
"原始发生时间: {0}".format(raw_start_time),
"原始更新时间: {0}".format(raw_update_time),
"发生时间: {0}".format(alert_start_time),
"更新时间: {0}".format(alert_update_time),
"告警等级: {0}".format(alert_level),
"触发阈值: {0}".format(alert_threshold),
"告警ID: {0}".format(alert_id),
"指标: {0}".format(alert_metric),
],
)
message_field_items = (
("告警状态", alert_status),
("异常类型", alert_exception_type),
("触发原因", alert_trigger_reason),
("环境名称", alert_env),
("告警名称", alert_name),
("主机IPV4地址", host_ipv4_address),
("主机名称", host_name),
("告警持续时长", alert_duration),
("发生时间", alert_start_time),
("更新时间", alert_update_time),
("告警等级", alert_level),
("触发阈值", alert_threshold),
("告警ID", alert_id),
("指标", alert_metric),
)
final_alert_msg = build_alert_message(message_field_items)
curl_payload = {
"message": final_alert_msg,
"status": alert_status or "",
"level": alert_level or "",
"threshold": alert_threshold or "",
"exceptionType": alert_exception_type or "",
"triggerReason": alert_trigger_reason or "",
"env": alert_env or "",
"alertName": alert_name or "",
"hostIpv4": host_ipv4_address or "",
"hostName": host_name or "",
"duration": alert_duration or "",
"startTime": alert_start_time or "",
"updateTime": alert_update_time or "",
"alertId": alert_id or "",
"metric": alert_metric or "",
"lang": LANG_CODE or "",
}
#send_notification(curl_payload)
send_notification(final_alert_msg)
get_content(final_alert_msg)
return 0
if __name__ == "__main__":
sys.exit(main())
常见场景
场景：对接内部自建 IM 系统 编写 Python 脚本，接收平台传入的告警参数后调用企业内部自建 IM 的 API 完成消息推送，弥补平台未内置该 IM 类型的空缺。
场景：触发内部工单系统 编写 Shell 脚本，在收到严重及以上级别告警时自动调用内部工单系统接口创建工单，实现告警到工单的自动化流转。
注意事项
危险
脚本以部署环境所在服务器的权限执行，请勿在脚本中引入未经审计的第三方代码或执行高危操作，避免引入安全风险。
信息
如后续升级或迁移部署环境，请确认脚本文件及其依赖（解释器版本、第三方库等）已同步迁移，否则脚本通知将无法正常执行。