vh/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2026-03-31 07:12:23 -07:00
Code Issues Actions 15 Packages Projects Releases Wiki Activity

BE+FE: refactor totals retrieval + LUCIRPC old field name

Browse Source 
Signed-off-by: jokob-sk <jokob.sk@gmail.com>
This commit is contained in:
jokob-sk
2026-02-10 07:39:11 +11:00
parent cedbd59897
commit e899f657c5
10 changed files with 142 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
server/db/authoritative_handler.py
View File

@@ -19,6 +19,7 @@ from logger import mylog # noqa: E402 [flake8 lint suppression]
from helper import get_setting_value # noqa: E402 [flake8 lint suppression]
from db.db_helper import row_to_json # noqa: E402 [flake8 lint suppression]
from plugin_helper import normalize_mac # noqa: E402 [flake8 lint suppression]
from const import NULL_EQUIVALENTS # noqa: E402 [flake8 lint suppression]
# Map of field to its source tracking field
@@ -96,7 +97,7 @@ def can_overwrite_field(field_name, current_value, current_source, plugin_prefix
bool: True if overwrite allowed.
"""
empty_values = ("0.0.0.0", "", "null", "(unknown)", "(name not found)", None)
empty_values = ("0.0.0.0", *NULL_EQUIVALENTS, None)
# Rule 1: USER/LOCKED protected
if current_source in ("USER", "LOCKED"):
@@ -188,9 +189,7 @@ def get_source_for_field_update_with_value(
if isinstance(field_value, str):
stripped = field_value.strip()
if stripped in ("", "null"):
return "NEWDEV"
if stripped.lower() in ("(unknown)", "(name not found)"):
if stripped.lower() in NULL_EQUIVALENTS:
return "NEWDEV"
return plugin_prefix

39
server/db/db_helper.py
View File

@@ -6,14 +6,34 @@ import os
INSTALL_PATH = os.getenv("NETALERTX_APP", "/app")
sys.path.extend([f"{INSTALL_PATH}/server"])
from helper import if_byte_then_to_str # noqa: E402 [flake8 lint suppression]
from helper import if_byte_then_to_str, get_setting_value # noqa: E402 [flake8 lint suppression]
from logger import mylog # noqa: E402 [flake8 lint suppression]
from const import NULL_EQUIVALENTS_SQL # noqa: E402 [flake8 lint suppression]
def get_device_conditions():
network_dev_types = ",".join(f"'{v.replace('\'', '\'\'')}'" for v in get_setting_value("NETWORK_DEVICE_TYPES"))
conditions = {
"all": "WHERE devIsArchived=0",
"my": "WHERE devIsArchived=0",
"connected": "WHERE devPresentLastScan=1",
"favorites": "WHERE devIsArchived=0 AND devFavorite=1",
"new": "WHERE devIsArchived=0 AND devIsNew=1",
"down": "WHERE devIsArchived=0 AND devAlertDown != 0 AND devPresentLastScan=0",
"offline": "WHERE devIsArchived=0 AND devPresentLastScan=0",
"archived": "WHERE devIsArchived=1",
"network_devices": f"WHERE devIsArchived=0 AND devType in ({network_dev_types})",
"network_devices_down": f"WHERE devIsArchived=0 AND devType in ({network_dev_types}) AND devPresentLastScan=0",
"unknown": f"WHERE devIsArchived=0 AND devName in ({NULL_EQUIVALENTS_SQL})",
"known": f"WHERE devIsArchived=0 AND devName not in ({NULL_EQUIVALENTS_SQL})",
}
return conditions
# -------------------------------------------------------------------------------
# Return the SQL WHERE clause for filtering devices based on their status.
def get_device_condition_by_status(device_status):
"""
Return the SQL WHERE clause for filtering devices based on their status.
@@ -32,17 +52,8 @@ def get_device_condition_by_status(device_status):
str: SQL WHERE clause corresponding to the device status.
Defaults to 'WHERE 1=0' for unrecognized statuses.
"""
conditions = {
"all": "WHERE devIsArchived=0",
"my": "WHERE devIsArchived=0",
"connected": "WHERE devIsArchived=0 AND devPresentLastScan=1",
"favorites": "WHERE devIsArchived=0 AND devFavorite=1",
"new": "WHERE devIsArchived=0 AND devIsNew=1",
"down": "WHERE devIsArchived=0 AND devAlertDown != 0 AND devPresentLastScan=0",
"offline": "WHERE devIsArchived=0 AND devPresentLastScan=0",
"archived": "WHERE devIsArchived=1",
}
return conditions.get(device_status, "WHERE 1=0")
return get_device_conditions().get(device_status, "WHERE 1=0")
# -------------------------------------------------------------------------------

4
server/db/db_upgrade.py
View File

@@ -147,7 +147,7 @@ def ensure_mac_lowercase_triggers(sql):
except Exception as e:
mylog("none", [f"[db_upgrade] ERROR while ensuring MAC triggers: {e}"])
return False
def ensure_views(sql) -> bool:
"""
@@ -228,7 +228,7 @@ def ensure_views(sql) -> bool:
)
SELECT
d.*, -- all Device fields
r.* -- all CurrentScan fields (cur_*)
r.* -- all CurrentScan fields
FROM Devices d
LEFT JOIN RankedScans r
ON d.devMac = r.scanMac