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 3 Packages Projects Releases Wiki Activity

Loading plugins v 0.35 🔌

Browse Source
This commit is contained in:
jokob-sk
2024-05-18 11:16:28 +10:00
parent 315921873d
commit 6f4ae1d29f
5 changed files with 86 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
front/settings.php
View File

@@ -282,10 +282,8 @@ while ($row = $result -> fetchArray (SQLITE3_ASSOC)) {
if(getSetting(group+"_RUN") != "")
{
// 🔺 update also in initialise.py if you update below list/array
let isEnabled = ["once", "schedule", "always_after_scan", "on_new_device", "on_notification", "before_config_save", "before_name_updates" ].includes(getSetting(group+"_RUN"));
isEnabled ? onOff = 'dot-circle' : onOff = 'circle';
// show all enabled plugins
getSetting(group+"_RUN") != 'disabled' ? onOff = 'dot-circle' : onOff = 'circle';
enabledHtml = `
<div class="enabled-disabled-icon">

75
server/helper.py
View File

@@ -283,14 +283,16 @@ def get_setting(key):
#-------------------------------------------------------------------------------
# Settings
#-------------------------------------------------------------------------------
#-------------------------------------------------------------------------------
# Return setting value
def get_setting_value(key):
# Returns empty string if not set
value = ''
setting = get_setting(key)
value = ''
if setting is not None:
# mylog('none', [f'[SETTINGS] setting json:{json.dumps(setting)}'])
@@ -299,41 +301,50 @@ def get_setting_value(key):
set_value = 'Error: Not handled'
set_value = setting["Value"] # Setting value
set_type = setting["Type"] # Setting type
set_type = setting["Type"] # Setting type
# Handle different types of settings
if set_type in ['text', 'string', 'password', 'password.SHA256', 'readonly', 'text.select']:
value = str(set_value)
elif set_type in ['boolean', 'integer.checkbox']:
value = False
value = setting_value_to_python_type(set_type, set_value)
if isinstance(set_value, str) and set_value.lower() in ['true', '1']:
value = True
elif isinstance(set_value, int) and set_value == 1:
value = True
elif isinstance(set_value, bool):
value = set_value
elif set_type in ['integer.select', 'integer']:
value = int(set_value)
elif set_type in ['text.multiselect', 'list', 'subnets']:
# Handle string
return value
mylog('debug', [f'[SETTINGS] Handling set_type: "{set_type}", set_value: "{set_value}"'])
#-------------------------------------------------------------------------------
# Convert the setting value to the corresponding python type
def setting_value_to_python_type(set_type, set_value):
if isinstance(set_value, str):
value = json.loads(set_value.replace("'", "\""))
# Assuming set_value is a list in this case
elif isinstance(set_value, list):
value = set_value
elif set_type == '.template':
# Assuming set_value is a JSON object in this case
value = json.loads(set_value)
else:
mylog('none', [f'[SETTINGS] ⚠ ERROR - set_type not handled:{set_type}'])
mylog('none', [f'[SETTINGS] ⚠ ERROR - setting json:{json.dumps(setting)}'])
value = ''
# Handle different types of settings
if set_type in ['text', 'string', 'password', 'password.SHA256', 'readonly', 'text.select']:
value = str(set_value)
elif set_type in ['boolean', 'integer.checkbox']:
value = False
if isinstance(set_value, str) and set_value.lower() in ['true', '1']:
value = True
elif isinstance(set_value, int) and set_value == 1:
value = True
elif isinstance(set_value, bool):
value = set_value
elif set_type in ['integer.select', 'integer']:
value = int(set_value)
elif set_type in ['text.multiselect', 'list', 'subnets']:
# Handle string
mylog('debug', [f'[SETTINGS] Handling set_type: "{set_type}", set_value: "{set_value}"'])
if isinstance(set_value, str):
value = json.loads(set_value.replace("'", "\""))
# Assuming set_value is a list in this case
elif isinstance(set_value, list):
value = set_value
elif set_type == '.template':
# Assuming set_value is a JSON object in this case
value = json.loads(set_value)
else:
mylog('none', [f'[SETTINGS] ⚠ ERROR - set_type not handled:{set_type}'])
mylog('none', [f'[SETTINGS] ⚠ ERROR - setting json:{json.dumps(setting)}'])
return value

34
server/initialise.py
View File

@@ -12,12 +12,12 @@ import re
import conf
from const import fullConfPath
from helper import collect_lang_strings, updateSubnets, initOrSetParam, isJsonObject, updateState
from helper import collect_lang_strings, updateSubnets, initOrSetParam, isJsonObject, updateState, setting_value_to_python_type
from logger import mylog
from api import update_api
from scheduler import schedule_class
from plugin import print_plugin_info, run_plugin_scripts
from plugin_utils import get_plugins_configs, get_plugin_setting
from plugin_utils import get_plugins_configs, get_plugin_setting_obj
#===============================================================================
# Initialise user defined values
@@ -164,30 +164,32 @@ def importConfigs (db, all_plugins):
print_plugin_info(plugin, ['display_name','description'])
# The below few lines are used to determine if tghe plugin should be loaded, or skipped
# The below lines are used to determine if the plugin should be loaded, or skipped based on user settings (conf.LOADED_PLUGINS)
# ...or based on if is already enabled, or if the default configuration loads the plugin (RUN function != disabled )
# get default plugin run value
plugin_run = get_plugin_setting(plugin, "RUN")
plugin_run = ''
setting_obj = get_plugin_setting_obj(plugin, "RUN")
if setting_obj is not None:
set_type = setting.get('type')
set_value = setting.get('default_value')
plugin_run = setting_value_to_python_type(set_type, set_value)
if plugin_run is None:
mylog('none', ['[Config] DEBUG1 plugin_run: None'])
mylog('none', ['[Config] plugin_run (default_value): None'])
else:
mylog('none', ['[Config] DEBUG2 plugin_run: ', plugin_run])
mylog('none', ['[Config] plugin_run (default_value): ', plugin_run])
# get user-defined run value if available
if pref + "_RUN" in c_d:
plugin_run = c_d[pref + "_RUN" ]
if plugin_run is None:
mylog('none', ['[Config] DEBUG3 plugin_run: None'])
else:
mylog('none', ['[Config] DEBUG4 plugin_run: ', plugin_run])
mylog('none', ['[Config] plugin_run (user defined): ', plugin_run])
# only include loaded plugins, and the ones that are enabled
# 🔺 update also in settings.php if you update below list/array
enabled_run_values = ["once", "schedule", "always_after_scan", "on_new_device", "on_notification", "before_config_save", "before_name_updates" ]
if pref in conf.LOADED_PLUGINS or plugin_run in enabled_run_values or plugin_run is None:
# only include loaded plugins, and the ones that are enabled
if pref in conf.LOADED_PLUGINS or plugin_run != 'disabled' or setting_obj is None or plugin_run is None:
stringSqlParams = []

14
server/plugin.py
View File

@@ -13,7 +13,7 @@ from const import pluginsPath, logPath, applicationPath, reportTemplatesPath
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, list_to_csv, combine_plugin_objects, resolve_wildcards_arr, handle_empty, custom_plugin_decoder
from plugin_utils import logEventStatusCounts, get_plugin_string, get_plugin_setting_obj, print_plugin_info, list_to_csv, combine_plugin_objects, resolve_wildcards_arr, handle_empty, custom_plugin_decoder
from notification import Notification_obj
@@ -113,7 +113,7 @@ def run_plugin_scripts(db, all_plugins, runType, pluginsState = plugins_state())
shouldRun = False
prefix = plugin["unique_prefix"]
set = get_plugin_setting(plugin, "RUN")
set = get_plugin_setting_obj(plugin, "RUN")
if set != None and set['value'] == runType:
if runType != "schedule":
shouldRun = True
@@ -130,7 +130,7 @@ def run_plugin_scripts(db, all_plugins, runType, pluginsState = plugins_state())
updateState(f"Plugins: {prefix}")
print_plugin_info(plugin, ['display_name'])
mylog('debug', ['[Plugins] CMD: ', get_plugin_setting(plugin, "CMD")["value"]])
mylog('debug', ['[Plugins] CMD: ', get_plugin_setting_obj(plugin, "CMD")["value"]])
pluginsState = execute_plugin(db, all_plugins, plugin, pluginsState)
# update last run time
if runType == "schedule":
@@ -155,7 +155,7 @@ def execute_plugin(db, all_plugins, plugin, pluginsState = plugins_state() ):
pluginsState = plugins_state()
# ------- necessary settings check --------
set = get_plugin_setting(plugin, "CMD")
set = get_plugin_setting_obj(plugin, "CMD")
# handle missing "function":"CMD" setting
if set == None:
@@ -163,7 +163,7 @@ def execute_plugin(db, all_plugins, plugin, pluginsState = plugins_state() ):
set_CMD = set["value"]
set = get_plugin_setting(plugin, "RUN_TIMEOUT")
set = get_plugin_setting_obj(plugin, "RUN_TIMEOUT")
# handle missing "function":"<unique_prefix>_TIMEOUT" setting
if set == None:
@@ -313,7 +313,7 @@ def execute_plugin(db, all_plugins, plugin, pluginsState = plugins_state() ):
mylog('verbose', ['[Plugins] Executing: ', q])
# ------- necessary settings check --------
set = get_plugin_setting(plugin, "DB_PATH")
set = get_plugin_setting_obj(plugin, "DB_PATH")
# handle missing "function":"DB_PATH" setting
if set == None:
@@ -706,7 +706,7 @@ class plugin_object_class:
self.watchedClmns = []
self.watchedIndxs = []
setObj = get_plugin_setting(plugin, 'WATCH')
setObj = get_plugin_setting_obj(plugin, 'WATCH')
indexNameColumnMapping = [(6, 'Watched_Value1' ), (7, 'Watched_Value2' ), (8, 'Watched_Value3' ), (9, 'Watched_Value4' )]

17
server/plugin_utils.py
View File

@@ -35,7 +35,22 @@ def print_plugin_info(plugin, elements = ['display_name']):
#-------------------------------------------------------------------------------
# Gets the whole setting object
def get_plugin_setting(plugin, function_key):
def get_plugin_setting_obj(plugin, function_key):
result = None
for set in plugin['settings']:
if set["function"] == function_key:
result = set
# if result == None:
# mylog('debug', [f'[{module_name}] Setting with "function":"', function_key, '" is missing in plugin: ', get_plugin_string(plugin, 'display_name')])
return result
#-------------------------------------------------------------------------------
# Gets the setting value for a plugin from the default JSON
def get_plugin_setting_value(plugin, function_key):
result = None