KyMAN/AnP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#wip(py): Scrap system debugging.

Browse Source
This commit is contained in:
KyMAN 2026-07-03 07:30:01 +02:00
parent 306a309df8
commit e3c99a1143
15 changed files with 233 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -1,5 +1,6 @@
/Data
/Public/data
[Ss]ecrets.*
*.[Ss]ecrets.*
*.[Ss]ecret.*
*.deleted.*

7
JSON/I18N/AnP.i18n.espanol.json
View File

@ -69,7 +69,12 @@
"anp_browser_action_not_done" : "La acción '{i}' de tipo '{action}' no ha sido ejecutada correctamente en el navegador '{key}'.",
"anp_browser_action_not_supported" : "La acción '{i}' de tipo '{action}' no es compatible con el navegador '{key}'.",
"anp_browser_action_not_action" : "El elemento '{i}' no es una acción válida en el navegador '{key}'.",
"AnP_BrowserAbstract_end" : null
"AnP_BrowserAbstract_end" : null,
"AnP_DirectoryCloneModel_start" : null,
"anp_directory_clone_created" : "El directorio '{directory_name}' ha sido clonado a '{to}'.",
"anp_directory_clone_file_created" : "El archivo '{file_name}' ha sido clonado desde '{from}/{directory_name}' a '{to}/{directory_name}'.",
"AnP_DirectoryCloneModel_end" : null
}
}

59
Python/AnP/Abstracts/BrowserAbstract.py
View File

