DGNum/uptime-kuma-api
6
0
Fork 0
Code Issues 17 Pull requests 5 Projects Releases 24 Packages Wiki Activity Actions

improve notification type conversion

Browse source
This commit is contained in:
lucasheld 2022-08-02 21:32:28 +02:00
parent d29d908b61
commit b92263dd47
9 changed files with 365 additions and 172 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
scripts/build_notification_info_return.py Normal file
View file

@ -0,0 +1,14 @@
from uptime_kuma_api import params_map_notification_providers, notification_provider_options, params_map_notification_provider_options, convert_from_socket
for provider_sock, provider_py in params_map_notification_providers.items():
options = notification_provider_options[provider_sock]
tmp = options
params_map = params_map_notification_provider_options[provider_py]
options = convert_from_socket(params_map, options)
for option in options:
print(f'{option}:')
print(f' description: {provider_py} provider option.')
print(f' returned: if type is {provider_py}')
print(' type: str')

4
scripts/build_notifications.py
View file

@ -5,7 +5,7 @@ from pprint import pprint
import jinja2
from bs4 import BeautifulSoup
from uptime_kuma_api import convert_from_socket, params_map_notification_provider
from uptime_kuma_api import convert_from_socket, params_map_notification_provider_options
def deduplicate_list(l):
@ -61,7 +61,7 @@ def build_notification_provider_conditions():
param_name = re.match(r'\$parent.notification.(.*)$', v_model).group(1)
if condition:
conditions[param_name] = condition
conditions = convert_from_socket(params_map_notification_provider, conditions)
conditions = convert_from_socket(params_map_notification_provider_options, conditions)
return conditions

28
scripts/build_params_map_notification_provider.py
View file

@ -1,28 +0,0 @@
import re
import os
from uptime_kuma_api import notification_provider_options
def build_notification_provider_map():
params_map_notification_providers = {}
for notification_provider in notification_provider_options:
options = notification_provider_options[notification_provider]
provider_name = notification_provider.__dict__["_value_"].lower().replace(".", "")
for option in options:
option_orig = option
prefix = os.path.commonprefix([o.lower() for o in options] + [provider_name])
option = option[len(prefix):]
option = re.sub('([A-Z]+)', r'_\1', option).lower()
option = provider_name + "_" + option
option = option.replace("__", "_")
params_map_notification_providers[option_orig] = option
return params_map_notification_providers
notification_provider_map = build_notification_provider_map()
print("params_map_notification_provider =", notification_provider_map)

10
scripts/build_params_map_notification_provider_options.j2 Normal file
View file

@ -0,0 +1,10 @@
params_map_notification_provider_options = {
{%- for provider in notification_provider_map %}
{%- set options = notification_provider_map[provider] %}
'{{ provider }}': {
{%- for key, value in options.items() %}
'{{ key }}': '{{ value }}',
{%- endfor %}
},
{%- endfor %}
}

40
scripts/build_params_map_notification_provider_options.py Normal file
View file

@ -0,0 +1,40 @@
import re
import os
import jinja2
from uptime_kuma_api import params_map_notification_providers, notification_provider_options
def build_notification_provider_map():
params_map_notification_provider_options = {}
for provider_sock, provider_py in params_map_notification_providers.items():
options_sock = notification_provider_options[provider_sock]
params_map_notification_provider_options[provider_py] = {}
for option in options_sock:
option_orig = option
# for example for rocket_chat
prefix = os.path.commonprefix([o.lower() for o in options_sock] + [provider_py])
option = option[len(prefix):]
option = re.sub('([A-Z]+)', r'_\1', option).lower()
# for example for smtp
if option.startswith(provider_py):
option = option[len(provider_py):]
option = provider_py + "_" + option
option = option.replace("__", "_")
params_map_notification_provider_options[provider_py][option_orig] = option
return params_map_notification_provider_options
notification_provider_map = build_notification_provider_map()
env = jinja2.Environment(loader=jinja2.FileSystemLoader("./"))
template = env.get_template("build_params_map_notification_provider_options.j2")
rendered = template.render(notification_provider_map=notification_provider_map)
print(rendered)

12
scripts/build_params_map_notification_providers.py Normal file
View file

