#!/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