Spearhead/checkmk-plugins
14
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Updated eCall plugin for CheckMK v2.3 to work with v2.4.

Browse Source
This commit is contained in:
Marsell Kukuljevic 2026-03-24 15:27:13 +01:00
parent fed4e940c0
commit f754cd2493
6 changed files with 925 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

112
ecall/2.4/local/lib/python3/cmk_addons/plugins/ecall/lib/api.py Normal file
View File

@ -0,0 +1,112 @@
#!/usr/bin/env python3
import logging
import requests
from urllib3.exceptions import InsecureRequestWarning
class DeliveryException(Exception):
pass
class API(object):
"""
Provide basic access to eCall's HTTP/HTTPS API
"""
def __init__(
self,
baseurl,
username,
password,
address,
message,
requests_proxy_hosts=None,
requests_verify_cert=True,
**kwargs,
):
self._logger = logging.getLogger("ecall_plugin")
self._logger.info("Using %s" % self.__class__.__name__)
# eCall arguments
self.baseurl = baseurl
self.username = username
self.password = password
self.address = address
self.message = message
self.params = kwargs
# python requests arguments
self.proxy_hosts = requests_proxy_hosts
self.verify_cert = requests_verify_cert if not requests_proxy_hosts else False
if not self.verify_cert:
# disable InsecureRequestWarning on unverified SSL requests
requests.packages.urllib3.disable_warnings(
category=InsecureRequestWarning
) # pylint: disable=no-member
def _get_query_params(self):
base = {
"Address": self.address,
"UserName": self.username,
"Password": self.password,
"Message": self.message,
}
base.update(self.params)
if self._logger.isEnabledFor(
logging.DEBUG
): # check for log level to not run map on every request
# log params but hide password entry
self._logger.debug(
"Query parameters are: %s",
str(dict(map(lambda i: (i[0], "***") if i[0] == "Password" else i, base.items()))),
)
return base
def send(self):
"""
Send message to eCall API via HTTPS.
"""
self._logger.info("Sending message")
self._logger.debug("Using base url %s", self.baseurl)
self._logger.debug("Verifying certificates: %s", self.verify_cert)
self._logger.debug("Using proxies: %s", self.proxy_hosts)
if not self.address:
self._logger.warning("Got no or empty address information, skipping this notification")
else:
ret = requests.get(
self.baseurl,
params=self._get_query_params(),
verify=self.verify_cert,
proxies=self.proxy_hosts,
)
if ret.status_code == 200:
self._logger.info("Successfully delivered message contents to eCall")
else:
msg = "Unable to deliver message to eCall: %i, Error: %s" % (
ret.status_code,
ret.text,
)
self._logger.critical(msg)
raise DeliveryException(msg)
class SMS(API):
def __init__(self, username, password, address, message, **kwargs):
super(SMS, self).__init__(
"https://url.ecall.ch/api/sms", username, password, address, message, **kwargs
)
class Voice(API):
def __init__(self, username, password, address, message, **kwargs):
super(Voice, self).__init__(
"https://url.ecall.ch/api/voice", username, password, address, message, **kwargs
)

390
ecall/2.4/local/lib/python3/cmk_addons/plugins/ecall/rulesets/notify_ecall_parameters.py Normal file
View File

