Made many refactoring changes, fixed bugs and improved code

[__init__.py]
    Added a library check to make sure its only imported
    Removal of config_data to Config.py (Created it as well)
[_dev.py]
    [color_print()] Added new color cyan and new option to not print but input (also returns values)
    Changed inputs to color_print(is_input=True)
    Changed some messages to make them look better
    Changed how the logic of getting the file changes work, now its WAY more accurate and better and faster
[Config.py]
    Added new file, this is for config data gain, [Ref: __init__.py]
[Flag.py, Logicytics.py, packet_sniffer.py]
    Changed config to the CONFIG global var from __init__.py
[Get.py]
    Changed list_of_files() logic to now only exclude files starting with '_' and any extensions and files passed down (rather than hardcoded files like Logicytics.py)
    Refactored the class and function to be more efficient as well as fixed some bugs
[Logicytics.py]
    Added exclusion to get files to accommodate the new changes [Ref: Get.py]
    Mitigated potential bugs in __threaded method with ThreadPoolExecutor, added a try-except
    Mitigated potential bugs in backup if the path doesn't exist
    Made the subprocess opening for execute_new_window more safe by mitigating a bug if python doesn't exist in PATH, now it takes python PATH from sys.executable
[packet_sniffer.py]
    Improved error logging messages
    Made method __get_port_info() safer by making the port_type optional defaulting to 'sport', also added validation to only allow 'sport' and 'dport'
    Added a retry timer in the code, thus script maxes execution at 30sec (based on max_retry_time)
    Made Sniff() init once only.
[vulnscan.py]
    Improved Time bottleneck by only reading readable files that aren't part of UNREADABLE_EXTENSIONS list, thus now it will read/parse more than scan, which improves time and reduces issues
    Renamed some methods to be protected members
    Improved _is_sensitive by using sparse matrices
    Fixed bug where if result is empty, it crashes due to append of a variable that doesn't exist
    Remade the ThreadPoolExecutor logic for getting the files by improving its speed and efficiency

Signed-off-by: Shahm Najeeb <[email protected]>

Author: DefinetlyNotAI

.idea/csv-editor.xml

@@ -1,24 +1,22 @@
 from __future__ import annotations
 
-import configparser
 import os
 import shutil
 import subprocess
+import sys
 import zipfile
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime
 
 import psutil
 from prettytable import PrettyTable
 
-from logicytics import Log, Execute, Check, Get, FileManagement, Flag, DEBUG, DELETE_LOGS
+from logicytics import Log, Execute, Check, Get, FileManagement, Flag, DEBUG, DELETE_LOGS, CONFIG
 
 # Initialization
 log = Log({"log_level": DEBUG, "delete_log": DELETE_LOGS})
 ACTION, SUB_ACTION = None, None
-config = configparser.ConfigParser()
-config.read("config.ini")
-MAX_WORKERS = config.getint("Settings", "max_workers", fallback=min(32, (os.cpu_count() or 1) + 4))
+MAX_WORKERS = CONFIG.getint("Settings", "max_workers", fallback=min(32, (os.cpu_count() or 1) + 4))
 log.debug(f"MAX_WORKERS: {MAX_WORKERS}")
 
 
@@ -67,7 +65,8 @@ def __generate_execution_list(self) -> list[str]:
             - Logs the final execution list for debugging purposes
             - Warns users about potential long execution times for certain actions
         """
-        execution_list = Get.list_of_files(".", extensions=(".py", ".exe", ".ps1", ".bat"))
+        execution_list = Get.list_of_files(".", only_extensions=(".py", ".exe", ".ps1", ".bat"),
+                                           exclude_files=["Logicytics.py"])
         files_to_remove = {
             "sensitive_data_miner.py",
             "dir_list.py",
@@ -100,9 +99,8 @@ def __generate_execution_list(self) -> list[str]:
 
         elif ACTION == "modded":
             # Add all files in MODS to execution list
-            execution_list = Get.list_of_files("../MODS",
-                                               extensions=(".py", ".exe", ".ps1", ".bat"),
-                                               append_file_list=execution_list)
+            execution_list = Get.list_of_files("../MODS", only_extensions=(".py", ".exe", ".ps1", ".bat"),
+                                               append_file_list=execution_list, exclude_files=["Logicytics.py"])
 
         elif ACTION == "depth":
             log.warning(
@@ -169,11 +167,14 @@ def __threaded(self):
 
             for future in as_completed(futures):
                 script = futures[future]
-                result, error = future.result()
-                if error:
-                    log.error(f"Failed to execute {script}")
-                else:
-                    log.debug(f"Completed {script}")
+                try:
+                    result, error = future.result()
+                    if error:
+                        log.error(f"Failed to execute {script}: {error}")
+                    else:
+                        log.debug(f"Completed {script}")
+                except Exception as e:
+                    log.error(f"Thread crashed while executing {script}: {e}")
 
     def __default(self):
         """Executes scripts sequentially."""
@@ -246,6 +247,10 @@ def backup(directory: str, name: str):
             Returns:
                 None
             """
