106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from typing import IO, Callable, Self, Optional, Any
|
|
from os import stat_result as StatResult, listdir as list_directory, remove as remove_path, stat as get_stat
|
|
from os.path import exists as path_exists
|
|
from stat import S_ISDIR as is_directory, S_ISREG as is_file
|
|
from Interfaces.Application.CXCVInterface import CXCVInterface
|
|
from Interfaces.FilesInterface import FilesInterface
|
|
from Managers.EventsManager import EventsManager
|
|
from Models.HashModel import HashModel
|
|
from Utils.Utils import Utils
|
|
|
|
class LocalFileDriver(FilesInterface):
|
|
|
|
def __init__(self:Self,
|
|
cxcv:Any,
|
|
path:str,
|
|
inputs:Optional[dict[str, Any|None]|list[Any|None]|tuple[Any|None, ...]] = None
|
|
) -> None:
|
|
|
|
self.cxcv:CXCVInterface = cxcv
|
|
self.path:str = path
|
|
self.chunk_size:int = self.cxcv.settings.get(("chunk_size", "default_chunk_size"), inputs, 4096)
|
|
self.__file:IO[Any]|None = None
|
|
self.mode:str = self.cxcv.settings.get(("mode", "file_mode", "default_file_mode"), inputs, "rb")
|
|
self.__closed:bool = False
|
|
self.on_load_chunk:EventsManager = EventsManager()
|
|
callback:Callable[[bytes], bool]|None = Utils.get_value("on_load_chunk", inputs)
|
|
self.__autoclose:bool = self.cxcv.settings.get("auto_close", inputs, True)
|
|
self.reusable:bool = self.cxcv.settings.get("reusable", inputs, False)
|
|
self.exists:bool = path_exists(self.path)
|
|
self.is_file:bool = False
|
|
self.is_directory:bool = False
|
|
self.size:int = 0
|
|
self.hash:HashModel = None
|
|
stat:StatResult
|
|
|
|
try:
|
|
stat = get_stat(self.path)
|
|
self.exists = True
|
|
self.is_file = is_file(stat.st_mode)
|
|
self.is_directory = is_directory(stat.st_mode)
|
|
self.size = stat.st_size
|
|
except Exception as exception:
|
|
pass
|
|
|
|
callback and self.on_load_chunk.add(callback)
|
|
|
|
self.cxcv.settings.get("auto_open", inputs, True) and self.open()
|
|
|
|
def open(self:Self, new_path:Optional[str] = None) -> Self:
|
|
|
|
if new_path or self.reusable:
|
|
if new_path:
|
|
self.path = new_path
|
|
if self.__file and not self.__closed:
|
|
self.__file.close()
|
|
self.__closed = False
|
|
elif self.__file is not None:
|
|
return self
|
|
|
|
self.__file = open(self.path, mode = self.mode)
|
|
|
|
return self
|
|
|
|
def close(self:Self) -> Self:
|
|
|
|
if not self.__closed:
|
|
self.__closed = True
|
|
self.hash = None
|
|
self.is_directory = False
|
|
self.is_file = False
|
|
self.exists = False
|
|
self.size = 0
|
|
self.__file and self.__file.close()
|
|
|
|
return self
|
|
|
|
def load(self:Self, all:bool = False) -> bytes:
|
|
|
|
data:bytes = self.__file.read(self.size if all else self.chunk_size)
|
|
|
|
if data:
|
|
self.on_load_chunk.execute(data)
|
|
elif self.__autoclose:
|
|
self.close()
|
|
|
|
def save(self:Self, chunk:Optional[bytes|bool] = None) -> None:
|
|
if chunk:
|
|
self.__file.write(chunk)
|
|
elif self.__autoclose:
|
|
self.close()
|
|
|
|
def list(self:Self) -> list[str]:
|
|
return list_directory(self.path) if self.is_directory else []
|
|
|
|
def remove(self:Self) -> None:
|
|
remove_path(self.path)
|
|
|
|
def reset(self:Self) -> Self:
|
|
|
|
self.close()
|
|
self.open()
|
|
|
|
return self |