@ -0,0 +1,390 @@
#!/usr/bin/env python3
from cmk.rulesets.v1.form_specs import (
Dictionary,
DictElement,
String,
Integer,
Password,
BooleanChoice,
MultilineText,
CascadingSingleChoice,
CascadingSingleChoiceElement,
SingleChoice,
SingleChoiceElement,
DefaultValue,
)
from cmk.rulesets.v1.rule_specs import (
Title,
Help,
NotificationParameters,
Topic,
)
from cmk.rulesets.v1.form_specs.validators import (
LengthInRange,
NumberInRange,
NetworkPort,
ValidationError,
)
class CallbackLength:
def __call__(self, value) -> None:
try:
# only numeric -> maxlen 16
int(value)
maxlen = 16
errtype = "numeric"
except ValueError:
# alphanumeric -> maxlen 11
maxlen = 11
errtype = "alphanumeric"
if len(value) > maxlen:
raise ValidationError(
"For callbacks using %s characters the maximum length is %i" % (errtype, maxlen),
)
def _params_form():
return Dictionary(
title=Title("Call with the following parameters"),
elements={
"action": DictElement(
required=True,
parameter_form=CascadingSingleChoice(
title=Title("eCall action"),
help_text=Help("Configure the action to take by eCall"),
prefill=DefaultValue("sms"),
elements=[
CascadingSingleChoiceElement(
name="sms",
title=Title("SMS"),
parameter_form=Dictionary(
title=Title("Custom parameters"),
help_text=Help(
"Specify any custom parameters to be sent to the eCall API"
),
elements={
"JobID": DictElement(
parameter_form=String(
title=Title("JobID"),
custom_validate=[LengthInRange(min_value=1, max_value=50)],
),
),
"NotificationAddress": DictElement(
parameter_form=String(
title=Title("NotificationAddress"),
custom_validate=[LengthInRange(min_value=1, max_value=100)],
),
),
"NotificationLevel": DictElement(
parameter_form=SingleChoice(
title=Title("NotificationLevel"),
elements=[
SingleChoiceElement(
name="level_0",
title=Title("Confirmation of receipt only when a receipt status is present"),
),
SingleChoiceElement(
name="level_1",
title=Title("Confirmation of receipt once the last possible monitoring point has been reached"),
),
SingleChoiceElement(
name="level_2",
title=Title("Confirmation of receipt once the last possible monitoring point has been reached, and when the job is still not sent after a number of seconds"),
),
SingleChoiceElement(
name="level_3",
title=Title("Send confirmation of receipt when the job could not be sent, i.e. in the case of sending errors or timeout when contacting the end device"),
),
],
prefill=DefaultValue("level_0"),
),
),
"CallBack": DictElement(
parameter_form=String(
title=Title("CallBack"),
custom_validate=[LengthInRange(min_value=1, max_value=16), CallbackLength()],
),
),
"Answer": DictElement(
parameter_form=String(
title=Title("Answer"),
custom_validate=[LengthInRange(min_value=1, max_value=100)],
),
),
"MsgType": DictElement(
parameter_form=SingleChoice(
title=Title("MsgType"),
elements=[
SingleChoiceElement(name="Normal", title=Title("Normal SMS")),
SingleChoiceElement(name="Flash", title=Title("Flash SMS")),
SingleChoiceElement(name="PrioSMS", title=Title("Flash and a 'Normal' SMS"))
],
prefill=DefaultValue("Normal"),
),
),
"NoLog": DictElement(
required=True,
parameter_form=BooleanChoice(
title=Title("Prevent message from showing up in the log book"),
prefill=DefaultValue(True),
),
),
},
),
),
CascadingSingleChoiceElement(
name="voice",
title=Title("Voice"),
parameter_form=Dictionary(
title=Title("Custom parameters"),
help_text=Help(
"Set custom paramters to be sent additionally to the eCall API"
),
elements={
"JobID": DictElement(
parameter_form=String(
title=Title("JobID"),
custom_validate=[LengthInRange(min_value=1, max_value=50)],
)
),
"Language": DictElement(
parameter_form=SingleChoice(
title=Title("Language"),
elements=[
SingleChoiceElement(name="DE", title=Title("German")),
SingleChoiceElement(name="FR", title=Title("French")),
SingleChoiceElement(name="IT", title=Title("Italian")),
SingleChoiceElement(name="EN", title=Title("English")),
],
prefill=DefaultValue("EN"),
),
),
"FromText": DictElement(
parameter_form=String(
title=Title("FromText"),
custom_validate=[LengthInRange(min_value=1)],
),
),
},
),
),
],
),
),
"username": DictElement(
required=True,
parameter_form=Password(
title=Title("eCall account name"),
help_text=Help("The user account used to authenticate against the eCall API."),
custom_validate=[LengthInRange(min_value=1)],
),
),
"password": DictElement(
required=True,
parameter_form=Password(
title=Title("eCall account password"),
help_text=Help("The password to authenticate against eCall"),
custom_validate=[LengthInRange(min_value=1)],
),
),
"content": DictElement(
required=True,
parameter_form=Dictionary(
title=Title("Message content"),
help_text=Help(
"Configure the content to be sent to eCall. Not setting any content will result in an empty message."
),
elements={
"shared": DictElement(
required=True,
parameter_form=MultilineText(
title=Title("Content head for host and service notifications"),
help_text=Help(
"""You may use any checkmk macro here, as well as the following special macros:
<tt>$PARSED_OUTPUT$</tt>: Output of either host or service check depending on notification type
<tt>$PARSED_STATE$</tt>: State of either host or service depending on notification type
<tt>$PARSED_ITEM$</tt>: Item name of the affected monitored object, either hostname/service or service
"""
),
monospaced=True,
prefill=DefaultValue("$PARSED_ITEM$ $PARSED_OUTPUT$"),
),
),
"host": DictElement(
parameter_form=MultilineText(
title=Title("Add information in case of host notifications"),
monospaced=True,
prefill=DefaultValue("""Event: $EVENT_TXT$
Output: $HOSTOUTPUT$
Perfdata: $HOSTPERFDATA$
$LONGHOSTOUTPUT$
"""),
),
),
"service": DictElement(
parameter_form=MultilineText(
title=Title("Add information in case of service notifications"),
monospaced=True,
prefill=DefaultValue("""Service: $SERVICEDESC$
Event: $EVENT_TXT$
Output: $SERVICEOUTPUT$
Perfdata: $SERVICEPERFDATA$
$LONGSERVICEOUTPUT$
"""),
),
),
"bulk_prefix": DictElement(
parameter_form=SingleChoice(
title=Title("Prefix for bulk notification messages"),
help_text=Help(
"This setting allows to define a prefix that will be prepended to bulk notification messages."
),
elements=[
SingleChoiceElement(name="HOSTGROUPNAMES", title=Title("Use host groups as prefix")),
SingleChoiceElement(name="SERVICEGROUPNAMES", title=Title("Use service groups as prefix")),
],
prefill=DefaultValue("HOSTGROUPNAMES"),
),
),
"show_host_if_service": DictElement(
required=True,
parameter_form=BooleanChoice(
title=Title(
"Show hostname in PARSED_ITEM for service notifications"
),
help_text=Help(
"When this option is set, the hostname will be shown in the PARSED_ITEM macro. Otherwise it won't."
),
prefill=DefaultValue(True),
),
),
},
),
),
"content_length_limit": DictElement(
required=True,
parameter_form=Integer(
title=Title("Content length limit"),
help_text=Help("Cut the message content after this count of characters"),
custom_validate=[NumberInRange(min_value=1, max_value=1530)], # Max value for GSM
prefill = DefaultValue(160),
),
),
"proxy": DictElement(
parameter_form=Dictionary(
title=Title("Proxy servers"),
help_text=Help(
"Proxy servers that should be used when connecting to eCall via HTTPs"
),
elements={
"host": DictElement(
required=True,
parameter_form=String(
title=Title("Hostname / IP address"),
help_text=Help("The hostname or IP address of the proxy system"),
custom_validate=[LengthInRange(min_value=1)],
),
),
"protocol": DictElement(
required=True,
parameter_form=SingleChoice(
title=Title("Proxy protocol"),
elements=[
SingleChoiceElement(name="http", title=Title("http")),
SingleChoiceElement(name="https", title=Title("https")),
],
prefill=DefaultValue("http"),
),
),
"port": DictElement(
required=True,
parameter_form=Integer(
title=Title("Port"),
help_text=Help("The port the proxy is serving requests on"),
custom_validate=[NetworkPort()]
),
),
"authentication": DictElement(
required=True,
parameter_form=Dictionary(
title=Title("Authentication"),
elements={
"user": DictElement(
parameter_form=String(
title=Title("Username"),
help_text=Help(
"The username to authenticate on the proxy"
),
custom_validate=[LengthInRange(min_value=1)],
),
),
"password": DictElement(
parameter_form=Password(
title=Title("Password"),
help_text=Help(
"The password to authenticate on the proxy"
),
custom_validate=[LengthInRange(min_value=1)],
),
),
},
),
),
},
),
),
"ssl_skip_verify": DictElement(
required=True,
parameter_form=BooleanChoice(
title=Title("Disable SSL certificate verification"),
help_text=Help(
"Disables verification of SSL certificates. This is useful when behind transparent proxies. When a proxy is configured, this is automatically enabled."
),
prefill=DefaultValue(True),
),
),
"log_file": DictElement(
required=True,
parameter_form=String(
title=Title("Log file"),
help_text=Help("The log file to log to"),
custom_validate=[LengthInRange(min_value=1)],
prefill=DefaultValue("~/var/log/ecall.log"),
),
),
"log_level": DictElement(
required=True,
parameter_form=SingleChoice(
title=Title("Log level"),
help_text=Help("Set the minimum log level to show logs for."),
elements=[
SingleChoiceElement(name="d", title=Title("Debug")),
SingleChoiceElement(name="i", title=Title("Info")),
SingleChoiceElement(name="w", title=Title("Warning")),
SingleChoiceElement(name="c", title=Title("Critical")),
],
prefill=DefaultValue("i"),
),
),
"stdout": DictElement(
required=True,
parameter_form=BooleanChoice(
title=Title("Log to stdout"),
help_text=Help(
"Enable log output to stdout. This will be seen as lines prefixed by <tt>Output</tt> in <tt>var/log/notify.log</tt>"
),
prefill=DefaultValue(True),
),
),
},
)
rule_spec_tbs_ticket = NotificationParameters(
name="eCall",
title=Title("eCall – Parameters"),
topic=Topic.NOTIFICATIONS,
parameter_form=_params_form,
)

