diff --git a/Python/AnP/Application/AnP.py b/Python/AnP/Application/AnP.py index a66e993..6832c33 100644 --- a/Python/AnP/Application/AnP.py +++ b/Python/AnP/Application/AnP.py @@ -25,6 +25,7 @@ from AnP.Managers.HTTPServersManager import HTTPServersManager from AnP.Managers.PseudoLoRAsManager import PseudoLoRAsManager from AnP.Managers.AIInterpretersManager import AIInterpretersManager from AnP.Managers.BrowsersManager import BrowsersManager +from AnP.Managers.DirectoriesClonesManager import DirectoriesClonesManager from AnP.Utils.Common import Common from AnP.Utils.Patterns import RE @@ -64,6 +65,7 @@ class AnP: self.pseudoloras:PseudoLoRAsManager = PseudoLoRAsManager(self) self.ai_interpreters:AIInterpretersManager = AIInterpretersManager(self) self.browsers:BrowsersManager = BrowsersManager(self) + self.directories_clones:DirectoriesClonesManager = DirectoriesClonesManager(self) def update(self:Self) -> None: self.settings.update() @@ -84,6 +86,7 @@ class AnP: self.pseudoloras.update() self.ai_interpreters.update() self.browsers.update() + self.directories_clones.update() def reset(self:Self) -> None: self.settings.reset() @@ -102,6 +105,7 @@ class AnP: self.http_servers.reset() self.ai_interpreters.reset() self.browsers.reset() + self.directories_clones.reset() def close(self:Self) -> None: self.__working = False @@ -110,6 +114,7 @@ class AnP: self.web_socket_servers.close() self.http_servers.close() self.browsers.close() + self.directories_clones.close() def working(self:Self) -> bool: return self.__working diff --git a/Python/AnP/Interfaces/Application/AnPInterface.py b/Python/AnP/Interfaces/Application/AnPInterface.py index ba41fa6..fa75fcb 100644 --- a/Python/AnP/Interfaces/Application/AnPInterface.py +++ b/Python/AnP/Interfaces/Application/AnPInterface.py @@ -21,6 +21,7 @@ from AnP.Interfaces.Managers.HTTPServersManagerInterface import HTTPServersManag from AnP.Interfaces.Managers.PseudoLoRAsManagerInterface import PseudoLoRAsManagerInterface from AnP.Interfaces.Managers.AIInterpretersManagerInterface import AIInterpretersManagerInterface from AnP.Interfaces.Managers.BrowsersManagerInterface import BrowsersManagerInterface +from AnP.Interfaces.Managers.DirectoriesClonesManagerInterface import DirectoriesClonesManagerInterface class AnPInterface(ABC): @@ -43,6 +44,7 @@ class AnPInterface(ABC): self.pseudoloras:PseudoLoRAsManagerInterface = None self.ai_interpreters:AIInterpretersManagerInterface = None self.browsers:BrowsersManagerInterface = None + self.directories_clones:DirectoriesClonesManagerInterface = None @abstractmethod def update(self:Self) -> None:pass diff --git a/Python/AnP/Interfaces/Managers/DirectoriesClonesManagerInterface.py b/Python/AnP/Interfaces/Managers/DirectoriesClonesManagerInterface.py new file mode 100644 index 0000000..3bf7bf7 --- /dev/null +++ b/Python/AnP/Interfaces/Managers/DirectoriesClonesManagerInterface.py @@ -0,0 +1,28 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +from typing import Self, Any, Optional +from abc import ABC, abstractmethod + +class DirectoriesClonesManagerInterface(ABC): + + @abstractmethod + def update(self:Self) -> None:pass + + @abstractmethod + def reset(self:Self) -> None:pass + + @abstractmethod + def close(self:Self) -> None:pass + + @abstractmethod + def add(self:Self, inputs:Any|None, overwrite:bool = False) -> None:pass + + @abstractmethod + def remove(self:Self, keys:Optional[str] = None) -> None:pass + + @abstractmethod + def play(self:Self, keys:Optional[str] = None) -> None:pass + + @abstractmethod + def stop(self:Self, keys:Optional[str] = None) -> None:pass \ No newline at end of file diff --git a/Python/AnP/Interfaces/Models/DirectoryCloneModelInterface.py b/Python/AnP/Interfaces/Models/DirectoryCloneModelInterface.py new file mode 100644 index 0000000..6f6d9c2 --- /dev/null +++ b/Python/AnP/Interfaces/Models/DirectoryCloneModelInterface.py @@ -0,0 +1,16 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +from typing import Self +from abc import ABC, abstractmethod + +class DirectoryCloneModelInterface(ABC): + + @abstractmethod + def is_working(self:Self) -> bool:pass + + @abstractmethod + def start(self:Self) -> None:pass + + @abstractmethod + def stop(self:Self) -> None:pass \ No newline at end of file diff --git a/Python/AnP/Managers/DirectoriesClonesManager.py b/Python/AnP/Managers/DirectoriesClonesManager.py new file mode 100644 index 0000000..58c53b1 --- /dev/null +++ b/Python/AnP/Managers/DirectoriesClonesManager.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +from typing import Self, Any, Optional, Sequence +from AnP.Interfaces.Application.AnPInterface import AnPInterface +from AnP.Models.DirectoryCloneModel import DirectoryCloneModel +from AnP.Utils.Checks import Check + +class DirectoriesClonesManager: + + def __init__(self:Self, anp:AnPInterface) -> None: + self.anp:AnPInterface = anp + self.__clones:dict[str, DirectoryCloneModel] = {} + + def update(self:Self) -> None: + + key:str + + for key in ( + "default_directories_clones_files", "directories_clones_files", + "default_directories_clones", "directories_clones" + ): + self.add(self.anp.settings.get(key), True) + + def reset(self:Self) -> None: + self.remove() + self.update() + + def close(self:Self) -> None: + self.remove() + + def add(self:Self, inputs:Any|None, overwrite:bool = False) -> None: + + block:dict[str, dict[str, Any|None]|Sequence[Any|None]] + + for block in self.anp.files.load_json(inputs, True): + + key:str + subinputs:dict[str, Any|None]|Sequence[Any|None] + + for key, subinputs in block.items(): + if overwrite or key not in self.__clones: + if key in self.__clones: + self.__clones[key].stop() + self.__clones[key] = ( + subinputs if isinstance(subinputs, DirectoriesClonesManager) else + DirectoryCloneModel(self.anp, key, subinputs)) + + def remove(self:Self, keys:Optional[str] = None) -> None: + + key:str + + for key in ( + keys if Check.is_array(keys) else + [keys] if Check.is_string(keys) else + self.__clones.keys() if keys is None else + []): + self.__clones[key].stop() + del self.__clones[key] + + def play(self:Self, keys:Optional[str] = None) -> None: + + key:str + + for key in ( + keys if Check.is_array(keys) else + [keys] if Check.is_string(keys) else + self.__clones.keys() if keys is None else + []): + self.__clones[key].start() + + def stop(self:Self, keys:Optional[str] = None) -> None: + + key:str + + for key in ( + keys if Check.is_array(keys) else + [keys] if Check.is_string(keys) else + self.__clones.keys() if keys is None else + []): + self.__clones[key].stop() \ No newline at end of file diff --git a/Python/AnP/Models/DirectoryCloneModel.py b/Python/AnP/Models/DirectoryCloneModel.py new file mode 100644 index 0000000..fb4a188 --- /dev/null +++ b/Python/AnP/Models/DirectoryCloneModel.py @@ -0,0 +1,101 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +from typing import Self, Any, Sequence +from sys import path as system_paths +from threading import Thread +from AnP.Interfaces.Application.AnPInterface import AnPInterface +from AnP.Models.HashModel import HashModel +from AnP.Utils.Common import Common +from AnP.Utils.Patterns import RE + +class DirectoryCloneModel: + + def __init__(self:Self, anp:AnPInterface, key:str, inputs:dict[str, Any|None]|Sequence[Any|None]) -> None: + self.anp:AnPInterface = anp + self.__key:str = key + self.__from_path:str = Common.get_value("from_path", inputs) + self.__to_path:str = Common.get_value("to_path", inputs) + self.__exceptions:Sequence[str] = Common.get_value("exceptions", inputs, []) + self.__time_refresh:float|int = Common.get_value("time_refresh", inputs, 5) + self.__module:str|None = Common.get_value("module", inputs) + self.__working:bool = False + self.__thread:Thread|None = None + self.__cache:dict[str, HashModel] = {} + self.__print_changes:bool = Common.get_value("print_changes", inputs, False) + + if self.__module is not None: + if self.__module == "python": + + module_path:str + + for module_path in system_paths: + if RE.PYTHON_MODULES_PATH.search(module_path): + self.__to_path = module_path + "/" + self.__to_path + break + + def is_working(self:Self) -> bool: + return self.__working and self.anp.working() + + def __analyze(self:Self, path:str) -> None: + + file_name:str + + if self.anp.files.exists(self.__to_path + "/" + path): + self.anp.files.make_directory(self.__to_path + "/" + path) + self.__print_changes and self.anp.print("info", "anp_directory_clone_created", { + "key" : self.__key, + "directory_name" : path, + "from" : self.__from_path, + "to" : self.__to_path + }) + + for file_name in self.anp.files.list_directory_items(self.__from_path + "/" + path): + if not self.is_working(): + break + if file_name in (".", ".."): + continue + + from_path:str = self.__from_path + "/" + path + "/" + file_name + + if any(exception in from_path for exception in self.__exceptions): + continue + if self.anp.files.is_directory(from_path): + self.__analyze(path + "/" + file_name) + elif self.anp.files.is_file(from_path): + + data:bytes = self.anp.files.load(from_path, "rb") + from_hash:HashModel = hash(data) + to_path:str = self.__to_path + "/" + path + "/" + file_name + + if self.anp.files.exists(to_path): + if to_path not in self.__cache: + self.__cache[to_path] = HashModel(self.anp.files.load(to_path, "rb")) + if from_hash.compare(self.__cache[to_path]): + continue + + self.__cache[to_path] = from_hash + self.anp.files.save(to_path, data) + self.__print_changes and self.anp.print("info", "anp_directory_clone_file_created", { + "key" : self.__key, + "directory_name" : path, + "from_directory" : self.__from_path, + "to_directory" : self.__to_path, + "file_name" : file_name + }) + + def __run(self:Self) -> None: + while self.is_working(): + self.__analyze("") + self.anp.wait(self.__time_refresh) + self.__working = False + + def start(self:Self) -> None: + if not self.is_working(): + self.__working = True + self.__thread = Thread(target = self.__run) + self.__thread.start() + + def stop(self:Self) -> None: + if self.is_working(): + self.__working = False \ No newline at end of file diff --git a/Python/AnP/Utils/Patterns.py b/Python/AnP/Utils/Patterns.py index d3afcce..57241c9 100644 --- a/Python/AnP/Utils/Patterns.py +++ b/Python/AnP/Utils/Patterns.py @@ -18,4 +18,5 @@ class RE: PARENT_PATH:REPattern = re_compile(r'^(.*)[\/\\][^\/\\]+[\/\\]?$') DOUBLE_NEW_LINE:REPattern = re_compile(r'(?:\r\n){2}|\r{2}|\n{2}') HTTP_HEADER_PARAMETER:REPattern = re_compile(r'([^:]+):\s*(.+)') - HTTP_REQUEST:REPattern = re_compile(r'^([^\s]+)\s([^\s\?\#]+)(?:\?([^#]+))?(?:\#([^\s]+))?\s([^\/]+)\/([0-9\.]+)$') \ No newline at end of file + HTTP_REQUEST:REPattern = re_compile(r'^([^\s]+)\s([^\s\?\#]+)(?:\?([^#]+))?(?:\#([^\s]+))?\s([^\/]+)\/([0-9\.]+)$') + PYTHON_MODULES_PATH:REPattern = re_compile(r'[\/\\]python[0-9]+[\/\\]site-packages[\/\\]?$', RE_IGNORECASE) \ No newline at end of file