168 lines
5.8 KiB
Python
168 lines
5.8 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from typing import Any, Self, Sequence
|
|
from os.path import dirname as directory_name, abspath as absolute_path, exists as path_exists, isfile as is_file, isdir as is_directory
|
|
from io import FileIO
|
|
from json import loads as json_loads
|
|
from Interfaces.Application.AnPInterface import AnPInterface
|
|
from Utils.Options import Options
|
|
from Utils.Validate import Validate
|
|
from Utils.Patterns import RE
|
|
|
|
class FilesDriver:
|
|
|
|
ROOT:str = directory_name(absolute_path(__file__))
|
|
SLASH:str = "/" if "/" in ROOT else "\\"
|
|
|
|
GET_ABSOLUTE_PATH_OPTIONS:Options = Options(Options.ALLOW_CHECKS + Options.SHOW_ERRORS)
|
|
LOAD_OPTIONS:Options = Options(Options.ALLOW_CHECKS + Options.SHOW_ERRORS)
|
|
LOAD_JSON_OPTIONS:Options = Options(Options.ALLOW_CHECKS + Options.SHOW_ERRORS)
|
|
|
|
def __init__(self:Self, anp:AnPInterface) -> None:
|
|
self.anp:AnPInterface = anp
|
|
|
|
self.__roots:list[str] = ["", self.ROOT]
|
|
|
|
for _ in range(2):
|
|
self.__roots += [RE.PARENT_DIRECTORY.sub(r'\1', self.__roots[-1])]
|
|
|
|
@classmethod
|
|
def fix_path(cls:type[Self], path:str) -> str:
|
|
|
|
path = RE.SLASHES.sub(cls.SLASH, path)
|
|
|
|
return path if path[0] != "\\" else "\\" + path
|
|
|
|
def get_absolute_path(self:Self, path:str, custom_options:int = 0) -> tuple[str|None, int]:
|
|
""" 5 bits. """
|
|
|
|
options:Options = Options(custom_options, self.GET_ABSOLUTE_PATH_OPTIONS)
|
|
error:int = (
|
|
(Validate.string(path, Validate.TRIM | Validate.EMPTY) << 1) |
|
|
0) if options.allow_checks else 0
|
|
absolute:str|None = None
|
|
|
|
if not error:
|
|
for root in self.__roots:
|
|
try:
|
|
|
|
absolute_path:str = self.fix_path((root + self.SLASH if root else "") + path)
|
|
|
|
if path_exists(absolute_path):
|
|
absolute = absolute_path
|
|
break
|
|
|
|
except Exception as exception:
|
|
error |= 1 << 0
|
|
|
|
if absolute is None:
|
|
error |= 1 << 4
|
|
|
|
return absolute, error
|
|
|
|
def load(self:Self, path:str, mode:str = "r", custom_options:int = 0) -> tuple[str|bytes|None, int]:
|
|
""" 10 bits. """
|
|
|
|
options:Options = Options(custom_options, self.LOAD_OPTIONS)
|
|
error:int = (
|
|
# (Validate.string(path, Validate.TRIM | Validate.EMPTY) << 1) |
|
|
((
|
|
Validate.string(mode, Validate.TRIM | Validate.EMPTY) or
|
|
(0 if mode in ("r", "rb") else 8)
|
|
) << 6) |
|
|
0) if options.allow_checks else 0
|
|
results:str|bytes|None = None
|
|
absolute:str|None
|
|
absolute_error:int
|
|
|
|
absolute, absolute_error = self.get_absolute_path(path, custom_options)
|
|
if options.allow_checks:
|
|
error |= absolute_error << 1
|
|
|
|
if not error:
|
|
try:
|
|
|
|
file:FileIO
|
|
|
|
with open(absolute, mode) as file:
|
|
results = file.read()
|
|
|
|
except Exception as exception:
|
|
error |= 1 << 0
|
|
|
|
return results, error
|
|
|
|
def load_json(self:Self, data:str|Sequence[Any|None]|dict[str, Any|None], custom_options:int = 0) -> tuple[Sequence[Any|None]|dict[str, Any|None]|None, int]:
|
|
""" 10 bits. """
|
|
|
|
options:Options = Options(custom_options, self.LOAD_JSON_OPTIONS)
|
|
error:int = (
|
|
(Validate.null_or_undefined(data) << 1) |
|
|
0) if options.allow_checks else 0
|
|
results:Sequence[Any|None]|dict[str, Any|None]|None = None
|
|
|
|
if isinstance(data, str):
|
|
|
|
if not options.allow_checks or not (
|
|
error := error | Validate.string(data, Validate.FROM_TYPE | Validate.TRIM | Validate.EMPTY) << 1
|
|
):
|
|
|
|
data = data.strip()
|
|
|
|
if data[0] + data[-1] in ("{}", "[]"):
|
|
try:
|
|
results = json_loads(data)
|
|
if options.allow_checks and results == None:
|
|
error |= 1 << 6
|
|
except Exception as exception:
|
|
error |= 1 << 0
|
|
else:
|
|
|
|
load_error:int
|
|
|
|
data, load_error = self.load(data, "r", custom_options)
|
|
|
|
if not options.allow_checks or (
|
|
error := error | (load_error << 7)
|
|
) or not (
|
|
error := error | Validate.string(data, Validate.FROM_TYPE | Validate.TRIM | Validate.EMPTY) << 1
|
|
):
|
|
|
|
data = data.strip()
|
|
|
|
if data[0] + data[-1] in ("{}", "[]"):
|
|
try:
|
|
results = json_loads(data)
|
|
if options.allow_checks and results == None:
|
|
error |= 1 << 6
|
|
except Exception as exception:
|
|
error |= 1 << 0
|
|
else:
|
|
error |= 1 << 18
|
|
|
|
if not error:
|
|
try:
|
|
results = json_loads(data)
|
|
if options.allow_checks and results == None:
|
|
error |= 1 << 6
|
|
except Exception as exception:
|
|
error |= 1 << 0
|
|
elif isinstance(data, (list, dict)):
|
|
results = data
|
|
else:
|
|
error |= 1 << 3
|
|
|
|
return results, error
|
|
|
|
def save(self:Self, path:str, content:str|bytes, mode:str = "w", custom_options:int = 0) -> int:
|
|
pass
|
|
|
|
def delete(self:Self, path:str, custom_options:int = 0) -> int:
|
|
pass
|
|
|
|
def is_file(self:Self, path:str, custom_options:int = 0) -> int:
|
|
pass
|
|
|
|
def is_directory(self:Self, path:str, custom_options:int = 0) -> int:
|
|
pass |