From 872ac1ce0f48c3996d709605b97404ddd9127ec6 Mon Sep 17 00:00:00 2001 From: jokob-sk Date: Sat, 22 Nov 2025 21:06:03 +1100 Subject: [PATCH] BE: linting fixes 3 Signed-off-by: jokob-sk --- front/plugins/_publisher_mqtt/mqtt.py | 4 +--- front/plugins/_publisher_ntfy/ntfy.py | 1 - front/plugins/_publisher_pushsafer/pushsafer.py | 4 +--- front/plugins/freebox/freebox.py | 3 ++- front/plugins/pihole_api_scan/pihole_api_scan.py | 4 ++-- server/api.py | 3 +-- server/api_server/devices_endpoint.py | 2 +- server/api_server/graphql_endpoint.py | 1 - server/db/db_upgrade.py | 2 +- server/helper.py | 2 -- server/plugin.py | 5 ++++- server/utils/datetime_utils.py | 6 +++--- 12 files changed, 16 insertions(+), 21 deletions(-) diff --git a/front/plugins/_publisher_mqtt/mqtt.py b/front/plugins/_publisher_mqtt/mqtt.py index 76e6afea..a087d255 100755 --- a/front/plugins/_publisher_mqtt/mqtt.py +++ b/front/plugins/_publisher_mqtt/mqtt.py @@ -233,7 +233,6 @@ class sensor_config: Store the sensor configuration in the global plugin_objects, which tracks sensors based on a unique combination of attributes including deviceId, sensorName, hash, and MAC. """ - global plugin_objects # Add the sensor to the global plugin_objects plugin_objects.add_object( @@ -318,7 +317,6 @@ def create_generic_device(mqtt_client, deviceId, deviceName): # ------------------------------------------------------------------------------ # Register sensor config on the broker def create_sensor(mqtt_client, deviceId, deviceName, sensorType, sensorName, icon, mac=""): - global mqtt_sensors # check previous configs sensorConfig = sensor_config(deviceId, deviceName, sensorType, sensorName, icon, mac) @@ -429,7 +427,7 @@ def mqtt_create_client(): # ----------------------------------------------------------------------------- def mqtt_start(db): - global mqtt_client, mqtt_connected_to_broker + global mqtt_client if not mqtt_connected_to_broker: mqtt_client = mqtt_create_client() diff --git a/front/plugins/_publisher_ntfy/ntfy.py b/front/plugins/_publisher_ntfy/ntfy.py index 9d86be91..16a75d8c 100755 --- a/front/plugins/_publisher_ntfy/ntfy.py +++ b/front/plugins/_publisher_ntfy/ntfy.py @@ -1,4 +1,3 @@ - #!/usr/bin/env python import json diff --git a/front/plugins/_publisher_pushsafer/pushsafer.py b/front/plugins/_publisher_pushsafer/pushsafer.py index a46681e5..b186c0a3 100755 --- a/front/plugins/_publisher_pushsafer/pushsafer.py +++ b/front/plugins/_publisher_pushsafer/pushsafer.py @@ -1,6 +1,4 @@ - #!/usr/bin/env python - import json import os import sys @@ -99,7 +97,7 @@ def send(text): "ut" : 'Open NetAlertX', "k" : token, } - response = requests.post(url, data=post_fields) + response = requests.post(url, data=post_fields, timeout=get_setting_value("PUSHSAFER_RUN_TIMEOUT")) response_status_code = response.status_code # Check if the request was successful (status code 200) diff --git a/front/plugins/freebox/freebox.py b/front/plugins/freebox/freebox.py index 9c3b8ea9..00539ae0 100755 --- a/front/plugins/freebox/freebox.py +++ b/front/plugins/freebox/freebox.py @@ -22,6 +22,7 @@ from logger import mylog, Logger # noqa: E402 [flake8 lint suppression] from const import logPath # noqa: E402 [flake8 lint suppression] from helper import get_setting_value # noqa: E402 [flake8 lint suppression] import conf # noqa: E402 [flake8 lint suppression] +from utils.datetime_utils import timeNowDB # noqa: E402 [flake8 lint suppression] # Make sure the TIMEZONE for logging is correct conf.tz = timezone(get_setting_value("TIMEZONE")) @@ -150,7 +151,7 @@ def main(): watched1=freebox["name"], watched2=freebox["operator"], watched3="Gateway", - watched4=datetime.now, + watched4=timeNowDB(), extra="", foreignKey=freebox["mac"], ) diff --git a/front/plugins/pihole_api_scan/pihole_api_scan.py b/front/plugins/pihole_api_scan/pihole_api_scan.py index d9f9822a..09ff5f69 100644 --- a/front/plugins/pihole_api_scan/pihole_api_scan.py +++ b/front/plugins/pihole_api_scan/pihole_api_scan.py @@ -268,7 +268,7 @@ def main(): if is_mac(entry['mac']): # Map to Plugin_Objects fields - mylog('verbose', [f'[{pluginName}] found: {entry['name']}|{entry['mac']}|{entry['ip']}']) + mylog('verbose', [f"[{pluginName}] found: {entry['name']}|{entry['mac']}|{entry['ip']}"]) plugin_objects.add_object( primaryId=str(entry['mac']), @@ -281,7 +281,7 @@ def main(): foreignKey=str(entry['mac']) ) else: - mylog('verbose', [f'[{pluginName}] Skipping invalid MAC: {entry['name']}|{entry['mac']}|{entry['ip']}']) + mylog('verbose', [f"[{pluginName}] Skipping invalid MAC: {entry['name']}|{entry['mac']}|{entry['ip']}"]) # Write result file for NetAlertX to ingest plugin_objects.write_result_file() diff --git a/server/api.py b/server/api.py index bfb801d7..9ea8d5ad 100755 --- a/server/api.py +++ b/server/api.py @@ -111,7 +111,6 @@ def update_api( # ------------------------------------------------------------------------------- class api_endpoint_class: def __init__(self, db, forceUpdate, query, path, is_ad_hoc_user_event=False): - global apiEndpoints current_time = timeNowTZ() @@ -222,7 +221,7 @@ periodic_write_thread = None def periodic_write(interval=1): """Periodically checks all endpoints for pending writes.""" - global apiEndpoints + while not stop_event.is_set(): with api_lock: for endpoint in apiEndpoints: diff --git a/server/api_server/devices_endpoint.py b/server/api_server/devices_endpoint.py index 003fa66c..e924aec4 100755 --- a/server/api_server/devices_endpoint.py +++ b/server/api_server/devices_endpoint.py @@ -96,7 +96,7 @@ def delete_unknown_devices(): def export_devices(export_format): """ - Export devices from the Devices table in teh desired format. + Export devices from the Devices table in the desired format. - If `macs` is None → delete ALL devices. - If `macs` is a list → delete only matching MACs (supports wildcard '*'). """ diff --git a/server/api_server/graphql_endpoint.py b/server/api_server/graphql_endpoint.py index 9ea995bf..78b1f3f8 100755 --- a/server/api_server/graphql_endpoint.py +++ b/server/api_server/graphql_endpoint.py @@ -364,7 +364,6 @@ class Query(ObjectType): Collect language strings, optionally filtered by language code and/or string key. Caches in memory for performance. Can fallback to 'en_us' if a string is missing. """ - global _langstrings_cache, _langstrings_cache_mtime langStrings = [] diff --git a/server/db/db_upgrade.py b/server/db/db_upgrade.py index 97da3dd0..35e6b58b 100755 --- a/server/db/db_upgrade.py +++ b/server/db/db_upgrade.py @@ -30,7 +30,7 @@ def ensure_column(sql, table: str, column_name: str, column_type: str) -> bool: if column_name in actual_columns: return True # Already exists - # Define the expected columns (hardcoded base schema) [v25.5.24] - available in teh default app.db + # Define the expected columns (hardcoded base schema) [v25.5.24] - available in the default app.db expected_columns = [ "devMac", "devName", diff --git a/server/helper.py b/server/helper.py index df355708..5c36bbf9 100755 --- a/server/helper.py +++ b/server/helper.py @@ -248,8 +248,6 @@ def get_setting_value(key): Any: The Python-typed setting value, or an empty string if not found. """ - global SETTINGS_SECONDARYCACHE - # Returns empty string if not found value = "" diff --git a/server/plugin.py b/server/plugin.py index ee64290b..6986b706 100755 --- a/server/plugin.py +++ b/server/plugin.py @@ -94,7 +94,10 @@ class plugin_manager: # 🔹 CMD also retrieved from cache cmd_setting = self._cache["settings"].get(prefix, {}).get("CMD") - mylog("debug", f"[Plugins] CMD: {cmd_setting["value"] if cmd_setting else None}") + + print_str = cmd_setting["value"] if cmd_setting else None + + mylog("debug", f"[Plugins] CMD: {print_str}") execute_plugin(self.db, self.all_plugins, plugin) diff --git a/server/utils/datetime_utils.py b/server/utils/datetime_utils.py index 9c30d3bb..d5d54333 100644 --- a/server/utils/datetime_utils.py +++ b/server/utils/datetime_utils.py @@ -5,7 +5,7 @@ from dateutil import parser import datetime import re import pytz -from typing import Union +from typing import Union, Optional from zoneinfo import ZoneInfo import email.utils import conf @@ -112,9 +112,9 @@ def normalizeTimeStamp(inputTimeStamp): # ------------------------------------------------------------------------------------------- -def format_date_iso(date1: str) -> str: +def format_date_iso(date1: str) -> Optional[str]: """Return ISO 8601 string for a date or None if empty""" - if date1 is None: + if not date1: return None dt = datetime.datetime.fromisoformat(date1) if isinstance(date1, str) else date1 return dt.isoformat()