vh/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2025-12-07 09:36:05 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Notification rework v0.5

Browse Source
This commit is contained in:
Jokob-sk
2023-10-08 11:15:10 +11:00
parent d4b590a9fc
commit 79c47015f4
7 changed files with 111 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
pialert/helper.py
View File

@@ -243,18 +243,32 @@ def write_file(pPath, pText):
#-------------------------------------------------------------------------------
# Return whole setting touple
def get_setting(key):
result = None
# index order: key, name, desc, inputtype, options, regex, result, group, events
for set in conf.mySettings:
if set[0] == key:
result = set
if result is None:
mylog('minimal', [' Error - setting_missing - Setting not found for key: ', key])
mylog('minimal', [' Error - logging the settings into file: ', logPath + '/setting_missing.json'])
write_file (logPath + '/setting_missing.json', json.dumps({ 'data' : conf.mySettings}))
return result
settingsFile = apiPath + '/table_settings.json'
try:
with open(settingsFile, 'r') as json_file:
data = json.load(json_file)
# if not isinstance(data, list):
# mylog('minimal', [f' [Settings] Data is not a list of dictionaries (file: {settingsFile})'])
for item in data.get("data",[]):
if item.get("Code_Name") == key:
return item
mylog('minimal', [f'[Settings] Error - setting_missing - Setting not found for key: {key} in file {settingsFile}'])
return None
except (FileNotFoundError, json.JSONDecodeError, ValueError) as e:
# Handle the case when the file is not found, JSON decoding fails, or data is not in the expected format
mylog('minimal', [f'[Settings] Error - JSONDecodeError or FileNotFoundError for file {settingsFile}'])
return None
#-------------------------------------------------------------------------------
# Return setting value
@@ -264,8 +278,8 @@ def get_setting_value(key):
if get_setting(key) is not None:
setVal = set[6] # setting value
setTyp = set[3] # setting type
setVal = set["Value"] # setting value
setTyp = set["Type"] # setting type
return setVal

64
pialert/plugin.py
View File

@@ -13,7 +13,7 @@ from const import pluginsPath, logPath
from logger import mylog
from helper import timeNowTZ, updateState, get_file_content, write_file, get_setting, get_setting_value
from api import update_api
from plugin_utils import logEventStatusCounts, get_plugin_string, get_plugin_setting, print_plugin_info, flatten_array, combine_plugin_objects, resolve_wildcards_arr, get_plugin_setting_value, handle_empty, custom_plugin_decoder
from plugin_utils import logEventStatusCounts, get_plugin_string, get_plugin_setting, print_plugin_info, list_to_csv, combine_plugin_objects, resolve_wildcards_arr, get_plugin_setting_value, handle_empty, custom_plugin_decoder
#-------------------------------------------------------------------------------
@@ -27,8 +27,8 @@ class plugin_param:
inputValue = get_setting(param["value"])
if inputValue != None:
setVal = inputValue[6] # setting value
setTyp = inputValue[3] # setting type
setVal = inputValue["Value"] # setting value
setTyp = inputValue["Type"] # setting type
noConversion = ['text', 'string', 'integer', 'boolean', 'password', 'readonly', 'integer.select', 'text.select', 'integer.checkbox' ]
arrayConversion = ['text.multiselect', 'list', 'subnets']
@@ -45,11 +45,8 @@ class plugin_param:
elif setTyp in arrayConversion:
# make them safely passable to a python or linux script
resolved = flatten_array(setVal)
resolved = list_to_csv(setVal)
elif setTyp in arrayConversionBase64:
# make them safely passable to a python or linux script by converting them to a base64 string if necessary (if the arg contains spaces)
resolved = flatten_array(setVal)
else:
for item in jsonConversion:
if setTyp.endswith(item):
@@ -66,7 +63,7 @@ class plugin_param:
paramValuesCount = len(inputValue)
# make them safely passable to a python or linux script
resolved = flatten_array(inputValue)
resolved = list_to_csv(inputValue)
mylog('debug', f'[Plugins] Resolved value: {resolved}')
@@ -750,7 +747,7 @@ def check_and_run_user_event(db, pluginsState):
return pluginsState
if event == 'test':
handle_test(param)
pluginsState = handle_test(param, db, pluginsState)
if event == 'run':
pluginsState = handle_run(param, db, pluginsState)
@@ -778,34 +775,41 @@ def handle_run(runType, db, pluginsState):
#-------------------------------------------------------------------------------
def handle_test(testType):
def handle_test(runType, db, pluginsState):
mylog('minimal', ['[', timeNowTZ(), '] START Test: ', testType])
# TODO finish
# Open text sample
sample_txt = get_file_content(pialertPath + '/back/report_sample.txt')
# # Open text sample
# sample_txt = get_file_content(pialertPath + '/back/report_sample.txt')
# Open html sample
sample_html = get_file_content(pialertPath + '/back/report_sample.html')
# # Open html sample
# sample_html = get_file_content(pialertPath + '/back/report_sample.html')
# Open json sample and get only the payload part
sample_json_payload = json.loads(get_file_content(pialertPath + '/back/webhook_json_sample.json'))[0]["body"]["attachments"][0]["text"]
# # Open json sample and get only the payload part
# sample_json_payload = json.loads(get_file_content(pialertPath + '/back/webhook_json_sample.json'))[0]["body"]["attachments"][0]["text"]
sample_msg = noti_obj(sample_json_payload, sample_txt, sample_html, "test_sample")
# sample_msg = noti_obj(sample_json_payload, sample_txt, sample_html, "test_sample")
pluginsState = handle_run(param, db, pluginsState)
if testType == 'Email':
send_email(sample_msg)
elif testType == 'Webhooks':
send_webhook (sample_msg)
elif testType == 'Apprise':
send_apprise (sample_msg)
elif testType == 'NTFY':
send_ntfy (sample_msg)
elif testType == 'PUSHSAFER':
send_pushsafer (sample_msg)
else:
mylog('none', ['[Test Publishers] No test matches: ', testType])
# if testType == 'Email':
# send_email(sample_msg)
# elif testType == 'Webhooks':
# send_webhook (sample_msg)
# elif testType == 'Apprise':
# send_apprise (sample_msg)
# elif testType == 'NTFY':
# send_ntfy (sample_msg)
# elif testType == 'PUSHSAFER':
# send_pushsafer (sample_msg)
# else:
# mylog('none', ['[Test Publishers] No test matches: ', testType])
mylog('minimal', ['[Test Publishers] END Test: ', testType])
# mylog('minimal', ['[Test Publishers] END Test: ', testType])
return pluginsState

