Script Notification
info
Script Notification is a custom notification channel available in private (on-premises) deployments. It sends notifications by executing a user-provided Shell or Python script on the deployment node, suitable for scenarios that the platform's built-in channels (Email, DingTalk, WeCom, Webhook, etc.) cannot cover — such as integrating with an internal OA system, an SMS gateway, or a self-built IM system.
warning
Script Notification is only supported in private deployments; it is not available in the public cloud SaaS edition.
Prerequisites
- The deployment is a private (on-premises) deployment
- You have server access to the deployment environment and can place script files in the designated directory
- You are familiar with Shell or Python, and understand that the script's execution environment (interpreter version, dependencies, etc.) must be compatible with the platform's runtime environment
Quick Start
Step 1: Write the Notification Script
Write a Shell or Python script following the agreed-upon parameter format. The script needs to read the alert context parameters passed in by the platform (such as alert title, level, status, etc.) and implement the actual delivery logic internally.
Step 2: Deploy the Script File
Place the finished script in the script directory agreed upon by the deployment environment, and grant it execute permission (chmod +x).
Step 3: Configure the Notification Channel
In the notification channel of a Notification Strategy, select Script Notification and specify the script file path.
Step 4: Test and Save
It is recommended to manually run the script once on the server where it resides to confirm it runs correctly and returns the expected result before saving the strategy.
Feature Description
Script Execution
| Configuration Item | Description |
|---|---|
| Script Type | Supports two script types:Shell and Python |
| Script Path | The absolute path of the executable file on the deployment environment's server. The platform is only responsible for invoking it, not for the security or correctness of the script's content Script Path"/data/br/conf/smartalert_service/script/alert"” |
| Input Parameters | The platform passes the alert context (such as title, level, status, trigger time, etc.) to the script as a parameter |
warning
The script is written and maintained by the user; the platform is not responsible for the correctness or security of the script's internal logic. It is recommended that the script handle exceptions properly internally, to avoid abnormal process exits caused by uncaught exceptions.
Example Script Data
Example json_data Passed to the Script
{
// [Notification body] Plain text rendered from the alert rule's notify.template (template variables already substituted)
"notifyContent": "Alert Level: Critical\nAlert ID: 260609150027760001\nAlert Time: 2026-06-09 15:00:00\nAlert Status: Alerting\nService: one 'framework' jar-Custom2\nService Identified Name: one_framework.jar",
// [Unique alert identifier] Corresponds to Alert.alertId
"alertId": "260609150027760001",
// [Alert name] Corresponds to Alert.alertName
"alertName": "Test Script Notification",
// [Alert source] 1=System Detection 2=Prometheus 3=Zabbix 4=Alibaba Cloud 5=Huawei Cloud 6=CTYun 7=Tencent Cloud 8=Third-party
"sourceType": 1,
// [Grouping dimension info] Evaluated from target + groupBy; entity dimensions carry a {key}_name field
"groupInfo": {
"serviceDetectedName": "one_framework.jar",
"serviceId_name": "one 'framework' jar-Custom2",
"serviceId": "848903"
},
// [Rule ID]
"ruleId": 70657,
// [Rule content MD5] Empty string when there is no association
"ruleMd5": "",
// [Rule name]
"ruleName": "Test Script Notification",
// [Rule type] 1=Threshold 2=Change 3=AI 4=Composite 5=Log 6=Event 7=AI Adaptive 8=AI Forecast
"ruleType": 1,
// [Rule condition configuration] Parsed result of the rule table's condition_config
"conditionConfig": {
"method": 0,
"exprs": [{
"name": "a",
"catalogId": "302611,302622,302625",
"path": ["Error", "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": ""
},
// [Alert status] 0=Resolved 1=Alerting 2=No Data 3=Closed 4=Closed (Manual)
"status": 1,
// [Alert level] 2=Info 3=General 4=Warning 5=Error 6=Critical
"level": 6,
// [Timestamps] Unix milliseconds; recoverTime/closeTime are usually "" when not yet applicable
"createTime": 1780988400000,
"startTime": 1780988400000,
"recoverTime": "",
"closeTime": "",
"updateTime": 1780988400000,
// [Tenant and resource domain]
"accountId": 20001207,
"envId": "default",
"resourceZoneId": 1,
// [Related entities] dimensionKey → list of entityId
"relatedEntity": {
"service": [848903]
},
// [Owning system] Alert.systemId → entity.system; [{id, name}]
"systems": [
{ "id": 7410748171534319, "name": "333333333" },
{ "id": 6472793535397552, "name": "mftest" }
],
// [Owning app] Alert.appId → entity.app; same structure as systems
"apps": [
{ "id": 7410748171, "name": "Order Center" },
{ "id": 6472793535, "name": "Payment Gateway" }
],
// [User-defined labels]
"labels": "",
// [Additional info] Written when the alert is created
"annotations": {
"metric": "Error Count",
"dimEntityMap": {
"service": { "serviceId": 848903 }
},
"alertObjects": [
"Service:one 'framework' jar-Custom2",
"Service Identified Name:one_framework.jar"
],
"modelKey": "service"
},
// [Notification strategy name] Injected at send time
"notifyStrategyName": "Script Notification",
// [Notification strategy type] 0=Repeat 1=Escalation; usually "" for the first alert
"strategyType": "",
// [Escalation type] Only when strategyType=1: 0=Not Closed 1=Not Recovered
"upgradeType": "",
// [Raw notification template] JSON string from the rule's notify configuration
"template": "{\"content\":\"\"}",
// [Notification template ID] Corresponds to Alert.templateId; usually "" when not configured
"templateId": "",
// [Entity dimension mapping] dimensionKey → groupKey → entityId
"dimEntityMap": {
"service": {
"serviceId": 848903
}
},
// [Metric display name]
"metric": "Error Count",
// [Metric value that triggered the alert]
"value": 14.0,
// [Entity attribute snapshot] Queried from Titan is_show=1 attributes at send time; modelKey → { attrKey: value }
"entityAttr": {
"service": {
"detectedName": "one_framework.jar",
"instanceId": 848903,
"lifeCycle": 1,
"customizedName": "one 'framework' jar-Custom2"
}
},
// [Entity tag snapshot] modelKey → ["tagKey:tagValue", ...]; serialized as "" when empty
"tagsAttr": {
"service": ["aaaaaaaaaa:aaa"]
},
// [Notification trigger reason] 1=Alert 2=Recovery 3=Close 4=Repeat 5=Escalation
"triggerReason": 1,
// [Exception type] 1=Availability 2=No Data 3=Error 4=Slow 5=Throughput 6=Resource 7=Info 8=Other
"exceptionType": 3
}
Shell Script Example
#!/bin/bash
# Strict validation: if any step fails, stop the script immediately to avoid propagating bad data
set -euo pipefail
# Common log path configuration
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=Chinese, any other value=English
LANG_CODE="en"
# Check whether jq is installed
check_jq() {
if ! command -v jq &> /dev/null; then
echo "jq is not installed, installing automatically..."
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 "Unable to install jq automatically, please install it manually and retry!"
exit 255
fi
fi
}
# Return display text for the status code based on the language parameter
translate_status() {
local status_code="$1"
local lang_code="$2"
case "${status_code}" in
0)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "Resolved"; else echo "Resolved"; fi
;;
1)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "Alerting"; else echo "Alerting"; fi
;;
2)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "No data"; else echo "No data"; fi
;;
3)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "Closed"; else echo "Closed"; fi
;;
4)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "Closed (Manual)"; else echo "Closed (Manual)"; fi
;;
*)
echo "${status_code}"
;;
esac
}
# Return display text for the level code based on the language parameter
translate_level() {
local level_code="$1"
local lang_code="$2"
case "${level_code}" in
2)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "remindful"; else echo "remindful"; fi
;;
3)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "general"; else echo "general"; fi
;;
4)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "warning"; else echo "warning"; fi
;;
5)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "serious"; else echo "serious"; fi
;;
6)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "fatal"; else echo "fatal"; fi
;;
*)
echo "${level_code}"
;;
esac
}
# Return display text for the exceptionType code based on the language parameter
translate_exception_type() {
local type_code="$1"
local lang_code="$2"
case "${type_code}" in
1)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "usability"; else echo "usability"; fi
;;
2)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "nodata"; else echo "nodata"; fi
;;
3)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "error"; else echo "error"; fi
;;
4)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "slow"; else echo "slow"; fi
;;
5)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "throughput"; else echo "throughput"; fi
;;
6)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "resource"; else echo "resource"; fi
;;
7)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "info"; else echo "info"; fi
;;
8)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "other"; else echo "other"; fi
;;
*)
echo "${type_code}"
;;
esac
}
# Return display text for the triggerReason code based on the language parameter
translate_trigger_reason() {
local reason_code="$1"
local lang_code="$2"
case "${reason_code}" in
1)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "TRIGGERED"; else echo "TRIGGERED"; fi
;;
2)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "RECOVERED"; else echo "RECOVERED"; fi
;;
3)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "CLOSED"; else echo "CLOSED"; fi
;;
4)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "RE_TRIGGERED"; else echo "RE_TRIGGERED"; fi
;;
5)
if [[ "${lang_code}" == "zh_CN" ]]; then echo "ESCALATED"; else echo "ESCALATED"; fi
;;
*)
echo "${reason_code}"
;;
esac
}
# Format a millisecond timestamp as YYYY-MM-DD HH:MM:SS; return as-is if already in that format
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 an alert message line; show the label even when the value is empty
append_alert_line() {
local label="$1"
local value="$2"
if [[ "${value}" == "null" ]]; then
value=""
fi
FINAL_ALERT_MSG="${FINAL_ALERT_MSG}${label}: ${value}"$'\n'
}
# Resolve the trigger threshold for the current alert level from conditionConfig.trigger.levels
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. Pre-checks
check_jq
# 2. Parameter validation
json_data="$1"
echo "Raw input: ${json_data}" >> "${RAW_PARAM_LOG}"
if [[ -z "${json_data}" ]]; then
printf "%s" "Parameter cannot be empty"
exit 255
fi
# 3. Base64 decode (handle possible line breaks/spaces)
json_data=$(printf "%s" "${json_data}" | base64 -d 2>/dev/null)
if [[ -z "${json_data}" ]]; then
printf "%s" "Base64 decoding failed or the result is empty"
exit 255
fi
echo "Decoded JSON: ${json_data}" >> "${DECODED_JSON_LOG}"
# 4. Extract the raw status / level / exceptionType / triggerReason codes
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. Translate status / level / exceptionType / triggerReason based on the language parameter
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. Resolve the current alert's trigger threshold from conditionConfig.trigger.levels
ALERT_THRESHOLD=$(resolve_alert_threshold "${json_data}" "${RAW_LEVEL}")
# 7. Extract other fields (empty string "" when missing)
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 // ""')
# Print all variable values to the log (regardless of whether they have a value)
{
echo "========== Decoded JSON field values =========="
echo "Language parameter: ${LANG_CODE}"
echo "Raw status code: ${RAW_STATUS}"
echo "Raw level code: ${RAW_LEVEL}"
echo "Raw exception type code: ${RAW_EXCEPTION_TYPE}"
echo "Raw trigger reason code: ${RAW_TRIGGER_REASON}"
echo "Alert status: ${ALERT_STATUS}"
echo "Exception type: ${ALERT_EXCEPTION_TYPE}"
echo "Trigger reason: ${ALERT_TRIGGER_REASON}"
echo "Environment name: ${ALERT_ENV}"
echo "Alert name: ${ALERT_NAME}"
echo "Host IPv4 address: ${HOST_IPV4_ADDRESS}"
echo "Host name: ${HOST_NAME}"
echo "Alert duration: ${ALERT_DURATION}"
echo "Raw start time: ${RAW_START_TIME}"
echo "Raw update time: ${RAW_UPDATE_TIME}"
echo "Start time: ${ALERT_START_TIME}"
echo "Update time: ${ALERT_UPDATE_TIME}"
echo "Alert level: ${ALERT_LEVEL}"
echo "Trigger threshold: ${ALERT_THRESHOLD}"
echo "Alert ID: ${ALERT_ID}"
echo "Metric: ${ALERT_METRIC}"
} >> "${LOG_DIR}/json_values.log"
# ===================== [Build a single, complete message variable for later use] =====================
FINAL_ALERT_MSG="[Alert Notification]"$'\n'
append_alert_line "Alert Status" "${ALERT_STATUS}"
append_alert_line "Exception Type" "${ALERT_EXCEPTION_TYPE}"
append_alert_line "Trigger Reason" "${ALERT_TRIGGER_REASON}"
append_alert_line "Environment" "${ALERT_ENV}"
append_alert_line "Alert Name" "${ALERT_NAME}"
append_alert_line "Host IPv4 Address" "${HOST_IPV4_ADDRESS}"
append_alert_line "Host Name" "${HOST_NAME}"
append_alert_line "Alert Duration" "${ALERT_DURATION}"
append_alert_line "Start Time" "${ALERT_START_TIME}"
append_alert_line "Update Time" "${ALERT_UPDATE_TIME}"
append_alert_line "Alert Level" "${ALERT_LEVEL}"
append_alert_line "Trigger Threshold" "${ALERT_THRESHOLD}"
append_alert_line "Alert ID" "${ALERT_ID}"
append_alert_line "Metric" "${ALERT_METRIC}"
FINAL_ALERT_MSG="${FINAL_ALERT_MSG%"${FINAL_ALERT_MSG##*[![:space:]]}"}"
# 9. Build the alert content function
get_content(){
echo "${FINAL_ALERT_MSG}"
}
# 10. Send the notification
send_notification() {
printf "%s\n" "start send notification"
# Build the JSON with jq to avoid special characters breaking the curl request body
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"
}
# Execute the core logic
send_notification
get_content
exit 0
Python Script Example
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Alert notification script (native Python implementation, no third-party dependencies).
Only supports Python 3.5+; if run under Python 2, it logs an error and exits immediately.
"""
import base64
import binascii
import datetime
import json
import os
import re
import sys
import urllib.parse
# Log directory (defined before the version check so Python 2 can still write a log on exit)
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 = "This script only supports Python 3; the current interpreter is Python 2. Please run it with 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")
# Notification endpoint URL
#NOTIFY_URL = "https://xxx/test"
NOTIFY_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
# Language parameter (hardcoded in the script): zh_CN=Chinese, any other value=English
LANG_CODE = "en"
STATUS_TEXT = {
"zh_CN": {
"0": "Resolved",
"1": "Alerting",
"2": "No data",
"3": "Closed",
"4": "Closed (Manual)",
},
"en": {
"0": "Resolved",
"1": "Alerting",
"2": "No data",
"3": "Closed",
"4": "Closed (Manual)",
},
}
LEVEL_TEXT = {
"zh_CN": {
"2": "remindful",
"3": "general",
"4": "warning",
"5": "serious",
"6": "fatal",
},
"en": {
"2": "remindful",
"3": "general",
"4": "warning",
"5": "serious",
"6": "fatal",
},
}
EXCEPTION_TYPE_TEXT = {
"zh_CN": {
"1": "usability",
"2": "nodata",
"3": "error",
"4": "slow",
"5": "throughput",
"6": "resource",
"7": "info",
"8": "other",
},
"en": {
"1": "usability",
"2": "nodata",
"3": "error",
"4": "slow",
"5": "throughput",
"6": "resource",
"7": "info",
"8": "other",
},
}
TRIGGER_REASON_TEXT = {
"zh_CN": {
"1": "TRIGGERED",
"2": "RECOVERED",
"3": "CLOSED",
"4": "RE_TRIGGERED",
"5": "ESCALATED",
},
"en": {
"1": "TRIGGERED",
"2": "RECOVERED",
"3": "CLOSED",
"4": "RE_TRIGGERED",
"5": "ESCALATED",
},
}
def append_log(log_file, message):
"""Append a single log line."""
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):
"""Append multiple log 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-decode the input parameter."""
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):
"""Fix common formatting issues in the script notification 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):
"""Extract a field from the JSON, returning an empty string when missing or 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):
"""Format a millisecond timestamp as YYYY-MM-DD HH:MM:SS; return as-is if already in that format."""
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):
"""Return display text for the status code based on the language parameter."""
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):
"""Return display text for the level code based on the language parameter."""
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):
"""Return display text for the exceptionType code based on the language parameter."""
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):
"""Return display text for the triggerReason code based on the language parameter."""
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):
"""Resolve the trigger threshold for the current alert level from conditionConfig.trigger.levels."""
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):
"""Build the complete alert message text; show the label even when the value is empty."""
lines = ["[Alert Notification]"]
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 the alert content (for external use)."""
return final_alert_msg
def send_notification(content_text):
"""Send the HTTP notification."""
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("Parameter cannot be empty", end="")
return 255
raw_param = sys.argv[1]
append_log(RAW_PARAM_LOG, "Raw input: {0}".format(raw_param))
if not raw_param:
print("Parameter cannot be empty", end="")
return 255
json_text = decode_base64_param(raw_param)
if not json_text:
print("Base64 decoding failed or the result is empty", end="")
return 255
append_log(DECODED_JSON_LOG, "Decoded JSON: {0}".format(json_text))
json_text = normalize_json_text(json_text)
try:
json_data = json.loads(json_text)
except ValueError:
print("JSON parsing failed", end="")
return 255
if not isinstance(json_data, dict):
print("The JSON root must be an object", 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,
[
"========== Decoded JSON field values ==========",
"Language parameter: {0}".format(LANG_CODE),
"Raw status code: {0}".format(raw_status),
"Raw level code: {0}".format(raw_level),
"Raw exception type code: {0}".format(raw_exception_type),
"Raw trigger reason code: {0}".format(raw_trigger_reason),
"Alert status: {0}".format(alert_status),
"Exception type: {0}".format(alert_exception_type),
"Trigger reason: {0}".format(alert_trigger_reason),
"Environment name: {0}".format(alert_env),
"Alert name: {0}".format(alert_name),
"Host IPv4 address: {0}".format(host_ipv4_address),
"Host name: {0}".format(host_name),
"Alert duration: {0}".format(alert_duration),
"Raw start time: {0}".format(raw_start_time),
"Raw update time: {0}".format(raw_update_time),
"Start time: {0}".format(alert_start_time),
"Update time: {0}".format(alert_update_time),
"Alert level: {0}".format(alert_level),
"Trigger threshold: {0}".format(alert_threshold),
"Alert ID: {0}".format(alert_id),
"Metric: {0}".format(alert_metric),
],
)
message_field_items = (
("Alert Status", alert_status),
("Exception Type", alert_exception_type),
("Trigger Reason", alert_trigger_reason),
("Environment", alert_env),
("Alert Name", alert_name),
("Host IPv4 Address", host_ipv4_address),
("Host Name", host_name),
("Alert Duration", alert_duration),
("Start Time", alert_start_time),
("Update Time", alert_update_time),
("Alert Level", alert_level),
("Trigger Threshold", alert_threshold),
("Alert ID", alert_id),
("Metric", 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())
Common Scenarios
Scenario: Integrating with an internal self-built IM system Write a Python script that receives the alert parameters passed in by the platform and calls the internal IM system's API to complete message delivery, filling the gap for an IM type the platform doesn't natively support.
Scenario: Triggering an internal ticketing system Write a Shell script that, upon receiving an Error-level or higher alert, automatically calls the internal ticketing system's API to create a ticket, achieving an automated flow from alert to ticket.
Notes
danger
The script runs with the permissions of the server on which the deployment environment resides. Do not introduce unaudited third-party code or perform high-risk operations in the script, to avoid introducing security risks.
info
If you later upgrade or migrate the deployment environment, make sure the script file and its dependencies (interpreter version, third-party libraries, etc.) have been migrated as well — otherwise Script Notification will not execute correctly.