⚙ NAME_CLEANUP_REGEX #735 #728

This commit is contained in:
jokob-sk
2024-07-10 11:27:21 +10:00
parent 41b5de9292
commit 9f44c0de01
3 changed files with 235 additions and 155 deletions

82
front/plugins/__test/test.py Executable file
View File

@@ -0,0 +1,82 @@
#!/usr/bin/env python
# Just a testing library plugin for development purposes
import json
import subprocess
import argparse
import os
import pathlib
import sys
from datetime import datetime
import time
import re
import hashlib
import sqlite3
# Register NetAlertX directories
INSTALL_PATH="/app"
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
# NetAlertX modules
import conf
from const import apiPath, confFileName
from plugin_utils import getPluginObject
from plugin_helper import Plugin_Objects
from logger import mylog, append_line_to_file
from helper import timeNowTZ, get_setting_value, bytes_to_string, sanitize_string, cleanDeviceName
from notification import Notification_obj
from database import DB, get_device_stats
CUR_PATH = str(pathlib.Path(__file__).parent.resolve())
RESULT_FILE = os.path.join(CUR_PATH, 'last_result.log')
# Initialize the Plugin obj output file
plugin_objects = Plugin_Objects(RESULT_FILE)
# Create an MD5 hash object
md5_hash = hashlib.md5()
pluginName = 'TESTONLY'
# globals
def main():
# START
mylog('verbose', [f'[{pluginName}] In script'])
# SPACE FOR TESTING 🔽
str = "ABC-MBP._another.localdomain."
# cleanDeviceName(str, match_IP)
# result = cleanDeviceName(str, True)
regexes = get_setting_value('NEWDEV_NAME_CLEANUP_REGEX')
print(regexes)
subnets = get_setting_value('SCAN_SUBNETS')
print(subnets)
for rgx in regexes:
mylog('debug', ["[cleanDeviceName] applying regex : " + rgx])
mylog('debug', ["[cleanDeviceName] name before regex : " + str])
str = re.sub(rgx, "", str)
mylog('debug', ["[cleanDeviceName] name after regex : " + str])
mylog('debug', ["[cleanDeviceName] output: " + str])
# SPACE FOR TESTING 🔼
# END
mylog('verbose', [f'[{pluginName}] result "{str}"'])
# -------------INIT---------------------
if __name__ == '__main__':
sys.exit(main())

View File

