85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
#!/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 |