@ -0,0 +1,12 @@
import re
from uptime_kuma_api import notification_provider_options
params_map_notification_providers = {}
for notification_provider in notification_provider_options:
provider_name_orig = notification_provider.__dict__["_value_"]
provider_name = re.sub('([A-Z]+)', r'_\1', provider_name_orig).lower().replace(".", "_").strip("_")
params_map_notification_providers[provider_name_orig] = provider_name
print(params_map_notification_providers)

13
uptime_kuma_api/__init__.py
View file

@ -4,8 +4,17 @@ from .monitor_type import MonitorType
from .notification_providers import NotificationType, notification_provider_options
from .proxy_protocol import ProxyProtocol
from .incident_style import IncidentStyle
from .converter import convert_from_socket, convert_to_socket, params_map_monitor, params_map_notification, \
params_map_notification_provider, params_map_proxy, params_map_status_page, params_map_info, \
from .converter import \
convert_from_socket,\
convert_to_socket, \
params_map_monitor, \
params_map_notification,\
params_map_notification_providers,\
params_map_notification_provider_options,\
get_params_map_notification, \
params_map_proxy, \
params_map_status_page, \
params_map_info, \
params_map_settings
from .exceptions import UptimeKumaException
from .api import UptimeKumaApi

47
uptime_kuma_api/api.py
View file

