vh/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2025-12-06 17:15:38 -08:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

BE: chore datetime_utils

Browse Source 
Signed-off-by: jokob-sk <jokob.sk@gmail.com>
This commit is contained in:
jokob-sk
2025-11-05 16:08:04 +11:00
parent 746f1a8922
commit c08eb1dbba
37 changed files with 234 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
server/__main__.py
View File

@@ -26,7 +26,8 @@ from pathlib import Path
import conf
from const import *
from logger import mylog
from helper import filePermissions, timeNowTZ, get_setting_value
from helper import filePermissions, get_setting_value
from utils.datetime_utils import timeNowTZ
from app_state import updateState
from api import update_api
from scan.session_events import process_scan

3
server/api.py
View File

@@ -7,7 +7,8 @@ import datetime
import conf
from const import (apiPath, sql_appevents, sql_devices_all, sql_events_pending_alert, sql_settings, sql_plugins_events, sql_plugins_history, sql_plugins_objects,sql_language_strings, sql_notifications_all, sql_online_history, sql_devices_tiles, sql_devices_filters)
from logger import mylog
from helper import write_file, get_setting_value, timeNowTZ
from helper import write_file, get_setting_value
from utils.datetime_utils import timeNowTZ
from app_state import updateState
from models.user_events_queue_instance import UserEventsQueueInstance
from messaging.in_app import write_notification

3
server/api_server/device_endpoint.py
View File

@@ -14,7 +14,8 @@ INSTALL_PATH="/app"
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
from database import get_temp_db_connection
from helper import is_random_mac, format_date, get_setting_value, timeNowDB
from helper import is_random_mac, get_setting_value
from utils.datetime_utils import timeNowDB, format_date
from db.db_helper import row_to_json, get_date_from_period
# --------------------------

3
server/api_server/devices_endpoint.py
View File

@@ -19,8 +19,9 @@ INSTALL_PATH="/app"
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
from database import get_temp_db_connection
from helper import is_random_mac, format_date, get_setting_value
from helper import is_random_mac, get_setting_value
from db.db_helper import get_table_json, get_device_condition_by_status
from utils.datetime_utils import format_date
# --------------------------

3
server/api_server/events_endpoint.py
View File

@@ -14,8 +14,9 @@ INSTALL_PATH="/app"
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
from database import get_temp_db_connection
from helper import is_random_mac, format_date, get_setting_value, format_date_iso, format_event_date, mylog, ensure_datetime
from helper import is_random_mac, get_setting_value, mylog
from db.db_helper import row_to_json, get_date_from_period
from utils.datetime_utils import format_date, format_date_iso, format_event_date, ensure_datetime
# --------------------------

3
server/api_server/history_endpoint.py
View File

@@ -14,7 +14,8 @@ INSTALL_PATH="/app"
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
from database import get_temp_db_connection
from helper import is_random_mac, format_date, get_setting_value
from helper import is_random_mac, get_setting_value
from utils.datetime_utils import format_date
# --------------------------------------------------

3
server/api_server/sessions_endpoint.py
View File

@@ -16,8 +16,9 @@ INSTALL_PATH="/app"
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
from database import get_temp_db_connection
from helper import is_random_mac, format_date, get_setting_value, format_date_iso, format_event_date, mylog, format_date_diff, format_ip_long, parse_datetime
from helper import is_random_mac, get_setting_value, mylog, format_ip_long
from db.db_helper import row_to_json, get_date_from_period
from utils.datetime_utils import format_date_iso, format_event_date, format_date_diff, parse_datetime, format_date
# --------------------------

3
server/api_server/sync_endpoint.py
View File

@@ -2,7 +2,8 @@ import os
import base64
from flask import jsonify, request
from logger import mylog
from helper import get_setting_value, timeNowDB
from helper import get_setting_value
from utils.datetime_utils import timeNowDB
from messaging.in_app import write_notification
INSTALL_PATH = "/app"

3
server/app_state.py
View File

@@ -4,7 +4,8 @@ import json
import conf
from const import *
from logger import mylog, logResult
from helper import timeNowDB, timeNow, checkNewVersion
from helper import checkNewVersion
from utils.datetime_utils import timeNowDB, timeNow
# Register NetAlertX directories
INSTALL_PATH="/app"

139
server/helper.py
View File