@@ -176,6 +176,120 @@
} }
] ]
}, },
{
"function": "LESS_NAME_CLEANUP",
"type": {
"dataType": "boolean",
"elements": [
{
"elementType": "input",
"elementOptions": [{ "type": "checkbox" }],
"transformers": []
}
]
},
"default_value": 0,
"options": [],
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "Less Name Cleanup"
}
],
"description": [
{
"language_code": "en_us",
"string": "Check to start using the new code for cleaning device names. Removes all labels starting with underscore and removes network domain and search list."
}
]
},
{
"function": "NAME_CLEANUP_REGEX",
"type": {
"dataType": "array",
"elements": [
{
"elementType": "input",
"elementOptions": [
{ "placeholder": "Enter value" },
{ "suffix": "_in" },
{ "cssClasses": "col-sm-10" },
{ "prefillValue": "null" }
],
"transformers": []
},
{
"elementType": "button",
"elementOptions": [
{ "sourceSuffixes": [] },
{ "separator": "" },
{ "cssClasses": "col-sm-3" },
{ "onClick": "removeAllOptions(this)" },
{ "getStringKey": "Gen_Remove_All" }
],
"transformers": []
},
{
"elementType": "button",
"elementOptions": [
{ "sourceSuffixes": [] },
{ "separator": "" },
{ "cssClasses": "col-sm-3" },
{ "onClick": "removeFromList(this)" },
{ "getStringKey": "Gen_Remove_Last" }
],
"transformers": []
},
{
"elementType": "button",
"elementOptions": [
{ "sourceSuffixes": ["_in"] },
{ "separator": "" },
{ "cssClasses": "col-sm-2" },
{ "onClick": "addList(this, false)" },
{ "getStringKey": "Gen_Add" }
],
"transformers": []
},
{
"elementType": "select",
"elementOptions": [
{ "multiple": "true" },
{ "readonly": "true" },
{ "editable": "true" }
],
"transformers": ["base64"]
}
]
},
"default_value": [
"XC5fYWlycGxheQ==",
"XC5fdGNw",
"XC5sb2NhbGRvbWFpbg==",
"XC5sb2NhbA==",
"XC5fZXNwaG9tZWxpYg==",
"XC5fZ29vZ2xlY2FzdA==",
"XC5sYW4=",
"XC5ob21l",
"LVthLWZBLUYwLTldezMyfQ==",
"Iy4q"
],
"options": [],
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "Name Cleanup REGEX"
}
],
"description": [
{
"language_code": "en_us",
"string": "All the newly discovered device names are clened up by applying the following REGEX expression in this order. All the below are replaced by a blank string."
}
]
},
{ {
"function": "dev_MAC", "function": "dev_MAC",
"type": { "type": {
@@ -913,120 +1027,6 @@
"string": "The icon associated with the device. Check the <a href=\"https://github.com/jokob-sk/NetAlertX/blob/main/docs/ICONS.md\" target=\"_blank\">documentation on icons</a> for more details." "string": "The icon associated with the device. Check the <a href=\"https://github.com/jokob-sk/NetAlertX/blob/main/docs/ICONS.md\" target=\"_blank\">documentation on icons</a> for more details."
} }
] ]
},
{
"function": "LESS_NAME_CLEANUP",
"type": {
"dataType": "boolean",
"elements": [
{
"elementType": "input",
"elementOptions": [{ "type": "checkbox" }],
"transformers": []
}
]
},
"default_value": 0,
"options": [],
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "Less Name Cleanup"
}
],
"description": [
{
"language_code": "en_us",
"string": "Check to start using the new code for cleaning device names. Removes all labels starting with underscore and removes network domain and search list."
}
]
},
{
"function": "NAME_CLEANUP_REGEX",
"type": {
"dataType": "array",
"elements": [
{
"elementType": "input",
"elementOptions": [
{ "placeholder": "Enter value" },
{ "suffix": "_in" },
{ "cssClasses": "col-sm-10" },
{ "prefillValue": "null" }
],
"transformers": []
},
{
"elementType": "button",
"elementOptions": [
{ "sourceSuffixes": [] },
{ "separator": "" },
{ "cssClasses": "col-sm-3" },
{ "onClick": "removeAllOptions(this)" },
{ "getStringKey": "Gen_Remove_All" }
],
"transformers": []
},
{
"elementType": "button",
"elementOptions": [
{ "sourceSuffixes": [] },
{ "separator": "" },
{ "cssClasses": "col-sm-3" },
{ "onClick": "removeFromList(this)" },
{ "getStringKey": "Gen_Remove_Last" }
],
"transformers": []
},
{
"elementType": "button",
"elementOptions": [
{ "sourceSuffixes": ["_in"] },
{ "separator": "" },
{ "cssClasses": "col-sm-2" },
{ "onClick": "addList(this, false)" },
{ "getStringKey": "Gen_Add" }
],
"transformers": []
},
{
"elementType": "select",
"elementOptions": [
{ "multiple": "true" },
{ "readonly": "true" },
{ "editable": "true" }
],
"transformers": ["base64"]
}
]
},
"default_value": [
"XC5fYWlycGxheQ==",
"XC5fdGNw",
"XC5sb2NhbGRvbWFpbg==",
"XC5sb2NhbA==",
"XC5fZXNwaG9tZWxpYg==",
"XC5fZ29vZ2xlY2FzdA==",
"XC5sYW4=",
"XC5ob21l",
"LVthLWZBLUYwLTldezMyfQ==",
"Iy4q"
],
"options": [],
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "Cleanup REGEX"
}
],
"description": [
{
"language_code": "en_us",
"string": "All the newly discovered device names are clened up by applying the following REGEX expression in this order."
}
]
} }
], ],
"required": [ "required": [

View File

@@ -342,22 +342,12 @@ def setting_value_to_python_type(set_type, set_value):
elementOptions = last_element.get('elementOptions', []) elementOptions = last_element.get('elementOptions', [])
transformers = last_element.get('transformers', []) transformers = last_element.get('transformers', [])
# Apply transformers to the value
for transformer in transformers:
if transformer == 'base64':
if isinstance(set_value, str):
set_value = base64.b64decode(set_value).decode('utf-8')
elif transformer == 'sha256':
if isinstance(set_value, str):
set_value = hashlib.sha256(set_value.encode()).hexdigest()
# Convert value based on dataType and elementType # Convert value based on dataType and elementType
if dataType == 'string' and elementType in ['input', 'select']: if dataType == 'string' and elementType in ['input', 'select']:
value = str(set_value) value = reverseTransformers(str(set_value))
elif dataType == 'integer' and (elementType == 'input' or elementType == 'select'): elif dataType == 'integer' and (elementType == 'input' or elementType == 'select'):
# handle storing/retrieving boolean values as 1/0 # handle storing/retrieving boolean values as 1/0
if set_value.lower() not in ['true', 'false'] and isinstance(set_value, str): if set_value.lower() not in ['true', 'false'] and isinstance(set_value, str):
value = int(set_value) value = int(set_value)
@@ -365,37 +355,46 @@ def setting_value_to_python_type(set_type, set_value):
value = 1 if set_value else 0 value = 1 if set_value else 0
elif isinstance(set_value, str): elif isinstance(set_value, str):
value = 1 if set_value.lower() == 'true' else 0 value = 1 if set_value.lower() == 'true' else 0
else: else:
value = int(set_value) value = int(set_value)
# boolean handling
elif dataType == 'boolean' and elementType == 'input': elif dataType == 'boolean' and elementType == 'input':
value = set_value.lower() in ['true', '1'] value = set_value.lower() in ['true', '1']
# array handling
elif dataType == 'array' and elementType == 'select': elif dataType == 'array' and elementType == 'select':
if isinstance(set_value, str): if isinstance(set_value, str):
try: try:
value = json.loads(set_value.replace("'", "\"")) value = json.loads(set_value.replace("'", "\""))
# reverse transformations to all entries
value = reverseTransformers(value, transformers)
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
print(f"Error decoding JSON array: {e}") mylog('none', [f'[setting_value_to_python_type] Error decoding JSON object: {e}'])
mylog('none', [set_value])
value = [] value = []
elif isinstance(set_value, list): elif isinstance(set_value, list):
value = set_value value = set_value
elif dataType == 'object' and elementType == 'input': elif dataType == 'object' and elementType == 'input':
if isinstance(set_value, str): if isinstance(set_value, str):
try: try:
value = json.loads(set_value) value = reverseTransformers(json.loads(set_value))
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
print(f"Error decoding JSON object: {e}") mylog('none', [f'[setting_value_to_python_type] Error decoding JSON object: {e}'])
mylog('none', [{set_value}])
value = {} value = {}
elif isinstance(set_value, dict): elif isinstance(set_value, dict):
value = set_value value = set_value
elif dataType == 'string' and elementType == 'input' and any(opt.get('readonly') == "true" for opt in elementOptions): elif dataType == 'string' and elementType == 'input' and any(opt.get('readonly') == "true" for opt in elementOptions):
value = str(set_value) value = reverseTransformers(str(set_value))
elif dataType == 'string' and elementType == 'input' and any(opt.get('type') == "password" for opt in elementOptions) and 'sha256' in transformers: elif dataType == 'string' and elementType == 'input' and any(opt.get('type') == "password" for opt in elementOptions) and 'sha256' in transformers:
value = hashlib.sha256(set_value.encode()).hexdigest() value = hashlib.sha256(set_value.encode()).hexdigest()
@@ -407,8 +406,24 @@ def setting_value_to_python_type(set_type, set_value):
return value return value
#-------------------------------------------------------------------------------
# Reverse transformed values if needed
def reverseTransformers(val, transformers):
# Function to apply transformers to a single value
def reverse_transformers(value, transformers):
for transformer in transformers:
if transformer == 'base64':
if isinstance(value, str):
value = base64.b64decode(value).decode('utf-8')
elif transformer == 'sha256':
mylog('none', [f'[reverseTransformers] sha256 is irreversible'])
return value
# Check if the value is a list
if isinstance(val, list):
return [reverse_transformers(item, transformers) for item in val]
else:
return reverse_transformers(val, transformers)
#------------------------------------------------------------------------------- #-------------------------------------------------------------------------------
# Generate a WHERE condition for SQLite based on a list of values. # Generate a WHERE condition for SQLite based on a list of values.
@@ -680,41 +695,24 @@ def cleanDeviceName(str, match_IP):
if str.endswith('.'): if str.endswith('.'):
str = str[:-1] str = str[:-1]
# done # done
mylog('debug', ["[Name cleanup] cleanDeviceName = " + str]) mylog('debug', ["[Name cleanup] cleanDeviceName = " + str])
return str return str
################################ # Applying cleanup REGEXEs
#
# OLD cleanDeviceName
mylog('debug', ["[Name cleanup] Using old cleanDeviceName(" + str + ")"]) mylog('debug', ["[Name cleanup] Using old cleanDeviceName(" + str + ")"])
# # alternative str.split('.')[0]
# str = str.replace("._airplay", "")
# str = str.replace("._tcp", "")
# str = str.replace(".localdomain", "")
# str = str.replace(".local", "")
# str = str.replace("._esphomelib", "")
# str = str.replace("._googlecast", "")
# str = str.replace(".lan", "")
# str = str.replace(".home", "")
# str = re.sub(r'-[a-fA-F0-9]{32}', '', str) # removing last part of e.g. Nest-Audio-ff77ff77ff77ff77ff77ff77ff77ff77
# str = re.sub(r'#.*', '', str) # Remove everything after '#' including the '#'
# # remove trailing dots
# if str.endswith('.'):
# str = str[:-1]
regexes = get_setting_value('NEWDEV_NAME_CLEANUP_REGEX') regexes = get_setting_value('NEWDEV_NAME_CLEANUP_REGEX')
for rgx in regexes: for rgx in regexes:
mylog('debug', ["[cleanDeviceName] applying regex : " + rgx]) mylog('debug', ["[cleanDeviceName] applying regex : " + rgx])
mylog('debug', ["[cleanDeviceName] name before regex : " + str]) mylog('debug', ["[cleanDeviceName] name before regex : " + str])
str = re.sub(rgx, "", str) str = re.sub(rgx, "", str)
mylog('debug', ["[cleanDeviceName] name after regex : " + str]) mylog('debug', ["[cleanDeviceName] name after regex : " + str])
# removing any trailing dots
str = re.sub(r'\b\.(\s)', r'\1', str)
mylog('debug', ["[cleanDeviceName] output: " + str]) mylog('debug', ["[cleanDeviceName] output: " + str])
return str return str