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 - Apprise v1 - working

Browse Source
This commit is contained in:
Jokob-sk
2023-10-08 14:52:22 +11:00
parent 79c47015f4
commit be4e0acdfc
11 changed files with 152 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
pialert/__main__.py
View File

@@ -155,10 +155,10 @@ def main ():
# Write the notifications into the DB
notification = Notification_obj(db)
hasNotification = notification.create(notiStructure.json, notiStructure.text, notiStructure.html, "")
notificationObj = notification.create(notiStructure.json, notiStructure.text, notiStructure.html, "")
# run all enabled publisher gateways
if hasNotification:
if notificationObj.HasNotifications:
pluginsState = run_plugin_scripts(db, 'on_notification', pluginsState)
notification.setAllProcessed()
@@ -177,6 +177,8 @@ def main ():
# DEBUG - print number of rows updated
mylog('minimal', ['[Notification] Notifications changes: ', sql.rowcount])
else:
mylog('verbose', ['[Notification] No changes to report'])
# Commit SQL
db.commitDB()

32
pialert/helper.py
View File

@@ -244,27 +244,24 @@ def write_file(pPath, pText):
# Return whole setting touple
def get_setting(key):
settingsFile = apiPath + '/table_settings.json'
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}'])
mylog('debug', [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}'])
mylog('none', [f'[Settings] Error - JSONDecodeError or FileNotFoundError for file {settingsFile}'])
return None
@@ -274,19 +271,32 @@ def get_setting(key):
# Return setting value
def get_setting_value(key):
set = get_setting(key)
setting = get_setting(key)
if get_setting(key) is not None:
if setting is not None:
setVal = set["Value"] # setting value
setTyp = set["Type"] # setting type
set_value = setting["Value"] # Setting value
set_type = setting["Type"] # Setting type
return setVal
# Handle different types of settings
if set_type in ['text', 'string', 'password', 'readonly', 'text.select']:
return str(set_value)
elif set_type in ['boolean', 'integer.checkbox']:
return bool(set_value)
elif set_type in ['integer.select', 'integer']:
return int(set_value)
elif set_type in ['text.multiselect', 'list', 'subnets']:
# Assuming set_value is a list in this case
return set_value
elif set_type == '.template':
# Assuming set_value is a JSON object in this case
return json.loads(set_value)
return ''
#-------------------------------------------------------------------------------
# IP validation methods
#-------------------------------------------------------------------------------

25
pialert/notification.py
View File

@@ -40,11 +40,8 @@ class Notification_obj:
# Check if nothing to report, end
if JSON["internet"] == [] and JSON["new_devices"] == [] and JSON["down_devices"] == [] and JSON["events"] == [] and JSON["plugins"] == []:
self.HasNotifications = False
# end if nothing to report
return self.HasNotifications
# continue and save into DB if notifications available
self.HasNotifications = True
else:
self.HasNotifications = True
self.GUID = str(uuid.uuid4())
self.DateTimeCreated = timeNowTZ()
@@ -56,9 +53,10 @@ class Notification_obj:
self.PublishedVia = ""
self.Extra = Extra
self.upsert()
if self.HasNotifications:
self.upsert()
return self.HasNotifications
return self
# Only updates the status
def updateStatus(self, newStatus):
@@ -69,9 +67,7 @@ class Notification_obj:
def updatePublishedVia(self, newPublishedVia):
self.PublishedVia = newPublishedVia
self.DateTimePushed = timeNowTZ()
self.upsert()
# TODO Index vs hash to minimize SQL calls, finish CRUD operations, expose via API, use API in plugins
self.upsert()
# create or update a notification
def upsert(self):
@@ -82,6 +78,15 @@ class Notification_obj:
self.save()
# Remove notification object by GUID
def remove(self, GUID):
# Execute an SQL query to delete the notification with the specified GUID
self.db.sql.execute("""
DELETE FROM Notifications
WHERE GUID = ?
""", (GUID,))
self.save()
# Get all with the "new" status
def getNew(self):
self.db.sql.execute("""

55
pialert/plugin.py
View File

@@ -9,11 +9,12 @@ from collections import namedtuple
# pialert modules
import conf
from const import pluginsPath, logPath
from const import pluginsPath, logPath, pialertPath
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, 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, handle_empty, custom_plugin_decoder
from notification import Notification_obj
#-------------------------------------------------------------------------------
@@ -484,10 +485,7 @@ def process_plugin_events(db, plugin, pluginsState, plugEventsArr):
history_to_insert = []
objects_to_update = []
statuses_to_report_on = get_plugin_setting_value(plugin, "REPORT_ON")
mylog('debug', ['[Plugins] statuses_to_report_on: ', statuses_to_report_on])
for plugObj in pluginObjects:
# keep old createdTime time if the plugObj already was created before
createdTime = plugObj.changed if plugObj.status == 'new' else plugObj.created
@@ -505,6 +503,8 @@ def process_plugin_events(db, plugin, pluginsState, plugEventsArr):
objects_to_update.append(values + (plugObj.index,)) # Include index for UPDATE
# only generate events that we want to be notified on
statuses_to_report_on = get_setting_value(plugObj.pluginPref + "_REPORT_ON")
if plugObj.status in statuses_to_report_on:
events_to_insert.append(values)
@@ -777,39 +777,24 @@ def handle_run(runType, db, pluginsState):
#-------------------------------------------------------------------------------
def handle_test(runType, db, pluginsState):
mylog('minimal', ['[', timeNowTZ(), '] START Test: ', testType])
mylog('minimal', ['[', timeNowTZ(), '] [Test] START Test: ', runType])
# TODO finish
# Prepare test samples
sample_txt = get_file_content(pialertPath + '/back/report_sample.txt')
sample_html = get_file_content(pialertPath + '/back/report_sample.html')
sample_json = json.loads(get_file_content(pialertPath + '/back/webhook_json_sample.json'))[0]["body"]["attachments"][0]["text"]
# Create fake notification
notification = Notification_obj(db)
notificationObj = notification.create(sample_json, sample_txt, sample_html, "")
# # Open text sample
# sample_txt = get_file_content(pialertPath + '/back/report_sample.txt')
# Run test
pluginsState = handle_run(runType, db, pluginsState)
# # Open html sample
# sample_html = get_file_content(pialertPath + '/back/report_sample.html')
# Remove sample notification
notificationObj.remove(notificationObj.GUID)
# # 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")
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])
# mylog('minimal', ['[Test Publishers] END Test: ', testType])
mylog('minimal', ['[Test] END Test: ', runType])
return pluginsState

19
pialert/plugin_utils.py
View File

@@ -77,11 +77,8 @@ def list_to_csv(arr):
arrayItemStr = ''
mylog('debug', '[Plugins] Flattening the below array')
mylog('debug', arr)
mylog('debug', f'[Plugins] isinstance(arr, list) : {isinstance(arr, list)}')
mylog('debug', f'[Plugins] isinstance(arr, str) : {isinstance(arr, str)}')
mylog('debug', arr)
mylog('debug', f'[Plugins] isinstance(arr, list) : {isinstance(arr, list)} | isinstance(arr, str) : {isinstance(arr, str)}')
if isinstance(arr, str):
return arr.replace('[','').replace(']','').replace("'", '') # removing brackets and single quotes (not allowed)
@@ -172,18 +169,6 @@ def get_plugins_configs():
return pluginsList # Return the list of plugin configurations
#-------------------------------------------------------------------------------
# Gets the setting value
def get_plugin_setting_value(plugin, function_key):
resultObj = get_plugin_setting(plugin, function_key)
if resultObj != None:
return resultObj["value"]
return None
#-------------------------------------------------------------------------------
def custom_plugin_decoder(pluginDict):
return namedtuple('X', pluginDict.keys())(*pluginDict.values())

23
pialert/reporting.py
View File

@@ -192,7 +192,7 @@ def get_notifications (db):
# collect "new_devices" for the webhook json
json_new_devices = notiStruc.json["data"]
mail_text = mail_text.replace ('<SECTION_NEW_DEVICES>', notiStruc.text + '\n')
mail_text = mail_text.replace ('<NEW_DEVICES_TABLE>', notiStruc.text + '\n')
mail_html = mail_html.replace ('<NEW_DEVICES_TABLE>', notiStruc.html)
mylog('verbose', ['[Notification] New Devices sections done.'])
@@ -208,7 +208,7 @@ def get_notifications (db):
# collect "down_devices" for the webhook json
json_down_devices = notiStruc.json["data"]
mail_text = mail_text.replace ('<SECTION_DEVICES_DOWN>', notiStruc.text + '\n')
mail_text = mail_text.replace ('<DOWN_DEVICES_TABLE>', notiStruc.text + '\n')
mail_html = mail_html.replace ('<DOWN_DEVICES_TABLE>', notiStruc.html)
mylog('verbose', ['[Notification] Down Devices sections done.'])
@@ -225,7 +225,7 @@ def get_notifications (db):
# collect "events" for the webhook json
json_events = notiStruc.json["data"]
mail_text = mail_text.replace ('<SECTION_EVENTS>', notiStruc.text + '\n')
mail_text = mail_text.replace ('<EVENTS_TABLE>', notiStruc.text + '\n')
mail_html = mail_html.replace ('<EVENTS_TABLE>', notiStruc.html)
mylog('verbose', ['[Notification] Events sections done.'])
@@ -256,7 +256,7 @@ def get_notifications (db):
final_text = removeDuplicateNewLines(mail_text)
# Create clickable MAC links
final_html = generate_mac_links (mail_html, deviceUrl)
final_html = generate_mac_links (mail_html, deviceUrl)
# Write output emails for debug
write_file (logPath + '/report_output.json', json.dumps(final_json))
@@ -265,12 +265,7 @@ def get_notifications (db):
return noti_obj(final_json, final_text, final_html)
# # Notify is something to report
# if hasNotifications:
# mylog('none', ['[Notification] Changes detected, sending reports'])
# msg = noti_obj(json_final, mail_text, mail_html)
# mylog('minimal', ['[Notification] Udating API files'])
# send_api()
@@ -281,12 +276,8 @@ def get_notifications (db):
# send_email (msg )
# else :
# mylog('verbose', ['[Notification] Skip email'])
# if conf.REPORT_APPRISE and check_config('apprise'):
# updateState("Send: Apprise")
# mylog('minimal', ['[Notification] Sending report by Apprise'])
# send_apprise (msg)
# else :
# mylog('verbose', ['[Notification] Skip Apprise'])
#
# if conf.REPORT_WEBHOOK and check_config('webhook'):
# updateState("Send: Webhook")
# mylog('minimal', ['[Notification] Sending report by Webhook'])