#!/usr/bin/env python3 # -*- coding: utf-8 -*- from typing import Self, Any from os.path import exists as path_exists, dirname as directory_name, abspath as absolute_path, isfile as is_file from Interfaces.Application.NucelarMonitorInterface import NucelarMonitorInterface from Utils.Utils import Utils from Utils.Patterns import RE class FilesDriver: ROOT:str = directory_name(absolute_path(__file__)) SLASH:str = "/" if "/" in ROOT else "\\\\" def __init__(self:Self, nucelar_monitor:NucelarMonitorInterface) -> None: self.nucelar_monitor:NucelarMonitorInterface = nucelar_monitor self.__root_paths:list[str] = ["", self.ROOT] for _ in range(2): self.__root_paths.append(RE.LAST_DIRECTORY.sub(r'\1', self.__root_paths[-1])) @classmethod def fix_path(cls:type[Self], path:str) -> str: return RE.SLASHES.sub(cls.SLASH, path) def get_absolute_path(self:Self, path:str) -> str|None: root:str absolute:str for root in self.__root_paths: absolute = self.fix_path((root + '/' if root else "") + path) if path_exists(absolute): return absolute return None def load_file(self:Self, path:str, mode:str = "r") -> str|bytes|None: absolute_path:str = self.get_absolute_path(path) if absolute_path and is_file(absolute_path): with open(absolute_path, mode) as file: return file.read() return None def load_json(self:Self, data:str|dict[str, Any|None]|list[Any|None], only_dictionaries:bool = True) -> list[dict[str, Any|None]|list[Any|None]]: results:list[dict[str, Any|None]|list[Any|None]] = [] if isinstance(data, str): json:list[Any|None]|dict[str, Any|None]|None try: json = Utils.json_decode(data) except Exception as exception: self.nucelar_monitor.exception(exception, "load_json_exception", { "data" : data, "length" : len(data) }) if json: results.extend(self.load_json(json, only_dictionaries)) try: results.extend(self.load_json(Utils.json_decode(self.load_file(data)), only_dictionaries)) except Exception as exception: self.nucelar_monitor.exception(exception, "load_json_by_file_exception", { "path" : data }) elif isinstance(data, dict): results.append(data) elif isinstance(data, (list, tuple)): if only_dictionaries: item:Any|None for item in data: results.extend(self.load_json(item, only_dictionaries)) else: results.extend(data) return results