@ -3,6 +3,7 @@
from typing import Optional, Self, Any, Callable, Sequence, TypeVar
from abc import abstractmethod
from time import time as timestamp, sleep
from AnP.Interfaces.Application.AnPInterface import AnPInterface
from AnP.DSLs.ScrapDSL import (
Execute, Action, Open, Close, Switch, Wait, Do, Go, Get, ForEach, And, Or, Not,
@ -36,20 +37,41 @@ class BrowserAbstract(ModelAbstract):
Or : self._or,
Not : self._not
}
self.__is_working:dict[str, Callable[[], bool]] = {}
def start(self:Self) -> None:
self.working = True
def close(self:Self) -> None:
def stop(self:Self) -> None:
self.working = False
def remove(self:Self) -> None:
self.stop()
self.tabs.clear()
self.tab_selected = ""
def is_working(self:Self) -> bool:
return self.working and self.anp.working()
def is_working(self:Self, shared:Optional[dict[str, Any|None]] = None) -> bool:
return self.working and self.anp.working() and (
shared is None or
"__is_working_key__" not in shared or
shared["__is_working_key__"] not in self.__is_working or
self.__is_working[shared["__is_working_key__"]]()
)
@abstractmethod
def run(self:Self, actions:Sequence[Action], shared:dict[str, Any|None] = {}) -> dict[str, Any|None]:pass
def run(self:Self,
actions:Sequence[Action],
shared:dict[str, Any|None] = {},
key:Optional[str] = None,
is_working:Optional[Callable[[], bool]] = None
) -> dict[str, Any|None]:
if key is not None and is_working is not None:
self.__is_working[key] = is_working
shared["__is_working_key__"] = key
self.do(Do(*actions), shared)
return shared
def execute(self:Self, action:Execute, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
try:
@ -68,13 +90,18 @@ class BrowserAbstract(ModelAbstract):
def switch(self:Self, action:Switch, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:pass
def wait(self:Self, action:Wait, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
if action.is_fixed:
self.anp.wait(action.time)
else:
if action.is_range:
self.anp.wait(Common.random_number(*action.time[0]))
else:
self.anp.wait(Common.random(*action.time))
date:float = timestamp()
time:int|float = (
action.time if action.is_fixed else
Common.random_number(*action.time[0]) if action.is_range else
Common.random(*action.time))
while self.is_working(shared):
sleep(.1)
if timestamp() - date >= time:
break
action.done = True
def do(self:Self, action:Do, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
@ -85,7 +112,7 @@ class BrowserAbstract(ModelAbstract):
action.done = True
for i, subaction in enumerate(action.actions):
if not self.anp.working():
if not self.is_working(shared):
action.done = False
return
@ -179,7 +206,7 @@ class BrowserAbstract(ModelAbstract):
self.get_selector(action.items, last_item, True) if is_selector else
action.items):
if not self.anp.working():
if not self.is_working(shared):
done = False
break
@ -291,7 +318,7 @@ class BrowserAbstract(ModelAbstract):
action.condition = Condition((action.condition,))
while True:
if not self.anp.working():
if not self.is_working(shared):
action.done = False
break
@ -322,7 +349,7 @@ class BrowserAbstract(ModelAbstract):
action.condition = Condition((action.condition,))
while True:
if not self.anp.working():
if not self.is_working(shared):
action.done = False
break

4
Python/AnP/Abstracts/FilesAbstract.py
View File

@ -15,10 +15,10 @@ class FilesAbstract(ABC):
self.anp:AnPInterface = anp
root_path:str = directory_name(absolute_path(__file__))
self._slash:str = Common.get_value("slash", inputs, "/" if "/" in root_path else "\\\\")
self._root_paths:list[str] = Common.get_value(
self._root_paths:list[str] = [""] + Common.get_value(
"root_paths",
inputs,
[""] + self.get_parents(root_path, 4)
self.get_parents(root_path, 4)
)
def set_root_paths(self:Self, paths:str|Sequence[str]) -> None:

2
Python/AnP/Application/AnP.py
View File

@ -42,7 +42,7 @@ class AnP:
def __init__(self:Self, inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None) -> None:
self.__working:bool = True
self.files:FilesAbstract = FilesDriver(self)
self.files:FilesAbstract = FilesDriver(self, inputs)
self.settings:SettingsManager = SettingsManager(self, inputs)
self.i18n:I18NManager = I18NManager(self)
self.print_types:PrintTypesManager = PrintTypesManager(self)

46
Python/AnP/Drivers/SeleniumDriver.py
View File

@ -10,6 +10,7 @@ from selenium.webdriver.common.by import By, ByType
from selenium.webdriver.remote.webdriver import WebDriver, WebElement
from seleniumwire.utils import decode as sw_decode
from seleniumwire.request import Request
from selenium.webdriver.common.action_chains import ActionChains
from AnP.DSLs.ScrapDSL import (
Open, Close, Switch, Go, Selector, XPath, CSS, Name, LinkText, Class, ID,
Get, Download
@ -29,6 +30,12 @@ class SeleniumDriver(BrowserAbstract):
self.__cache_directory:str = self.anp.settings.get(("selenium_cache_directory", "cache_directory"), inputs, "/Data/Cache/Selenium")
self.__download_timeout:int = self.anp.settings.get(("selenium_download_timeout", "download_timeout"), inputs, 60)
def remove(self:Self) -> None:
self.stop()
self.browser and self.browser.quit()
self.tabs.clear()
self.tab_selected = ""
def __load(self:Self, url:str|Callable[[dict[str, Any|None]], str], shared:dict[str, Any|None] = {}) -> None:
if Check.is_function(url):
@ -103,13 +110,29 @@ class SeleniumDriver(BrowserAbstract):
shared[action.key] = []
if multiple:
shared[action.key].extend(
[self.get_value(action, element.text) for element in item] if action.attribute == "CDATA" else
[self.get_value(action, element.get_attribute(action.attribute)) for element in item])
element:WebElement
for element in item:
ActionChains(self.__browser).move_to_element(element).perform()
shared[action.key].append(element.get_attribute(
"textContent" if action.attribute == "CDATA" else
action.attribute))
else:
shared[action.key] = (
self.get_value(action, item.text) if action.attribute == "CDATA" else
self.get_value(action, item.get_attribute(action.attribute)))
ActionChains(self.__browser).move_to_element(item).perform()
shared[action.key] = (item.get_attribute(
"textContent" if action.attribute == "CDATA" else
action.attribute))
# if multiple:
# shared[action.key].extend(
# [self.get_value(action, element.text) for element in item] if action.attribute == "CDATA" else
# [self.get_value(action, element.get_attribute(action.attribute)) for element in item])
# else:
# shared[action.key] = (
# self.get_value(action, item.text) if action.attribute == "CDATA" else
# self.get_value(action, item.get_attribute(action.attribute)))
action.done = True
@ -179,8 +202,15 @@ class SeleniumDriver(BrowserAbstract):
data:bytes|None = method(url)
if data:
if data is not None:
if action.key:
shared[action.key] = data
if action.multiple:
if action.key not in shared:
shared[action.key] = []
shared[action.key].append(data)
else:
shared[action.key] = data
if action.path and (action.overwrite or not self.anp.files.exists(action.path)):
self.anp.files.save(action.path, data)
action.done = True
break

6
Python/AnP/Drivers/XMLScrapDriver.py
View File

@ -18,12 +18,6 @@ class XMLScrapDriver(BrowserAbstract):
def __init__(self:Self, anp:AnPInterface, key:str, inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None) -> None:
super().__init__(anp, key, XMLScrapTabModel, inputs)
def run(self:Self, actions:Sequence[Action], shared:dict[str, Any|None] = {}) -> dict[str, Any|None]:
self.do(Do(*actions), shared)
return shared
def __load(self:Self, url:str) -> Exception|None:
try:

5
Python/AnP/Managers/BrowsersManager.py
View File

@ -2,6 +2,7 @@
# -*- coding: utf-8 -*-
from typing import Self, Any, Sequence
from inspect import isclass
from AnP.Interfaces.Application.AnPInterface import AnPInterface
from AnP.Abstracts.BrowserAbstract import BrowserAbstract
from AnP.Utils.Checks import Check
@ -30,7 +31,7 @@ class BrowsersManager:
def close(self:Self) -> None:
for key, browser in list(self.__browsers.items()):
browser.close()
browser.remove()
del self.__browsers[key]
def add(self:Self, inputs:Any|None, overwrite:bool = False) -> None:
@ -48,7 +49,7 @@ class BrowsersManager:
if Check.is_string(browser):
BrowserClass = self.anp.models.get(BrowserAbstract, browser)
elif issubclass(browser, BrowserAbstract):
elif isclass(browser) and issubclass(browser, BrowserAbstract):
BrowserClass = browser
elif Check.is_dictionary(browser) and "type" in browser and Check.is_string(browser["type"]):
BrowserClass = self.anp.models.get(BrowserAbstract, browser["type"])

8
Python/AnP/Managers/DirectoriesClonesManager.py
View File

@ -12,6 +12,8 @@ class DirectoriesClonesManager:
self.anp:AnPInterface = anp
self.__clones:dict[str, DirectoryCloneModel] = {}
self.update()
def update(self:Self) -> None:
key:str
@ -53,7 +55,7 @@ class DirectoriesClonesManager:
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
list(self.__clones.keys()) if keys is None else
[]):
self.__clones[key].stop()
del self.__clones[key]
@ -65,7 +67,7 @@ class DirectoriesClonesManager:
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
list(self.__clones.keys()) if keys is None else
[]):
self.__clones[key].start()
@ -76,6 +78,6 @@ class DirectoriesClonesManager:
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
list(self.__clones.keys()) if keys is None else
[]):
self.__clones[key].stop()

5
Python/AnP/Managers/ModelsManager.py
View File

@ -2,6 +2,7 @@
# -*- coding: utf-8 -*-
from typing import Self, Any, Sequence, TypeVar
from inspect import isclass
from AnP.Interfaces.Application.AnPInterface import AnPInterface
from AnP.Abstracts.ModelAbstract import ModelAbstract
from AnP.Utils.Common import Common
@ -38,7 +39,7 @@ class ModelsManager:
for key in Common.get_keys(keys):
if key in self.__models and (
isinstance(self.__models[key], Type) or
issubclass(self.__models[key], Type)
(isclass(self.__models[key]) and issubclass(self.__models[key], Type))
):
return self.__models[key]
return None
@ -51,7 +52,7 @@ class ModelsManager:
for key, model in subinputs.items():
if Check.is_mark_key(key) and model is None:
continue
if issubclass(model, ModelAbstract) and (
if isclass(model) and issubclass(model, ModelAbstract) and (
overwrite or key not in self.__models
):
self.__models[key] = model

49
Python/AnP/Models/DirectoryCloneModel.py
View File

@ -14,9 +14,9 @@ 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.__from_path:str = Common.get_value("from", inputs)
self.__to_path:str = Common.get_value("to", inputs)
self.__exclude:Sequence[str] = Common.get_value("exclude", 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
@ -34,6 +34,8 @@ class DirectoryCloneModel:
self.__to_path = module_path + "/" + self.__to_path
break
Common.get_value("autorun", inputs, True) and self.start()
def is_working(self:Self) -> bool:
return self.__working and self.anp.working()
@ -41,8 +43,11 @@ class DirectoryCloneModel:
file_name:str
if self.anp.files.exists(self.__to_path + "/" + path):
self.anp.files.make_directory(self.__to_path + "/" + path)
to_path:str = self.__to_path + ("/" + path if path else "")
from_path:str = self.__from_path + ("/" + path if path else "")
if not self.anp.files.exists(to_path):
self.anp.files.make_directory(to_path)
self.__print_changes and self.anp.print("info", "anp_directory_clone_created", {
"key" : self.__key,
"directory_name" : path,
@ -50,37 +55,37 @@ class DirectoryCloneModel:
"to" : self.__to_path
})
for file_name in self.anp.files.list_directory_items(self.__from_path + "/" + path):
for file_name in self.anp.files.list_directory_items(from_path):
if not self.is_working():
break
if file_name in (".", ".."):
continue
from_path:str = self.__from_path + "/" + path + "/" + file_name
from_path_file:str = from_path + "/" + file_name
if any(exception in from_path for exception in self.__exceptions):
if any(exception in path for exception in self.__exclude):
continue
if self.anp.files.is_directory(from_path):
self.__analyze(path + "/" + file_name)
elif self.anp.files.is_file(from_path):
if self.anp.files.is_directory(from_path_file):
self.__analyze((path + "/" if path else "") + file_name)
elif self.anp.files.is_file(from_path_file):
data:bytes = self.anp.files.load(from_path, "rb")
from_hash:HashModel = hash(data)
to_path:str = self.__to_path + "/" + path + "/" + file_name
data:bytes = self.anp.files.load(from_path_file, "rb")
from_hash:HashModel = HashModel(data)
to_path_file:str = to_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]):
if self.anp.files.exists(to_path_file):
if to_path_file not in self.__cache:
self.__cache[to_path_file] = HashModel(self.anp.files.load(to_path_file, "rb"))
if from_hash.compare(self.__cache[to_path_file]):
continue
self.__cache[to_path] = from_hash
self.anp.files.save(to_path, data)
self.__cache[to_path_file] = from_hash
self.anp.files.save(to_path_file, 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,
"from" : self.__from_path,
"to" : self.__to_path,
"file_name" : file_name
})

2
Python/AnP/Utils/Patterns.py
View File

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

4
Python/base.py
View File

@ -9,6 +9,7 @@ from AnP.Drivers.HTTPServerDriver import HTTPServerDriver
from AnP.Drivers.OllamaDriver import OllamaDriver
from AnP.Drivers.FilesDriver import FilesDriver
from AnP.Drivers.XMLScrapDriver import XMLScrapDriver
from AnP.Drivers.SeleniumDriver import SeleniumDriver
from AnP.Controllers.CloneController import CloneController
from AnP.Controllers.AITranslateController import AITranslateController
@ -22,7 +23,8 @@ inputs:dict[str, dict[str, Any|None]] = {
"HTTPServerDriver" : HTTPServerDriver,
"OllamaDriver" : OllamaDriver,
"FilesDriver" : FilesDriver,
"XMLScrapDriver" : XMLScrapDriver
"XMLScrapDriver" : XMLScrapDriver,
"SeleniumDriver" : SeleniumDriver
}
}