@@ -7,7 +7,6 @@ import os
import re
import unicodedata
import subprocess
from typing import Union
import pytz
from pytz import timezone
import json
@@ -29,144 +28,6 @@ from logger import mylog, logResult
# Register NetAlertX directories
INSTALL_PATH="/app"
#-------------------------------------------------------------------------------
# DateTime
#-------------------------------------------------------------------------------
def timeNowTZ():
if conf.tz:
return datetime.datetime.now(conf.tz).replace(microsecond=0)
else:
return datetime.datetime.now().replace(microsecond=0)
def timeNow():
return datetime.datetime.now().replace(microsecond=0)
def get_timezone_offset():
now = datetime.datetime.now(conf.tz)
offset_hours = now.utcoffset().total_seconds() / 3600
offset_formatted = "{:+03d}:{:02d}".format(int(offset_hours), int((offset_hours % 1) * 60))
return offset_formatted
def timeNowDB(local=True):
"""
Return the current time (local or UTC) as ISO 8601 for DB storage.
Safe for SQLite, PostgreSQL, etc.
Example local: '2025-11-04 18:09:11'
Example UTC: '2025-11-04 07:09:11'
"""
if local:
try:
if isinstance(conf.tz, datetime.tzinfo):
tz = conf.tz
elif conf.tz:
tz = ZoneInfo(conf.tz)
else:
tz = None
except Exception:
tz = None
return datetime.datetime.now(tz).strftime('%Y-%m-%d %H:%M:%S')
else:
return datetime.datetime.now(datetime.UTC).strftime('%Y-%m-%d %H:%M:%S')
#-------------------------------------------------------------------------------
# Date and time methods
#-------------------------------------------------------------------------------
# -------------------------------------------------------------------------------------------
def format_date_iso(date1: str) -> str:
"""Return ISO 8601 string for a date or None if empty"""
if date1 is None:
return None
dt = datetime.datetime.fromisoformat(date1) if isinstance(date1, str) else date1
return dt.isoformat()
# -------------------------------------------------------------------------------------------
def format_event_date(date_str: str, event_type: str) -> str:
"""Format event date with fallback rules."""
if date_str:
return format_date(date_str)
elif event_type == "<missing event>":
return "<missing event>"
else:
return "<still connected>"
# -------------------------------------------------------------------------------------------
def ensure_datetime(dt: Union[str, datetime.datetime, None]) -> datetime.datetime:
if dt is None:
return timeNowTZ()
if isinstance(dt, str):
return datetime.datetime.fromisoformat(dt)
return dt
def parse_datetime(dt_str):
if not dt_str:
return None
try:
# Try ISO8601 first
return datetime.datetime.fromisoformat(dt_str)
except ValueError:
# Try RFC1123 / HTTP format
try:
return datetime.datetime.strptime(dt_str, '%a, %d %b %Y %H:%M:%S GMT')
except ValueError:
return None
def format_date(date_str: str) -> str:
try:
dt = parse_datetime(date_str)
if dt.tzinfo is None:
# Set timezone if missing — change to timezone.utc if you prefer UTC
now = datetime.datetime.now(conf.tz)
dt = dt.replace(tzinfo=now.astimezone().tzinfo)
return dt.astimezone().isoformat()
except Exception:
return "invalid"
def format_date_diff(date1, date2):
"""
Return difference between two datetimes as 'Xd HH:MM'.
Uses app timezone if datetime is naive.
date2 can be None (uses now).
"""
# Get timezone from settings
tz_name = get_setting_value("TIMEZONE") or "UTC"
tz = pytz.timezone(tz_name)
def parse_dt(dt):
if dt is None:
return datetime.datetime.now(tz)
if isinstance(dt, str):
try:
dt_parsed = email.utils.parsedate_to_datetime(dt)
except Exception:
# fallback: parse ISO string
dt_parsed = datetime.datetime.fromisoformat(dt)
# convert naive GMT/UTC to app timezone
if dt_parsed.tzinfo is None:
dt_parsed = tz.localize(dt_parsed)
else:
dt_parsed = dt_parsed.astimezone(tz)
return dt_parsed
return dt if dt.tzinfo else tz.localize(dt)
dt1 = parse_dt(date1)
dt2 = parse_dt(date2)
delta = dt2 - dt1
total_minutes = int(delta.total_seconds() // 60)
days, rem_minutes = divmod(total_minutes, 1440) # 1440 mins in a day
hours, minutes = divmod(rem_minutes, 60)
return {
"text": f"{days}d {hours:02}:{minutes:02}",
"days": days,
"hours": hours,
"minutes": minutes,
"total_minutes": total_minutes
}
#-------------------------------------------------------------------------------
# File system permission handling

3
server/initialise.py
View File

@@ -12,7 +12,8 @@ import re
# Register NetAlertX libraries
import conf
from const import fullConfPath, applicationPath, fullConfFolder, default_tz
from helper import getBuildTimeStamp, fixPermissions, collect_lang_strings, updateSubnets, isJsonObject, setting_value_to_python_type, timeNowDB, get_setting_value, generate_random_string
from helper import getBuildTimeStamp, fixPermissions, collect_lang_strings, updateSubnets, isJsonObject, setting_value_to_python_type, get_setting_value, generate_random_string
from utils.datetime_utils import timeNowDB
from app_state import updateState
from logger import mylog
from api import update_api

37
server/logger.py
View File

@@ -7,40 +7,15 @@ import time
import logging
from zoneinfo import ZoneInfo
# Register NetAlertX directories
INSTALL_PATH="/app"
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
# NetAlertX imports
import conf
from const import *
#-------------------------------------------------------------------------------
# duplication from helper to avoid circle
#-------------------------------------------------------------------------------
def timeNowTZ():
if conf.tz:
return datetime.datetime.now(conf.tz).replace(microsecond=0)
else:
return datetime.datetime.now().replace(microsecond=0)
def timeNowDB(local=True):
"""
Return the current time (local or UTC) as ISO 8601 for DB storage.
Safe for SQLite, PostgreSQL, etc.
Example local: '2025-11-04 18:09:11'
Example UTC: '2025-11-04 07:09:11'
"""
if local:
try:
if isinstance(conf.tz, datetime.tzinfo):
tz = conf.tz
elif conf.tz:
tz = ZoneInfo(conf.tz)
else:
tz = None
except Exception:
tz = None
return datetime.datetime.now(tz).strftime('%Y-%m-%d %H:%M:%S')
else:
return datetime.datetime.now(datetime.UTC).strftime('%Y-%m-%d %H:%M:%S')
from utils.datetime_utils import timeNowTZ
#-------------------------------------------------------------------------------
# Map custom debug levels to Python logging levels

3
server/messaging/in_app.py
View File

@@ -20,7 +20,8 @@ sys.path.extend([f"{INSTALL_PATH}/server"])
import conf
from const import applicationPath, logPath, apiPath, confFileName, reportTemplatesPath
from logger import logResult, mylog
from helper import generate_mac_links, removeDuplicateNewLines, timeNowDB, get_file_content, write_file, get_setting_value, get_timezone_offset
from helper import generate_mac_links, removeDuplicateNewLines, get_file_content, write_file, get_setting_value
from utils.datetime_utils import timeNowDB
NOTIFICATION_API_FILE = apiPath + 'user_notifications.json'

3
server/messaging/reporting.py
View File

@@ -20,9 +20,10 @@ sys.path.extend([f"{INSTALL_PATH}/server"])
import conf
from const import applicationPath, logPath, apiPath, confFileName
from helper import get_file_content, write_file, get_timezone_offset, get_setting_value
from helper import get_file_content, write_file, get_setting_value
from logger import logResult, mylog
from db.sql_safe_builder import create_safe_condition_builder
from utils.datetime_utils import get_timezone_offset
#===============================================================================
# REPORTING

6
server/models/notification_instance.py
View File

@@ -16,12 +16,10 @@ from const import applicationPath, logPath, apiPath, reportTemplatesPath
from logger import mylog, Logger
from helper import generate_mac_links, \
removeDuplicateNewLines, \
timeNowDB, \
timeNowTZ, \
write_file, \
get_setting_value, \
get_timezone_offset
get_setting_value
from messaging.in_app import write_notification
from utils.datetime_utils import timeNowDB, get_timezone_offset
# -----------------------------------------------------------------------------

3
server/plugin.py
View File

@@ -12,7 +12,8 @@ from collections import namedtuple
import conf
from const import pluginsPath, logPath, applicationPath, reportTemplatesPath
from logger import mylog, Logger
from helper import timeNowDB, timeNowTZ, get_file_content, write_file, get_setting, get_setting_value
from helper import get_file_content, write_file, get_setting, get_setting_value
from utils.datetime_utils import timeNowTZ, timeNowDB
from app_state import updateState
from api import update_api
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, decode_and_rename_files

3
server/scan/device_handling.py
View File

@@ -10,7 +10,8 @@ from dateutil import parser
INSTALL_PATH="/app"
sys.path.extend([f"{INSTALL_PATH}/server"])
from helper import timeNowDB, timeNowTZ, get_setting_value, check_IP_format
from helper import get_setting_value, check_IP_format
from utils.datetime_utils import timeNowDB
from logger import mylog, Logger
from const import vendorsPath, vendorsPathNewest, sql_generateGuid
from models.device_instance import DeviceInstance

3
server/scan/session_events.py
View File

@@ -6,7 +6,8 @@ sys.path.extend([f"{INSTALL_PATH}/server"])
import conf
from scan.device_handling import create_new_devices, print_scan_stats, save_scanned_devices, exclude_ignored_devices, update_devices_data_from_scan
from helper import timeNowDB, get_setting_value
from helper import get_setting_value
from utils.datetime_utils import timeNowDB
from db.db_helper import print_table_schema
from logger import mylog, Logger
from messaging.reporting import skip_repeated_notifications

162
server/utils/datetime_utils.py Normal file
View File

@@ -0,0 +1,162 @@
#!/usr/bin/env python
import os
import pathlib
import sys
from datetime import datetime
import pytz
from pytz import timezone
import datetime
from typing import Union
# Register NetAlertX directories
INSTALL_PATH="/app"
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
# Register NetAlertX directories
INSTALL_PATH="/app"
import conf
from const import *
#-------------------------------------------------------------------------------
# DateTime
#-------------------------------------------------------------------------------
def timeNowTZ():
if conf.tz:
return datetime.datetime.now(conf.tz).replace(microsecond=0)
else:
return datetime.datetime.now().replace(microsecond=0)
def timeNow():
return datetime.datetime.now().replace(microsecond=0)
def get_timezone_offset():
now = datetime.datetime.now(conf.tz)
offset_hours = now.utcoffset().total_seconds() / 3600
offset_formatted = "{:+03d}:{:02d}".format(int(offset_hours), int((offset_hours % 1) * 60))
return offset_formatted
def timeNowDB(local=True):
"""
Return the current time (local or UTC) as ISO 8601 for DB storage.
Safe for SQLite, PostgreSQL, etc.
Example local: '2025-11-04 18:09:11'
Example UTC: '2025-11-04 07:09:11'
"""
if local:
try:
if isinstance(conf.tz, datetime.tzinfo):
tz = conf.tz
elif conf.tz:
tz = ZoneInfo(conf.tz)
else:
tz = None
except Exception:
tz = None
return datetime.datetime.now(tz).strftime('%Y-%m-%d %H:%M:%S')
else:
return datetime.datetime.now(datetime.UTC).strftime('%Y-%m-%d %H:%M:%S')
#-------------------------------------------------------------------------------
# Date and time methods
#-------------------------------------------------------------------------------
# -------------------------------------------------------------------------------------------
def format_date_iso(date1: str) -> str:
"""Return ISO 8601 string for a date or None if empty"""
if date1 is None:
return None
dt = datetime.datetime.fromisoformat(date1) if isinstance(date1, str) else date1
return dt.isoformat()
# -------------------------------------------------------------------------------------------
def format_event_date(date_str: str, event_type: str) -> str:
"""Format event date with fallback rules."""
if date_str:
return format_date(date_str)
elif event_type == "<missing event>":
return "<missing event>"
else:
return "<still connected>"
# -------------------------------------------------------------------------------------------
def ensure_datetime(dt: Union[str, datetime.datetime, None]) -> datetime.datetime:
if dt is None:
return timeNowTZ()
if isinstance(dt, str):
return datetime.datetime.fromisoformat(dt)
return dt
def parse_datetime(dt_str):
if not dt_str:
return None
try:
# Try ISO8601 first
return datetime.datetime.fromisoformat(dt_str)
except ValueError:
# Try RFC1123 / HTTP format
try:
return datetime.datetime.strptime(dt_str, '%a, %d %b %Y %H:%M:%S GMT')
except ValueError:
return None
def format_date(date_str: str) -> str:
try:
dt = parse_datetime(date_str)
if dt.tzinfo is None:
# Set timezone if missing — change to timezone.utc if you prefer UTC
now = datetime.datetime.now(conf.tz)
dt = dt.replace(tzinfo=now.astimezone().tzinfo)
return dt.astimezone().isoformat()
except Exception:
return "invalid"
def format_date_diff(date1, date2):
"""
Return difference between two datetimes as 'Xd HH:MM'.
Uses app timezone if datetime is naive.
date2 can be None (uses now).
"""
# Get timezone from settings
tz_name = get_setting_value("TIMEZONE") or "UTC"
tz = pytz.timezone(tz_name)
def parse_dt(dt):
if dt is None:
return datetime.datetime.now(tz)
if isinstance(dt, str):
try:
dt_parsed = email.utils.parsedate_to_datetime(dt)
except Exception:
# fallback: parse ISO string
dt_parsed = datetime.datetime.fromisoformat(dt)
# convert naive GMT/UTC to app timezone
if dt_parsed.tzinfo is None:
dt_parsed = tz.localize(dt_parsed)
else:
dt_parsed = dt_parsed.astimezone(tz)
return dt_parsed
return dt if dt.tzinfo else tz.localize(dt)
dt1 = parse_dt(date1)
dt2 = parse_dt(date2)
delta = dt2 - dt1
total_minutes = int(delta.total_seconds() // 60)
days, rem_minutes = divmod(total_minutes, 1440) # 1440 mins in a day
hours, minutes = divmod(rem_minutes, 60)
return {
"text": f"{days}d {hours:02}:{minutes:02}",
"days": days,
"hours": hours,
"minutes": minutes,
"total_minutes": total_minutes
}