40
pialert/plugin_utils.py
View File

@@ -71,30 +71,42 @@ def get_plugin_string(props, el):
#-------------------------------------------------------------------------------
def flatten_array(arr):
# generates a comma separated list of values from a list (or a string representing a list)
def list_to_csv(arr):
tmp = ''
arrayItemStr = ''
mylog('debug', '[Plugins] Flattening the below array')
mylog('debug', '[Plugins] Flattening the below array')
mylog('debug', arr)
for arrayItem in arr:
# only one column flattening is supported
if isinstance(arrayItem, list):
arrayItemStr = str(arrayItem[0]).replace("'", '') # removing single quotes - not allowed
else:
# is string already
arrayItemStr = arrayItem
mylog('debug', f'[Plugins] isinstance(arr, list) : {isinstance(arr, list)}')
mylog('debug', f'[Plugins] isinstance(arr, str) : {isinstance(arr, str)}')
if isinstance(arr, str):
return arr.replace('[','').replace(']','').replace("'", '') # removing brackets and single quotes (not allowed)
elif isinstance(arr, list):
for arrayItem in arr:
# only one column flattening is supported
if isinstance(arrayItem, list):
arrayItemStr = str(arrayItem[0]).replace("'", '') # removing single quotes - not allowed
else:
# is string already
arrayItemStr = arrayItem
tmp += f'{arrayItemStr},'
tmp += f'{arrayItemStr},'
tmp = tmp[:-1] # Remove last comma ','
tmp = tmp[:-1] # Remove last comma ','
mylog('debug', f'[Plugins] Flattened array: {tmp}')
mylog('debug', f'[Plugins] Flattened array: {tmp}')
return tmp
else:
mylog('none', f'[Plugins] ERROR Could not convert array: {arr}')
return tmp

28
pialert/reporting.py
View File

@@ -317,26 +317,26 @@ def get_notifications (db):
#-------------------------------------------------------------------------------
def check_config(service):
# #-------------------------------------------------------------------------------
# def check_config(service):
if service == 'email':
return email_check_config()
# if service == 'email':
# return email_check_config()
if service == 'apprise':
return apprise_check_config()
# if service == 'apprise':
# return apprise_check_config()
if service == 'webhook':
return webhook_check_config()
# if service == 'webhook':
# return webhook_check_config()
if service == 'ntfy':
return ntfy_check_config ()
# if service == 'ntfy':
# return ntfy_check_config ()
if service == 'pushsafer':
return pushsafer_check_config()
# if service == 'pushsafer':
# return pushsafer_check_config()
if service == 'mqtt':
return mqtt_check_config()
# if service == 'mqtt':
# return mqtt_check_config()
#-------------------------------------------------------------------------------
# Replacing table headers