@ -8,14 +8,19 @@ from . import MonitorType
from . import NotificationType, notification_provider_options
from . import ProxyProtocol
from . import IncidentStyle
from . import convert_from_socket, convert_to_socket, params_map_monitor, params_map_notification, \
params_map_notification_provider, params_map_proxy, params_map_status_page, params_map_info, \
from . import \
convert_from_socket,\
convert_to_socket, \
params_map_monitor, \
params_map_notification,\
params_map_notification_providers, \
get_params_map_notification, \
params_map_proxy, \
params_map_status_page, \
params_map_info, \
params_map_settings
from . import UptimeKumaException
params_map_notification_and_provider = {**params_map_notification, **params_map_notification_provider}
def int_to_bool(data, keys):
if type(data) == list:
for d in data:
@ -180,6 +185,8 @@ def _build_notification_data(
apply_existing: bool = False,
**kwargs
):
params_map = get_params_map_notification(type_)
type_ = convert_to_socket(params_map, type_)
data = {
"name": name,
"type_": type_,
@ -187,7 +194,7 @@ def _build_notification_data(
"apply_existing": apply_existing,
**kwargs
}
data = convert_to_socket(params_map_notification_and_provider, data)
data = convert_to_socket(params_map, data)
return data
@ -336,12 +343,13 @@ def _check_arguments_monitor(kwargs):
def _check_arguments_notification(kwargs):
required_args = ["type_", "name"]
_check_missing_arguments(required_args, kwargs, params_map_notification_and_provider)
_check_missing_arguments(required_args, kwargs, params_map_notification)
type_ = kwargs[convert_to_socket(params_map_notification, "type")]
required_args_sock = notification_provider_options[type_]
required_args = convert_from_socket(params_map_notification_provider, required_args_sock)
_check_missing_arguments(required_args, kwargs, params_map_notification_and_provider)
params_map = get_params_map_notification(type_sock=type_)
required_args = convert_from_socket(params_map, required_args_sock)
_check_missing_arguments(required_args, kwargs, params_map)
provider_conditions = {
'gotify_priority': {
@ -357,7 +365,7 @@ def _check_arguments_notification(kwargs):
'min': 0
}
}
_check_argument_conditions(provider_conditions, kwargs, params_map_notification_and_provider)
_check_argument_conditions(provider_conditions, kwargs, params_map)
def _check_arguments_proxy(kwargs):
@ -379,7 +387,7 @@ class UptimeKumaApi(object):
def __init__(self, url):
self.sio = socketio.Client()
self._event_data = {
self._event_data: dict = {
"monitorList": None,
"notificationList": None,
"proxyList": None,
@ -563,8 +571,11 @@ class UptimeKumaApi(object):
config = json.loads(notification["config"])
del notification["config"]
notification.update(config)
notification["type"] = convert_from_socket(params_map_notification_providers, notification["type"])
params_map = get_params_map_notification(notification["type"])
notification = convert_from_socket(params_map, notification)
r.append(notification)
r = convert_from_socket(params_map_notification_and_provider, r)
return r
def get_notification(self, id_: int):
@ -589,17 +600,23 @@ class UptimeKumaApi(object):
def edit_notification(self, id_: int, **kwargs):
notification = self.get_notification(id_)
if "type_" in kwargs and kwargs["type_"] != notification["type_"]:
# remove old notification provider options from notification object
if "type_" in kwargs and kwargs != notification["type_"]:
for provider in notification_provider_options:
provider_options = notification_provider_options[provider]
if provider != kwargs:
params_map = get_params_map_notification(type_sock=provider)
provider_options = convert_from_socket(params_map, provider_options)
if provider != kwargs["type_"]:
for option in provider_options:
if option in notification:
del notification[option]
# convert type from py to sock
kwargs["type_"] = convert_to_socket(params_map_notification_providers, kwargs["type_"])
notification.update(kwargs)
notification = convert_to_socket(params_map_notification_and_provider, notification)
params_map = get_params_map_notification(type_sock=kwargs["type_"])
notification = convert_to_socket(params_map, notification)
_check_arguments_notification(notification)
return self._call('addNotification', (notification, id_))

203
uptime_kuma_api/converter.py
View file

@ -38,31 +38,88 @@ params_map_notification = {
"applyExisting": "apply_existing",
}
params_map_notification_provider = {
params_map_notification_providers = {
'alerta': 'alerta',
'AliyunSMS': 'aliyun_sms',
'apprise': 'apprise',
'Bark': 'bark',
'clicksendsms': 'clicksendsms',
'DingDing': 'ding_ding',
'discord': 'discord',
'Feishu': 'feishu',
'GoogleChat': 'google_chat',
'gorush': 'gorush',
'gotify': 'gotify',
'line': 'line',
'lunasea': 'lunasea',
'matrix': 'matrix',
'mattermost': 'mattermost',
'ntfy': 'ntfy',
'octopush': 'octopush',
'OneBot': 'one_bot',
'PagerDuty': 'pager_duty',
'promosms': 'promosms',
'pushbullet': 'pushbullet',
'PushDeer': 'push_deer',
'pushover': 'pushover',
'pushy': 'pushy',
'rocket.chat': 'rocket_chat',
'serwersms': 'serwersms',
'signal': 'signal',
'slack': 'slack',
'smtp': 'smtp',
'stackfield': 'stackfield',
'teams': 'teams',
'PushByTechulus': 'push_by_techulus',
'telegram': 'telegram',
'webhook': 'webhook',
'WeCom': 'we_com'
}
params_map_notification_provider_options = {
'alerta': {
'alertaApiEndpoint': 'alerta_api_endpoint',
'alertaApiKey': 'alerta_api_key',
'alertaEnvironment': 'alerta_environment',
'alertaAlertState': 'alerta_alert_state',
'alertaRecoverState': 'alerta_recover_state',
'phonenumber': 'aliyunsms_phonenumber',
'templateCode': 'aliyunsms_template_code',
'signName': 'aliyunsms_sign_name',
'accessKeyId': 'aliyunsms_access_key_id',
'secretAccessKey': 'aliyunsms_secret_access_key',
'appriseURL': 'apprise_apprise_url',
},
'aliyun_sms': {
'phonenumber': 'aliyun_sms_phonenumber',
'templateCode': 'aliyun_sms_template_code',
'signName': 'aliyun_sms_sign_name',
'accessKeyId': 'aliyun_sms_access_key_id',
'secretAccessKey': 'aliyun_sms_secret_access_key',
},
'apprise': {
'appriseURL': 'apprise_url',
'title': 'apprise_title',
},
'bark': {
'barkEndpoint': 'bark_endpoint',
},
'clicksendsms': {
'clicksendsmsLogin': 'clicksendsms_login',
'clicksendsmsPassword': 'clicksendsms_password',
'clicksendsmsToNumber': 'clicksendsms_to_number',
'clicksendsmsSenderName': 'clicksendsms_sender_name',
'webHookUrl': 'dingding_web_hook_url',
'secretKey': 'dingding_secret_key',
},
'ding_ding': {
'webHookUrl': 'ding_ding_web_hook_url',
'secretKey': 'ding_ding_secret_key',
},
'discord': {
'discordUsername': 'discord_username',
'discordWebhookUrl': 'discord_webhook_url',
'discordPrefixMessage': 'discord_prefix_message',
},
'feishu': {
'feishuWebHookUrl': 'feishu_web_hook_url',
'googleChatWebhookURL': 'googlechat_webhook_url',
},
'google_chat': {
'googleChatWebhookURL': 'google_chat_chat_webhook_url',
},
'gorush': {
'gorushDeviceToken': 'gorush_device_token',
'gorushPlatform': 'gorush_platform',
'gorushTitle': 'gorush_title',
@ -70,23 +127,37 @@ params_map_notification_provider = {
'gorushRetry': 'gorush_retry',
'gorushTopic': 'gorush_topic',
'gorushServerURL': 'gorush_server_url',
},
'gotify': {
'gotifyserverurl': 'gotify_serverurl',
'gotifyapplicationToken': 'gotify_application_token',
'gotifyPriority': 'gotify_priority',
},
'line': {
'lineChannelAccessToken': 'line_channel_access_token',
'lineUserID': 'line_user_id',
},
'lunasea': {
'lunaseaDevice': 'lunasea_device',
},
'matrix': {
'internalRoomId': 'matrix_internal_room_id',
'accessToken': 'onebot_access_token',
'accessToken': 'matrix_access_token',
'homeserverUrl': 'matrix_homeserver_url',
},
'mattermost': {
'mattermostusername': 'mattermost_username',
'mattermostWebhookUrl': 'mattermost_webhook_url',
'mattermostchannel': 'mattermost_channel',
'mattermosticonemo': 'mattermost_iconemo',
'mattermosticonurl': 'mattermost_iconurl',
},
'ntfy': {
'ntfyserverurl': 'ntfy_serverurl',
'ntfytopic': 'ntfy_topic',
'ntfyPriority': 'ntfy_priority',
},
'octopush': {
'octopushVersion': 'octopush_version',
'octopushAPIKey': 'octopush_apikey',
'octopushLogin': 'octopush_login',
@ -98,70 +169,108 @@ params_map_notification_provider = {
'octopushDMPhoneNumber': 'octopush_dmphone_number',
'octopushDMSenderName': 'octopush_dmsender_name',
'octopushDMSMSType': 'octopush_dmsmstype',
'httpAddr': 'onebot_http_addr',
'msgType': 'onebot_msg_type',
'recieverId': 'onebot_reciever_id',
'pagerdutyAutoResolve': 'pagerduty_auto_resolve',
'pagerdutyIntegrationUrl': 'pagerduty_integration_url',
'pagerdutyPriority': 'pagerduty_priority',
'pagerdutyIntegrationKey': 'pagerduty_integration_key',
},
'one_bot': {
'httpAddr': 'one_bot_http_addr',
'accessToken': 'one_bot_access_token',
'msgType': 'one_bot_msg_type',
'recieverId': 'one_bot_reciever_id',
},
'pager_duty': {
'pagerdutyAutoResolve': 'pager_duty_duty_auto_resolve',
'pagerdutyIntegrationUrl': 'pager_duty_duty_integration_url',
'pagerdutyPriority': 'pager_duty_duty_priority',
'pagerdutyIntegrationKey': 'pager_duty_duty_integration_key',
},
'promosms': {
'promosmsLogin': 'promosms_login',
'promosmsPassword': 'promosms_password',
'promosmsPhoneNumber': 'promosms_phone_number',
'promosmsSMSType': 'promosms_smstype',
'promosmsSenderName': 'promosms_sender_name',
},
'pushbullet': {
'pushbulletAccessToken': 'pushbullet_access_token',
'pushdeerKey': 'pushdeer_key',
},
'push_deer': {
'pushdeerKey': 'push_deer_deer_key',
},
'pushover': {
'pushoveruserkey': 'pushover_userkey',
'pushoverapptoken': 'pushover_apptoken',
'pushoversounds': 'pushover_sounds',
'pushoverpriority': 'pushover_priority',
'pushovertitle': 'pushover_title',
'pushoverdevice': 'pushover_device',
},
'pushy': {
'pushyAPIKey': 'pushy_apikey',
'pushyToken': 'pushy_token',
'rocketchannel': 'rocketchat_channel',
'rocketusername': 'rocketchat_username',
'rocketiconemo': 'rocketchat_iconemo',
'rocketwebhookURL': 'rocketchat_webhook_url',
'rocketbutton': 'rocketchat_button',
},
'rocket_chat': {
'rocketchannel': 'rocket_chat_channel',
'rocketusername': 'rocket_chat_username',
'rocketiconemo': 'rocket_chat_iconemo',
'rocketwebhookURL': 'rocket_chat_webhook_url',
'rocketbutton': 'rocket_chat_button',
},
'serwersms': {
'serwersmsUsername': 'serwersms_username',
'serwersmsPassword': 'serwersms_password',
'serwersmsPhoneNumber': 'serwersms_phone_number',
'serwersmsSenderName': 'serwersms_sender_name',
},
'signal': {
'signalNumber': 'signal_number',
'signalRecipients': 'signal_recipients',
'signalURL': 'signal_url',
},
'slack': {
'slackbutton': 'slack_button',
'slackchannel': 'slack_channel',
'slackusername': 'slack_username',
'slackiconemo': 'slack_iconemo',
'slackwebhookURL': 'slack_webhook_url',
'smtpHost': 'smtp_smtp_host',
'smtpPort': 'smtp_smtp_port',
'smtpSecure': 'smtp_smtp_secure',
'smtpIgnoreTLSError': 'smtp_smtp_ignore_tlserror',
'smtpDkimDomain': 'smtp_smtp_dkim_domain',
'smtpDkimKeySelector': 'smtp_smtp_dkim_key_selector',
'smtpDkimPrivateKey': 'smtp_smtp_dkim_private_key',
'smtpDkimHashAlgo': 'smtp_smtp_dkim_hash_algo',
'smtpDkimheaderFieldNames': 'smtp_smtp_dkimheader_field_names',
'smtpDkimskipFields': 'smtp_smtp_dkimskip_fields',
'smtpUsername': 'smtp_smtp_username',
'smtpPassword': 'smtp_smtp_password',
},
'smtp': {
'smtpHost': 'smtp_host',
'smtpPort': 'smtp_port',
'smtpSecure': 'smtp_secure',
'smtpIgnoreTLSError': 'smtp_ignore_tlserror',
'smtpDkimDomain': 'smtp_dkim_domain',
'smtpDkimKeySelector': 'smtp_dkim_key_selector',
'smtpDkimPrivateKey': 'smtp_dkim_private_key',
'smtpDkimHashAlgo': 'smtp_dkim_hash_algo',
'smtpDkimheaderFieldNames': 'smtp_dkimheader_field_names',
'smtpDkimskipFields': 'smtp_dkimskip_fields',
'smtpUsername': 'smtp_username',
'smtpPassword': 'smtp_password',
'customSubject': 'smtp_custom_subject',
'smtpFrom': 'smtp_smtp_from',
'smtpCC': 'smtp_smtp_cc',
'smtpBCC': 'smtp_smtp_bcc',
'smtpTo': 'smtp_smtp_to',
'smtpFrom': 'smtp_from',
'smtpCC': 'smtp_cc',
'smtpBCC': 'smtp_bcc',
'smtpTo': 'smtp_to',
},
'stackfield': {
'stackfieldwebhookURL': 'stackfield_webhook_url',
},
'teams': {
'webhookUrl': 'teams_webhook_url',
'pushAPIKey': 'pushbytechulus_apikey',
},
'push_by_techulus': {
'pushAPIKey': 'push_by_techulus_apikey',
},
'telegram': {
'telegramBotToken': 'telegram_bot_token',
'telegramChatID': 'telegram_chat_id',
},
'webhook': {
'webhookContentType': 'webhook_content_type',
'webhookURL': 'webhook_url',
'weComBotKey': 'wecom_bot_key'
},
'we_com': {
'weComBotKey': 'we_com_com_bot_key',
},
}
params_map_proxy = {
@ -234,3 +343,13 @@ def convert_from_socket(params_map, params):
def convert_to_socket(params_map, params):
return _convert_to_from_socket(params_map, params, to_socket=True)
def get_params_map_notification(type_py=None, type_sock=None):
if not type_py:
type_py = convert_from_socket(params_map_notification_providers, type_sock)
return {
**params_map_notification,
**params_map_notification_providers,
**params_map_notification_provider_options[type_py]
}