423
ecall/2.4/local/share/check_mk/notifications/eCall Executable file
View File

@ -0,0 +1,423 @@
#!/usr/bin/env python3
# eCall
# Bulk: yes
# -*- coding: utf-8 -*-
"eCall SMS notification plugin for Check_MK monitoring systems using rulebased notifications"
import argparse
import logging
import os
from urllib.parse import quote
from logging.handlers import TimedRotatingFileHandler
from cmk.notification_plugins.utils import (
collect_context,
read_bulk_contexts,
get_password_from_env_or_context,
substitute_context,
)
from cmk_addons.plugins.ecall.lib.api import SMS, Voice
class MessageFormat(object):
def __init__(self, shared, service=None, host=None):
self._shared = shared
self._service = service
self._host = host
@property
def service(self):
if self._shared and self._service:
return "{}\n{}".format(self._shared, self._service)
elif self._service: # shared is empty
return self._service
return self._shared
@property
def host(self):
if self._shared and self._host:
return "{}\n{}".format(self._shared, self._host)
elif self._host: # shared is empty
return self._host
return self._shared
class SettingsException(Exception):
pass
class Settings(object):
def __init__(self, notification_parameters):
p = notification_parameters
# keys that may just be copied:
# username, password, log_file, log_level, stdout
# handle simple mandatory parameters
self.username = get_password_from_env_or_context("USERNAME", p)
self.password = get_password_from_env_or_context("PASSWORD", p)
# in case of using the store, there will be None values when changes have not been activated
if self.username is None:
raise SettingsException(
"Got empty username. Did you activate changes after adding the user to the store?"
)
if self.password is None:
raise SettingsException(
"Got empty password. Did you activate changes after adding the password to the store?"
)
self.message_length_limit = int(p["CONTENT_LENGTH_LIMIT"])
self.log_file = os.path.expanduser(p["LOG_FILE"])
self.log_level = {
"d": logging.DEBUG,
"i": logging.INFO,
"w": logging.WARNING,
"c": logging.CRITICAL,
}.get(p["LOG_LEVEL"])
# handle simple optional parameters
if "PROXY_HOST" in p:
proto = p["PROXY_PROTOCOL"]
host = p["PROXY_HOST"]
port = p["PROXY_PORT"]
user = p.get("PROXY_AUTHENTICATION_USER")
if user:
user = quote(user)
if p.get("PROXY_AUTHENTICATION_PASSWORD_1"):
passwd = get_password_from_env_or_context("PROXY_AUTHENTICATION_PASSWORD", p)
passwd = quote(passwd)
if user and passwd:
url = f"{proto}://{user}:{passwd}@{host}:{port}"
elif user:
url = f"{proto}://{user}@{host}:{port}"
else:
url = f"{proto}://{host}:{port}"
self.proxy = {
"https": url
#proto: url
}
else:
self.proxy = None
if "SSL_SKIP_VERIFY" in p and p["SSL_SKIP_VERIFY"] == "True":
self.ssl_skip_verify = True
else:
self.ssl_skip_verify = False
if "STDOUT" in p and p["STDOUT"] == "True":
self.log_stdout = True
else:
self.log_stdout = False
# handle complex mandatory parameters
## parse message format
self.bulk_message_prefix = ""
if "CONTENT_BULK_PREFIX" in p:
self.bulk_message_prefix = p["CONTENT_BULK_PREFIX"]
message_format_shared = p["CONTENT_SHARED"]
message_format_specifc = {}
if "CONTENT_HOST" in p:
message_format_specifc["host"] = p["CONTENT_HOST"]
if "CONTENT_SERVICE" in p:
message_format_specifc["service"] = p["CONTENT_SERVICE"]
self.message_format = MessageFormat(message_format_shared, **message_format_specifc)
## iterate over every action parameter key in params
self.api_class = SMS if p["ACTION_1"] == "sms" else Voice
self.api_params = {}
for param_key in filter(lambda k: k.startswith("ACTION_2_"), p.keys()):
key = param_key.split("_")[2].lower()
val = p[param_key]
self.api_params[key] = val
if level := self.api_params.get("notificationlevel"):
self.api_params["notificationlevel"] = level.split("_")[1]
if nolog := self.api_params.get("nolog"):
if nolog == "True":
self.api_params["nolog"] = 1
else:
self.api_params["nolog"] = 0
class Notification(object):
def __init__(self, context, message_format, message_length_limit):
self._logger = logging.getLogger("ecall_plugin")
self._context = context
self._message_format = message_format
self._length_limit = message_length_limit
self.contact = context["CONTACTPAGER"]
def _limit(self, text):
self._logger.info("Limiting message length to at most %i characters", self._length_limit)
return text[: self._length_limit]
@property
def message(self):
if self._context["WHAT"] == "SERVICE":
message = substitute_context(self._message_format.service, self._context)
else:
message = substitute_context(self._message_format.host, self._context)
self._logger.debug("Unlimited message is '%s'", message)
return self._limit(message)
class BulkNotification(Notification):
def __init__(self, context_list, message_format, message_length_limit, bulk_message_prefix):
# call super to get shared properties
# just use the first context as reference for contact
# the official plugins also assume that contact is the same for every context
super(BulkNotification, self).__init__(
context_list[0], message_format, message_length_limit
)
self._prefix = bulk_message_prefix
self._context_list = context_list
def __format_message(self):
prefixes = []
affected_hosts = [] # TODO: This is currently not being used
service_states = {}
host_states = {}
self._logger.info("Processing bulk context list")
for context in self._context_list:
self._logger.debug("Processing context %s", context)
affected_host = context["HOSTNAME"]
# keep track of prefixes (if any)
if self._prefix:
# prefix is either HOSTGROUPNAMES or SERVICEGROUPNAMES
# Both are lists of group names separated by whitespace
prefixes += context[self._prefix].split(" ")
# fill list of affected hosts
if affected_host not in affected_hosts:
affected_hosts.append(affected_host)
# keep track of states
# TODO: We could code logic here to only send the most recent, worst state for each item
if context["WHAT"] == "SERVICE":
if context["PARSED_STATE"] not in service_states:
service_states[context["PARSED_STATE"]] = 1
else:
service_states[context["PARSED_STATE"]] += 1
else:
if context["PARSED_STATE"] not in host_states:
host_states[context["PARSED_STATE"]] = 1
else:
host_states[context["PARSED_STATE"]] += 1
# consolidate prefixes
prefixes = list(set(prefixes))
prefix_info = ""
if prefixes:
prefix_info = ", ".join(prefixes)
self._logger.debug("Got the following prefix for bulk message: %s", prefix_info)
prefix_info += ": "
# generate message
self._logger.debug("Raw service states are: %s", service_states)
self._logger.debug("Raw host states are: %s", host_states)
service_info = map(lambda i: "%s: %i" % (i[0].capitalize(), i[1]), service_states.items())
host_info = map(lambda i: "%s: %i" % (i[0].capitalize(), i[1]), host_states.items())
message_items = []
if host_info:
self._logger.debug("Processed host info is: %s", host_info)
message_items.append("Hosts %s" % ", ".join(host_info))
if service_info:
self._logger.debug("Processed service info is: %s", service_info)
message_items.append("Services %s" % ", ".join(service_info))
return prefix_info + ", ".join(message_items)
@property
def message(self):
# use base class when there is only one notification in bulk context
if len(self._context_list) == 1:
self._logger.info(
"Only one context in bulk notification, use default notification method"
)
return super(BulkNotification, self).message
return self._limit(self.__format_message())
class ContextParser(object):
"Parse variable inputs from checkmk notifier"
def __init__(self):
argument_parser = argparse.ArgumentParser(
description="Send eCall SMS from checkmk monitoring notifications."
)
argument_parser.add_argument(
"--bulk", help="Enable checkmk bulk mode", action="store_true", default=False
)
args = argument_parser.parse_args()
# TODO: maybe this should be configurable in the GUI.
# right now, the only notificationtype left is "PROBLEM".
# Only for "PROBLEM" the message format is actually being
# used. This is not optimal. May be let the user set a
# default message format and format and state per type of
# notification? This would be more flexible.
self.__special_states = {
"FLAPPINGSTART": ("WARNING", "started flapping"),
"FLAPPINGSTOP": ("OK", "stopped flapping"),
"ACKNOWLEDGEMENT": ("OK", "acknowledged"),
"RECOVERY": ("OK", "recovered"),
"DOWNTIMESTART": ("OK", "downtime started"),
"DOWNTIMECANCELLED": ("OK", "downtime cancelled"),
"DOWNTIMEEND": ("OK", "downtime ended"),
}
self.bulk = args.bulk
if args.bulk:
self.params, self.context = self.__parse_bulk()
else:
self.params, self.context = self.__parse_rulebased()
def __parse_filter_context(self, context, params, is_bulk=False):
"Add custom macros to the context and strip out any parameters"
what = context["WHAT"]
notification_type = context["NOTIFICATIONTYPE"]
context["PARSED_ITEM"] = context["HOSTNAME"]
if what == "SERVICE":
if params.get("SHOW_HOST_IF_SERVICE", "True") == "True":
context["PARSED_ITEM"] = "%s on %s" % (
context["SERVICEDESC"],
context["PARSED_ITEM"],
)
else:
context["PARSED_ITEM"] = context["SERVICEDESC"]
if notification_type in self.__special_states:
context["PARSED_STATE"], context["PARSED_OUTPUT"] = self.__special_states[
notification_type
]
if is_bulk:
# Since we are merely counting the notifications for each state
# it makes no sense to keep the original OK/WARN/CRIT/UP/DOWN here.
# Instead, we overwrite it with the pretty output (e.g. "started flapping")
# so that this is counted in the bulk message.
# e.g. Host Started flapping: 1, Downtime ended: 2, Down: 2, Service OK: 3, acknowledged: 4
context["PARSED_STATE"] = context["PARSED_OUTPUT"]
else:
context["PARSED_OUTPUT"] = context["%sOUTPUT" % what]
context["PARSED_STATE"] = context["%sSTATE" % what]
# return new context without parameters (since they would expose passwords)
return dict(filter(lambda i: not i[0].startswith("PARAMETER_"), context.items()))
def __parse_parameters(self, context):
params = {}
for param_key in filter(lambda k: k.startswith("PARAMETER_"), context.keys()):
params[param_key[10::]] = context[param_key]
return params
def __parse_rulebased(self):
context = collect_context()
params = self.__parse_parameters(context)
return params, self.__parse_filter_context(context, params)
def __parse_bulk(self):
params, context_list = read_bulk_contexts()
params = self.__parse_parameters(params)
# add custom fields to bulk contexts
# when there are no more than 1 contexts in the list, do not use bulk mode (--> is_bulk=False)
enriched_contexts = [
self.__parse_filter_context(context, params, is_bulk=len(context_list) > 1)
for context in context_list
]
return params, enriched_contexts
class Plugin(object):
"""
Check_MK notification plugin to notify events using eCall's HTTP/S API.
"""
def __init__(self):
parser = ContextParser()
settings = Settings(parser.params)
self._logger = logging.getLogger("ecall_plugin")
self.__configure_logging(settings.log_level, settings.log_file, settings.log_stdout)
self._logger.info("Starting eCall notification process")
self._logger.debug("Notification context is %s", parser.context)
if parser.bulk:
notification = BulkNotification(
parser.context,
settings.message_format,
settings.message_length_limit,
settings.bulk_message_prefix,
)
else:
notification = Notification(
parser.context, settings.message_format, settings.message_length_limit
)
self._logger.debug("Formatted message is %s", notification.message)
self._logger.debug("Initialising eCall API Object")
self.api = settings.api_class(
settings.username,
settings.password,
notification.contact,
notification.message,
requests_proxy_hosts=settings.proxy,
requests_verify_cert=not settings.ssl_skip_verify,
**settings.api_params,
)
self._logger.debug("Initialisation done")
def __configure_logging(self, log_level, log_file, log_stdout):
self._logger.setLevel(log_level)
log_format = logging.Formatter("%(asctime)s - %(levelname)s | %(message)s")
loghandler_file = TimedRotatingFileHandler(
log_file, when="midnight", interval=1, backupCount=10
)
loghandler_file.setFormatter(log_format)
loghandler_file.setLevel(log_level)
self._logger.addHandler(loghandler_file)
if log_stdout:
logger_console = logging.StreamHandler()
logger_console.setFormatter(log_format)
logger_console.setLevel(log_level)
self._logger.addHandler(logger_console)
def notify(self):
"""
Prepare input data and initiate notification sending
"""
self._logger.info("Notification process started")
self.api.send()
self._logger.info("Notification process finished")
Plugin().notify()

BIN
ecall/umb_ecall_notify-2.2.0.mkp Executable file
View File

Binary file not shown.

BIN
ecall/umb_ecall_notify-2.3.0.mkp Executable file
View File

Binary file not shown.

BIN
ecall/umb_ecall_notify-2.4.0.mkp Executable file
View File

Binary file not shown.