191 lines
6.2 KiB
Python
191 lines
6.2 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from typing import Self, Any, Callable
|
|
from Interfaces.Application.CXCVInterface import CXCVInterface
|
|
from Utils.Patterns import RE
|
|
from Utils.Utils import Utils
|
|
from os.path import exists as path_exists, isfile as is_file, isdir as is_directory
|
|
from os import mkdir as make_directory, remove as remove_file
|
|
from os import listdir as list_directory, stat as get_stat, stat_result as StatResult
|
|
from shutil import rmtree as remove_directory
|
|
|
|
class FilesDriver:
|
|
|
|
def __init__(self:Self, cxcv:CXCVInterface, root_path:str) -> None:
|
|
self.cxcv:CXCVInterface = cxcv
|
|
|
|
self.root_paths:tuple[str, ...] = ("", root_path)
|
|
self.slash:str = "/" if "/" in root_path else "\\"
|
|
|
|
for _ in range(2):
|
|
self.root_paths += (RE.LAST_PATH_ITEM.sub(r'', self.root_paths[-1]),)
|
|
|
|
def fix_path(self:Self, path:str) -> str:
|
|
return RE.SLASHES.sub(self.slash, path)
|
|
|
|
def exists(self:Self, path:str, analyze_roots:bool = True) -> bool:
|
|
if analyze_roots:
|
|
return self.get_absolute_path(path) is not None
|
|
return path_exists(path)
|
|
|
|
def is_file(self:Self, path:str, analyze_roots:bool = True) -> bool:
|
|
if analyze_roots and (path := self.get_absolute_path(path)) is None:
|
|
return False
|
|
return is_file(path)
|
|
|
|
def is_directory(self:Self, path:str, analyze_roots:bool = True) -> bool:
|
|
if analyze_roots and (path := self.get_absolute_path(path)) is None:
|
|
return False
|
|
return is_directory(path)
|
|
|
|
def get_absolute_path(self:Self, path:str) -> str|None:
|
|
|
|
root:str
|
|
|
|
for root in self.root_paths:
|
|
absolute_path:str = self.fix_path((root + self.slash if root else "") + path)
|
|
if self.exists(absolute_path, False):
|
|
return absolute_path
|
|
return None
|
|
|
|
def load(self:Self, path:str, mode:str = "r") -> str|bytes|None:
|
|
try:
|
|
|
|
absolute_path:str = self.get_absolute_path(path)
|
|
|
|
if absolute_path is not None and is_file(absolute_path):
|
|
with open(absolute_path, mode) as file:
|
|
return file.read()
|
|
|
|
except Exception as exception:
|
|
pass
|
|
return None
|
|
|
|
def load_by_chunks(self:Self,
|
|
path:str,
|
|
callback:Callable[[bytes], None],
|
|
chunk_size:int = 4096
|
|
) -> None:
|
|
try:
|
|
|
|
absolute_path:str = self.get_absolute_path(path)
|
|
|
|
if absolute_path is not None and is_file(absolute_path):
|
|
with open(absolute_path, "rb") as file:
|
|
while True:
|
|
chunk:bytes = file.read(chunk_size)
|
|
if not chunk:
|
|
break
|
|
callback(chunk)
|
|
|
|
except Exception as exception:
|
|
pass
|
|
|
|
def save(self:Self, path:str, data:str|bytes, mode:str = "w") -> bool:
|
|
try:
|
|
self.prepare_path(self.get_directory_path(path))
|
|
with open(self.fix_path(path), mode) as file:
|
|
file.write(data)
|
|
return True
|
|
except Exception as exception:
|
|
self.cxcv.exception(exception, "files_driver_save_exception", {
|
|
"path" : path,
|
|
"mode" : mode,
|
|
"length" : len(data) if data is not None else 0
|
|
})
|
|
return False
|
|
|
|
def remove(self:Self, path:str) -> bool:
|
|
if self.exists(path := self.fix_path(path), False):
|
|
try:
|
|
if self.is_file(path, False):
|
|
remove_file(path)
|
|
elif self.is_directory(path, False):
|
|
remove_directory(path)
|
|
return True
|
|
except Exception as exception:
|
|
pass
|
|
return False
|
|
|
|
def load_json(self:Self,
|
|
inputs:Any|None,
|
|
only_dictionaries:bool = True
|
|
) -> tuple[dict[str, Any|None]|tuple[Any|None, ...]|list[Any|None], ...]:
|
|
|
|
items:tuple[dict[str, Any|None]|tuple[Any|None, ...]|list[Any|None], ...] = ()
|
|
|
|
if isinstance(inputs, dict):
|
|
items += (inputs,)
|
|
elif isinstance(inputs, list) or isinstance(inputs, tuple):
|
|
if only_dictionaries:
|
|
|
|
subinputs:Any|None
|
|
|
|
for subinputs in inputs:
|
|
items += self.load_json(subinputs, only_dictionaries)
|
|
|
|
else:
|
|
items += (inputs,)
|
|
elif isinstance(inputs, str):
|
|
|
|
json:dict[str, Any|None]|tuple[Any|None, ...]|list[Any|None]|None = Utils.json_decode(inputs)
|
|
|
|
items += (
|
|
self.load_json(Utils.json_decode(self.load(inputs, "r")), only_dictionaries) if json is None else
|
|
self.load_json(json, only_dictionaries))
|
|
|
|
return items
|
|
|
|
def get_directory_path(self:Self, path:str) -> str:
|
|
return RE.LAST_PATH_ITEM.sub(r'', self.fix_path(path))
|
|
|
|
def prepare_path(self:Self, path:str) -> bool:
|
|
try:
|
|
|
|
parts:list[str] = RE.SLASHES.split(path)
|
|
directory:str = "/" if self.slash == "/" else ""
|
|
part:str
|
|
|
|
for part in parts:
|
|
if part:
|
|
directory += parts[0] + self.slash
|
|
self.exists(directory, False) or make_directory(directory)
|
|
parts = parts[1:]
|
|
|
|
return True
|
|
except Exception as exception:
|
|
print(exception)
|
|
return False
|
|
|
|
def list(self:Self, path:str) -> list[str]:
|
|
|
|
items:list[str] = []
|
|
|
|
try:
|
|
|
|
absolute_path:str = self.get_absolute_path(path)
|
|
|
|
if absolute_path is not None and is_directory(absolute_path):
|
|
items = list_directory(absolute_path)
|
|
|
|
except Exception as exception:
|
|
pass
|
|
|
|
return items
|
|
|
|
def get_stat(self:Self, path:str) -> StatResult|None:
|
|
|
|
stat:StatResult|None = None
|
|
|
|
try:
|
|
|
|
absolute_path:str = self.get_absolute_path(path)
|
|
|
|
if absolute_path is not None and self.exists(absolute_path, False):
|
|
stat = get_stat(absolute_path)
|
|
|
|
except Exception as exception:
|
|
pass
|
|
|
|
return stat |