+        if not os.path.exists(directory):
+            log.critical(f"Directory {directory} does not exist!")
+            return
+
         # Check if backup exists, delete it if so
         if os.path.exists(f"../ACCESS/BACKUP/{name}.zip"):
             os.remove(f"../ACCESS/BACKUP/{name}.zip")
@@ -308,7 +313,7 @@ def execute_new_window(file_path: str):
         """
         sr_current_dir = os.path.dirname(os.path.abspath(__file__))
         sr_script_path = os.path.join(sr_current_dir, file_path)
-        sr_process = subprocess.Popen(["cmd.exe", "/c", "start", "python", sr_script_path])
+        sr_process = subprocess.Popen(["cmd.exe", "/c", "start", sys.executable, sr_script_path])
         sr_process.wait()
         exit(0)
 
@@ -507,6 +512,7 @@ def Logicytics():
     except KeyboardInterrupt:
         log.warning("Shutting down Logicytics utility with force causes many files to remain where they shouldn't")
         log.warning("Please don't force shut Logicytics again - As we don't have a cleanup function yet.")
+        # TODO v3.4.2 -> Cleanup function
         exit(0)
 else:
     log.error("This script cannot be imported!")

CODE/Logicytics.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 import re
 import subprocess
 
@@ -8,16 +9,20 @@
 from logicytics import log, Get, FileManagement, CURRENT_FILES, VERSION
 
 
-def color_print(text, color="reset"):
+def color_print(text, color="reset", is_input=False) -> None | str:
     colors = {
         "reset": "\033[0m",
         "red": "\033[31m",
         "green": "\033[32m",
         "yellow": "\033[33m",
+        "cyan": "\033[36m",
     }
 
     color_code = colors.get(color.lower(), colors["reset"])
-    print(f"{color_code}{text}{colors['reset']}")
+    if is_input:
+        return input(f"{color_code}{text}{colors['reset']}")
+    else:
+        print(f"{color_code}{text}{colors['reset']}")
 
 
 def _update_ini_file(filename: str, new_data: list | str, key: str) -> None:
@@ -31,14 +36,16 @@ def _update_ini_file(filename: str, new_data: list | str, key: str) -> None:
         None
     """
     try:
-        config = configobj.ConfigObj(filename, encoding='utf-8', write_empty_values=True)
+        config = configobj.ConfigObj(filename, encoding="utf-8", write_empty_values=True)
+
         if key == "files":
             config["System Settings"][key] = ", ".join(new_data)
         elif key == "version":
             config["System Settings"][key] = new_data
         else:
             color_print(f"[!] Invalid key: {key}", "yellow")
             return
+
         config.write()
     except FileNotFoundError:
         color_print("[x] INI file not found", "red")
@@ -69,7 +76,7 @@ def _prompt_user(question: str, file_to_open: str = None, special: bool = False)
         - Provides optional file opening and reminder messaging
     """
     try:
-        answer = input(f"\033[36m[?] {question} (y)es or (n)o:- \033[0m")
+        answer = color_print(f"[?] {question} (y)es or (n)o:- ", "cyan", is_input=True)
         if not (answer.lower() == "yes" or answer.lower() == "y"):
             if file_to_open:
                 subprocess.run(["start", file_to_open], shell=True)
@@ -91,11 +98,11 @@ def _perform_checks() -> bool:
         bool: True if all checks are confirmed by the user, False otherwise.
     """
     checks = [
-        ("Have you read the required contributing guidelines?", "../CONTRIBUTING.md"),
-        ("Have you made files you don't want to be run start with '_'?", "."),
-        ("Have you added the file to CODE dir?", "."),
-        ("Have you added docstrings and comments?", "../CONTRIBUTING.md"),
-        ("Is each file containing around 1 main feature?", "../CONTRIBUTING.md"),
+        ("[-] Have you read the required contributing guidelines?", "..\\CONTRIBUTING.md"),
+        ("[-] Have you made files you don't want to be run start with '_'?", "."),
+        ("[-] Have you added the file to CODE dir?", "."),
+        ("[-] Have you added docstrings and comments?", "..\\CONTRIBUTING.md"),
+        ("[-] Is each file containing around 1 main feature?", "..\\CONTRIBUTING.md"),
     ]
 
     for question, file_to_open in checks:
