Try this script, it works for me:
#!/usr/bin/env python3
# Copyright (C) 2019 Checkmk GmbH - License: GNU General Public License v2
# This file is part of Checkmk (https://checkmk.com). It is subject to the terms and
# conditions defined in the file COPYING, which is part of this source code package.
import sys
import ast
import requests
from cmk.notification_plugins import utils
api_url = "https://api.pushover.net/1/messages.json"
# Mapping from Checkmk priority names to Pushover priority values
PRIORITY_MAP = {
"lowest": "-2",
"low": "-1",
"normal": "0",
"high": "1",
"emergency": "2"
}
def main() -> int:
context = utils.collect_context()
subject = get_subject(context)
text = get_text(context)
api_key = context["PARAMETER_API_KEY"]
recipient_key = context["PARAMETER_RECIPIENT_KEY"]
return send_push_notification(api_key, recipient_key, subject, text, context)
def get_subject(context: dict[str, str]) -> str:
s = context["HOSTNAME"]
if context["WHAT"] != "HOST":
s += "/" + context["SERVICEDESC"]
s += " "
notification_type = context["NOTIFICATIONTYPE"]
if notification_type in ["PROBLEM", "RECOVERY"]:
s += "$PREVIOUS@HARDSHORTSTATE$ \u2192 $@SHORTSTATE$"
elif notification_type.startswith("FLAP"):
if "START" in notification_type:
s += "Started Flapping"
else:
s += "Stopped Flapping ($@SHORTSTATE$)"
elif notification_type.startswith("DOWNTIME"):
what = notification_type[8:].title()
s += "Downtime " + what + " ($@SHORTSTATE$)"
elif notification_type == "ACKNOWLEDGEMENT":
s += "Acknowledged ($@SHORTSTATE$)"
elif notification_type == "CUSTOM":
s += "Custom Notification ($@SHORTSTATE$)"
else:
s += notification_type
return utils.substitute_context(s.replace("@", context["WHAT"]), context)
def get_text(context: dict[str, str]) -> str:
s = ""
s += "$@OUTPUT$"
if "PARAMETER_URL_PREFIX_1" in context:
s += " <i>Link: </i>"
s += utils.format_link(
'<a href="%s">%s</a>', utils.host_url_from_context(context), context["HOSTNAME"]
)
if context["WHAT"] != "HOST":
s += utils.format_link(
'<a href="%s">%s</a>',
utils.service_url_from_context(context),
context["SERVICEDESC"],
)
return utils.substitute_context(s.replace("@", context["WHAT"]), context)
def get_priority_value(context: dict[str, str]) -> str | None:
"""Extract priority value from various context fields"""
priority = (
context.get("PARAMETER_PRIORITY")
or context.get("priority")
or context.get("PARAMETER_PRIORITY_PRIORITY")
)
if not priority:
return None
# If it's a string representation of a tuple, parse it
if isinstance(priority, str) and priority.startswith("(") and priority.endswith(")"):
try:
priority = ast.literal_eval(priority)
except:
return None
# Tuple format: ('emergency', (60.0, 120.0, 'receipts')) or ('normal', None)
if isinstance(priority, tuple) and len(priority) >= 1:
priority_name = priority[0].lower()
return PRIORITY_MAP.get(priority_name, None)
# Fallback for string values
priority_str = str(priority).lower()
if priority_str in ["-2", "-1", "0", "1", "2"]:
return priority_str
return PRIORITY_MAP.get(priority_str, None)
def get_emergency_params(context: dict[str, str]) -> tuple[int, int, str | None]:
"""Extract emergency parameters from priority tuple or context"""
priority = (
context.get("PARAMETER_PRIORITY")
or context.get("priority")
or context.get("PARAMETER_PRIORITY_PRIORITY")
)
if not priority:
return 60, 3600, None
# Parse string representation of tuple
if isinstance(priority, str) and priority.startswith("("):
try:
priority = ast.literal_eval(priority)
except:
return 60, 3600, None
# Try to extract parameters from tuple
if isinstance(priority, tuple) and len(priority) >= 2 and isinstance(priority[1], tuple):
emergency_params = priority[1]
if len(emergency_params) >= 2:
retry = int(emergency_params[0]) if emergency_params[0] else 60
expire = int(emergency_params[1]) if emergency_params[1] else 3600
receipts = emergency_params[2] if len(emergency_params) > 2 else None
return retry, expire, receipts
# Fallback to context parameters
retry = int(context.get("PARAMETER_PRIORITY_RETRY", 60))
expire = int(context.get("PARAMETER_PRIORITY_EXPIRE", 3600))
receipts = context.get("PARAMETER_PRIORITY_RECEIPTS")
return retry, expire, receipts
def send_push_notification(
api_key: str, recipient_key: str, subject: str, text: str, context: dict[str, str]
) -> int:
params: list[tuple[str, str | int | bytes]] = [
("token", api_key),
("user", recipient_key),
("title", subject.encode("utf-8")),
("message", text.encode("utf-8")),
("timestamp", int(float(context["MICROTIME"]) / 1000000.0)),
("html", 1),
]
priority = get_priority_value(context)
if priority is not None:
params.append(("priority", int(priority)))
# Special handling for emergency (priority 2)
if priority == "2":
retry, expire, receipts = get_emergency_params(context)
params.append(("expire", expire))
params.append(("retry", retry))
if receipts:
params.append(("receipts", receipts))
if context.get("PARAMETER_SOUND", "none") != "none":
params.append(("sound", context["PARAMETER_SOUND"]))
proxy_url = context.get("PARAMETER_PROXY_URL")
proxies = {"https": proxy_url} if proxy_url else None
session = requests.Session()
try:
response = session.post(
api_url,
params=dict(params),
proxies=proxies,
)
except requests.exceptions.ProxyError:
sys.stdout.write("Cannot connect to proxy: %s\n" % context["PARAMETER_PROXY_URL"])
return 1
except requests.exceptions.RequestException:
sys.stdout.write("POST request to server failed: %s\n" % api_url)
return 1
if response.status_code not in [200, 204]:
sys.stdout.write(
f"Failed to send notification. Status: {response.status_code}, Response: {response.text}\n"
)
return 1
try:
data = response.json()
except ValueError:
sys.stdout.write("Failed to decode JSON response: %s\n" % response.text)
return 1
if data.get("status") != 1:
sys.stdout.write("Received an error from the Pushover API: %s" % response.text)
return 1
return 0
My pushover.py is mapped into the Docker container, making it update-safe and persistent across Checkmk updates.