2
Python/run.py
View File

@ -6,4 +6,4 @@ from base import inputs
anp:AnP = AnP(inputs)
anp.controllers.execute("clonations_processes", "execute", None)
# anp.controllers.execute("clonations_processes", "execute", None)

100
README.md
View File

@ -1,3 +1,101 @@
# AnP
Framework.
Framework.
Secrets Python file:
```py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# /Python/secrets.py
from typing import Any
secrets:dict[str, Any|None] = {
"root_paths" : ["/ABSOLUTE/PATH/AnP"]
}
```
Web Socket Server Settings:
```json
{
"default_web_sockets_servers" : {
"anp" : {
"type" : "WebSocketServerDriver",
"host" : "",
"port" : 18765
}
}
}
```
HTTP Settings:
```json
{
"default_http_servers" : {
"anp" : {
"type" : "HTTPServerDriver",
"type2" : "HTTPSocketServerDriver",
"host" : "",
"port" : 18000
}
}
}
```
AI Interpreters Settings:
```json
{
"default_ai_interpreters" : {
"anp_titles" : {
"type" : "OllamaDriver",
"url" : "http://localhost:11434/api/generate/",
"model" : "gemma3:1b",
"pool" : "anp",
"format" : {
"type": "array",
"items": {
"type": "boolean"
}
},
"stream" : false,
"think" : false,
"temperature" : 0.0,
"allow_contexts" : true
},
"anp_responses" : {
"type" : "OllamaDriver",
"url" : "http://localhost:11434/api/generate/",
"model" : "gemma4:e4b",
"stream" : true,
"pool" : "anp",
"think" : false,
"temperature" : 0.7,
"allow_contexts" : true
}
}
}
```
Directory Cloner Settings:
```json
{
"default_directories_clones_files" : {
"anp" : {
"from" : "/ABSOLUTE/PATH/AnP/Python/AnP",
"to" : "AnP",
"exclude" : ["__pycache__", "deleted", "secret", ".pyc"],
"time_refresh" : 5,
"module" : "python",
"autorun" : true,
"print_changes" : true
}
}
}
```