@@ -109,48 +116,49 @@ def _handle_file_operations() -> None:
     Handles file operations and logging for added, removed, and normal files.
     """
     EXCLUDE_FILES = ["logicytics\\User_History.json.gz", "logicytics\\User_History.json"]
-    files = Get.list_of_files(".", True, exclude_files=EXCLUDE_FILES)
+    files = Get.list_of_files(".", exclude_files=EXCLUDE_FILES)
     added_files, removed_files, normal_files = [], [], []
     clean_files_list = [file.replace('"', '') for file in CURRENT_FILES]
 
-    for f in files:
-        clean_f = f.replace('"', '')
-        if clean_f in clean_files_list and clean_f not in EXCLUDE_FILES:
-            normal_files.append(clean_f)
-        elif clean_f not in EXCLUDE_FILES:
-            added_files.append(clean_f)
-
-    for f in clean_files_list:
-        clean_f = f.replace('"', '')
-        if clean_f not in files and clean_f not in EXCLUDE_FILES:
-            removed_files.append(clean_f)
-
-    for file in added_files:
-        color_print(f"+ {file}", "green")
-    for file in removed_files:
-        color_print(f"- {file}", "red")
-    for file in normal_files:
-        print(f"* {file}")
-
-    if not _prompt_user("Does the list above include your added files?"):
+    files_set = set(os.path.abspath(f) for f in files)
+    clean_files_set = set(os.path.abspath(f) for f in clean_files_list)
+
+    for file in files_set:
+        if file in clean_files_set and file not in EXCLUDE_FILES:
+            normal_files.append(file)
+        elif file not in clean_files_set and file not in EXCLUDE_FILES:
+            added_files.append(file)
+
+    for file in clean_files_set:
+        if file not in files_set and file not in EXCLUDE_FILES:
+            removed_files.append(file)
+
+    print("\n".join([f"\033[92m+ {file}\033[0m" for file in added_files]))  # Green +
+    print("\n".join([f"\033[91m- {file}\033[0m" for file in removed_files]))  # Red -
+    print("\n".join([f"* {file}" for file in normal_files]))
+
+    if not _prompt_user("[-] Does the list above include your added files?"):
         color_print("[x] Something went wrong! Please contact support.", "red")
         return
 
     max_attempts = 10
     attempts = 0
     _update_ini_file("config.ini", files, "files")
+
     while True:
-        version = input(f"\033[36m[?] Enter the new version of the project (Old version is {VERSION}): \033[0m")
+        version = color_print(f"[?] Enter the new version of the project (Old version is {VERSION}): ", "cyan",
+                              is_input=True)
         if attempts >= max_attempts:
             color_print("[x] Maximum attempts reached. Please run the script again.", "red")
             exit()
-        if re.match(r'^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)$', version):
+        if re.match(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)$", version):
             _update_ini_file("config.ini", version, "version")
             break
         else:
             color_print("[!] Please enter a valid version number (e.g., 1.2.3)", "yellow")
             attempts += 1
             color_print(f"[!] {max_attempts - attempts} attempts remaining", "yellow")
+
     color_print("\n[-] Great Job! Please tick the box in the GitHub PR request for completing steps in --dev", "green")
 
 
@@ -188,4 +196,4 @@ def dev_checks() -> None:
 if __name__ == "__main__":
     dev_checks()
     # Wait for the user to press Enter to exit the program
-    input("\nPress Enter to exit the program... ")
+    input("\n[*] Press Enter to exit the program... ")

CODE/_dev.py

@@ -0,0 +1,50 @@
+import configparser
+import os
+
+
+def __config_data() -> tuple[str, str, list[str], bool, str]:
+    """
+    Retrieves configuration data from the 'config.ini' file.
+
+    If the configuration file is not found in any of these locations,
+    the program exits with an error message.
+
+    Returns:
+        tuple[str, str, list[str], bool]: A tuple containing:
+            - Log level (str): Either "DEBUG" or "INFO"
+            - Version (str): System version from configuration
+            - Files (list[str]): List of files specified in configuration
+            - Delete old logs (bool): Flag indicating whether to delete old log files
+            - CONFIG itself
+
+    Raises:
+        SystemExit: If the 'config.ini' file cannot be found in any of the attempted locations
+    """
+
+    def _config_path() -> str:
+        configs_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.ini")
+
+        if os.path.exists(configs_path):
+            return configs_path
+        else:
+            print("The config.ini file is not found in the expected location.")
+            exit(1)
+
+    config = configparser.ConfigParser()
+    path = _config_path()
+    config.read(path)
+
+    log_using_debug = config.getboolean("Settings", "log_using_debug")
+    delete_old_logs = config.getboolean("Settings", "delete_old_logs")
+    version = config.get("System Settings", "version")
+    files = config.get("System Settings", "files").split(", ")
+
+    log_using_debug = "DEBUG" if log_using_debug else "INFO"
+
+    return log_using_debug, version, files, delete_old_logs, config
+
+
+# Check if the script is being run directly, if not, set up the library
+if __name__ == '__main__':
+    exit("This is a library, Please import rather than directly run.")
+DEBUG, VERSION, CURRENT_FILES, DELETE_LOGS, CONFIG = __config_data()