feat(py): Directories Clones.

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

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

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

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

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()

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

Python/AnP/Utils/Patterns.py

@@ -19,3 +19,4 @@ class RE:
     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\.]+)$')
+    PYTHON_MODULES_PATH:REPattern = re_compile(r'[\/\\]python[0-9]+[\/\\]site-packages[\/\\]?$', RE_IGNORECASE)