#wip(py): Selenium Driver.
This commit is contained in:
parent
bc0a86900a
commit
f9a13c14ac
@ -6,7 +6,7 @@ from abc import abstractmethod
|
||||
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,
|
||||
If, While, DoWhile, Download
|
||||
If, While, DoWhile, Download, Selector, Condition
|
||||
)
|
||||
from AnP.Utils.Common import Common
|
||||
from AnP.Abstracts.ModelAbstract import ModelAbstract
|
||||
@ -51,8 +51,12 @@ class BrowserAbstract(ModelAbstract):
|
||||
@abstractmethod
|
||||
def run(self:Self, actions:Sequence[Action], shared:dict[str, Any|None] = {}) -> dict[str, Any|None]:pass
|
||||
|
||||
@abstractmethod
|
||||
def execute(self:Self, action:Execute, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:pass
|
||||
def execute(self:Self, action:Execute, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
try:
|
||||
action.callback(shared)
|
||||
action.done = True
|
||||
except Exception as exception:
|
||||
action.exception = exception
|
||||
|
||||
@abstractmethod
|
||||
def open(self:Self, action:Open, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:pass
|
||||
@ -153,29 +157,193 @@ class BrowserAbstract(ModelAbstract):
|
||||
@abstractmethod
|
||||
def go(self:Self, action:Go, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:pass
|
||||
|
||||
@abstractmethod
|
||||
def get_selector(self:Self, selector:Selector, from_item:Optional[Any] = None, force_multiple:bool = False) -> Any|list[Any]|None:pass
|
||||
|
||||
@staticmethod
|
||||
def get_value(action:Get, value:str) -> str:
|
||||
if action.pattern is None:
|
||||
return value
|
||||
return action.pattern.sub(action.matches, value)
|
||||
|
||||
@abstractmethod
|
||||
def get(self:Self, action:Get, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:pass
|
||||
|
||||
@abstractmethod
|
||||
def foreach(self:Self, action:ForEach, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:pass
|
||||
def foreach(self:Self, action:ForEach, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
|
||||
@abstractmethod
|
||||
def _and(self:Self, action:And, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> bool:pass
|
||||
item:Any|None
|
||||
done:bool = True
|
||||
is_selector:bool = isinstance(action.items, Selector)
|
||||
|
||||
for item in (
|
||||
self.get_selector(action.items, last_item, True) if is_selector else
|
||||
action.items):
|
||||
|
||||
if not self.anp.working():
|
||||
done = False
|
||||
break
|
||||
|
||||
@abstractmethod
|
||||
def _or(self:Self, action:Or, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> bool:pass
|
||||
if action.callback is not None:
|
||||
try:
|
||||
item = action.callback(item)
|
||||
except Exception as exception:
|
||||
action.exception = exception
|
||||
done = False
|
||||
break
|
||||
|
||||
@abstractmethod
|
||||
def _not(self:Self, action:Not, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> bool:pass
|
||||
subactions:Do = Do(*action.actions)
|
||||
|
||||
@abstractmethod
|
||||
def _if(self:Self, action:If, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:pass
|
||||
self.do(subactions, shared, item if is_selector else last_item)
|
||||
if not subactions.done or subactions.exception is not None:
|
||||
action.exception = subactions.exception
|
||||
done = False
|
||||
break
|
||||
|
||||
@abstractmethod
|
||||
def _while(self:Self, action:While, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:pass
|
||||
action.done = done
|
||||
|
||||
@abstractmethod
|
||||
def do_while(self:Self, action:DoWhile, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:pass
|
||||
def _and(self:Self, action:And, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> bool:
|
||||
|
||||
condition:Callable[[dict[str, Any|None]], bool]|Selector|Condition
|
||||
|
||||
action.ok = True
|
||||
|
||||
for condition in action.conditions:
|
||||
if isinstance(condition, Condition):
|
||||
self.do(condition, shared, last_item)
|
||||
if not condition.ok:
|
||||
action.ok = False
|
||||
break
|
||||
elif isinstance(condition, Selector):
|
||||
if not self.get_selector(condition, last_item):
|
||||
action.ok = False
|
||||
break
|
||||
elif callable(condition):
|
||||
try:
|
||||
if not condition(shared):
|
||||
action.ok = False
|
||||
break
|
||||
except Exception as exception:
|
||||
action.exception = exception
|
||||
action.ok = False
|
||||
break
|
||||
|
||||
def _or(self:Self, action:Or, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> bool:
|
||||
|
||||
condition:Callable[[dict[str, Any|None]], bool]|Selector|Condition
|
||||
|
||||
action.ok = False
|
||||
|
||||
for condition in action.conditions:
|
||||
if isinstance(condition, Condition):
|
||||
self.do(condition, shared, last_item)
|
||||
if condition.ok:
|
||||
action.ok = True
|
||||
break
|
||||
elif isinstance(condition, Selector):
|
||||
if self.get_selector(condition, last_item):
|
||||
action.ok = True
|
||||
break
|
||||
elif callable(condition):
|
||||
try:
|
||||
if condition(shared):
|
||||
action.ok = True
|
||||
break
|
||||
except Exception as exception:
|
||||
action.exception = exception
|
||||
action.ok = False
|
||||
break
|
||||
|
||||
def _not(self:Self, action:Not, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> bool:
|
||||
if isinstance(action.condition, Condition):
|
||||
self.do(action.condition, shared, last_item)
|
||||
action.ok = not action.condition.ok
|
||||
elif isinstance(action.condition, Selector):
|
||||
action.ok = not self.get_selector(action.condition, last_item)
|
||||
elif callable(action.condition):
|
||||
try:
|
||||
action.ok = not action.condition(shared)
|
||||
except Exception as exception:
|
||||
action.exception = exception
|
||||
action.ok = False
|
||||
|
||||
def _if(self:Self, action:If, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
|
||||
if not isinstance(action.condition, Condition):
|
||||
action.condition = Condition((action.condition,))
|
||||
|
||||
self.do(action.condition, shared, last_item)
|
||||
|
||||
if action.condition.done and action.condition.exception is None:
|
||||
|
||||
subactions:Do = Do(*((action.if_actions if action.condition.ok else action.else_actions) or []))
|
||||
|
||||
self.do(subactions, shared, last_item)
|
||||
action.done = subactions.done and subactions.exception is None
|
||||
action.exception = subactions.exception
|
||||
|
||||
else:
|
||||
action.done = action.condition.done
|
||||
action.exception = action.condition.exception
|
||||
|
||||
def _while(self:Self, action:While, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
|
||||
if not isinstance(action.condition, Condition):
|
||||
action.condition = Condition((action.condition,))
|
||||
|
||||
while True:
|
||||
if not self.anp.working():
|
||||
action.done = False
|
||||
break
|
||||
|
||||
self.do(action.condition, shared, last_item)
|
||||
|
||||
if not action.condition.done or action.condition.exception is not None:
|
||||
action.done = action.condition.done
|
||||
action.exception = action.condition.exception
|
||||
break
|
||||
|
||||
if action.condition.ok:
|
||||
|
||||
subactions:Do = Do(*action.actions)
|
||||
|
||||
self.do(subactions, shared, last_item)
|
||||
if not subactions.done or subactions.exception is not None:
|
||||
action.exception = subactions.exception
|
||||
action.done = False
|
||||
break
|
||||
|
||||
else:
|
||||
action.done = True
|
||||
break
|
||||
|
||||
def do_while(self:Self, action:DoWhile, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
|
||||
if not isinstance(action.condition, Condition):
|
||||
action.condition = Condition((action.condition,))
|
||||
|
||||
while True:
|
||||
if not self.anp.working():
|
||||
action.done = False
|
||||
break
|
||||
|
||||
subactions:Do = Do(*action.actions)
|
||||
|
||||
self.do(subactions, shared, last_item)
|
||||
if not subactions.done or subactions.exception is not None:
|
||||
action.exception = subactions.exception
|
||||
action.done = False
|
||||
break
|
||||
|
||||
self.do(action.condition, shared, last_item)
|
||||
|
||||
if not action.condition.done or action.condition.exception is not None:
|
||||
action.done = action.condition.done
|
||||
action.exception = action.condition.exception
|
||||
break
|
||||
|
||||
if not action.condition.ok:
|
||||
action.done = True
|
||||
break
|
||||
|
||||
@abstractmethod
|
||||
def download(self:Self, action:Download, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:pass
|
||||
@ -3,6 +3,7 @@
|
||||
|
||||
from typing import Self, Sequence, Any, Optional, Callable
|
||||
from re import Pattern as REPattern, Match as REMatch
|
||||
from random import random
|
||||
|
||||
class Action:
|
||||
def __init__(self:Self) -> None:
|
||||
@ -59,6 +60,8 @@ class Selector(Action):
|
||||
super().__init__()
|
||||
self.selector:str = selector
|
||||
self.multiple:bool = multiple
|
||||
self.x:int|float|None = None
|
||||
self.y:int|float|None = None
|
||||
|
||||
class XPath(Selector):pass
|
||||
|
||||
@ -163,4 +166,35 @@ class Download(Action):
|
||||
self.attribute:str|None = attribute
|
||||
self.path:str|None = path
|
||||
self.multiple:bool = multiple
|
||||
self.overwrite:bool = overwrite
|
||||
self.overwrite:bool = overwrite
|
||||
|
||||
class Position:
|
||||
def __init__(self:Self, x:int|float, y:int|float) -> None:
|
||||
self.x:int|float = x
|
||||
self.y:int|float = y
|
||||
|
||||
class MouseAction(Action):
|
||||
def __init__(self:Self, position:Position|Selector, margin:int|float = 3, size:Optional[Position] = None) -> None:
|
||||
super().__init__()
|
||||
self.position:Position|Selector = position
|
||||
self.margin:int|float = margin
|
||||
self.size:Position|None = size
|
||||
|
||||
def get_size(self:Self) -> Position:
|
||||
return (
|
||||
self.size if self.size is not None else
|
||||
Position(self.position.x, self.position.y) if isinstance(self.position, Selector) else
|
||||
Position(10, 10))
|
||||
|
||||
def get(self:Self) -> tuple[float, float]:
|
||||
|
||||
size:Position = self.get_size()
|
||||
|
||||
return (
|
||||
self.position.x + random() * (size.x - 2 * self.margin.x) + self.margin.x,
|
||||
self.position.y + random() * (size.y - 2 * self.margin.y) + self.margin.y
|
||||
)
|
||||
|
||||
class Move(MouseAction):pass
|
||||
|
||||
class Click(MouseAction):pass
|
||||
@ -1,12 +1,186 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from typing import Self
|
||||
from AnP.DSLs.ScrapDSL import Action
|
||||
from typing import Self, Optional, Any, Sequence, Union, Callable
|
||||
from time import time as timestamp, sleep
|
||||
from os import listdir as list_directory
|
||||
from seleniumwire import webdriver
|
||||
from selenium.webdriver.remote.webdriver import WebDriver
|
||||
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 AnP.DSLs.ScrapDSL import (
|
||||
Open, Close, Switch, Go, Selector, XPath, CSS, Name, LinkText, Class, ID,
|
||||
Get, Download
|
||||
)
|
||||
from AnP.Interfaces.Application import AnPInterface
|
||||
from AnP.Abstracts.BrowserAbstract import BrowserAbstract
|
||||
from AnP.Models.SeleniumTabModel import SeleniumTabModel
|
||||
from AnP.Utils.Checks import Check
|
||||
|
||||
WireWebDriver = Union[WebDriver, webdriver.Firefox]
|
||||
|
||||
class SeleniumDriver(BrowserAbstract):
|
||||
|
||||
def __init__(self:Self, anp:AnPInterface) -> None:
|
||||
super().__init__(anp)
|
||||
def __init__(self:Self, anp:AnPInterface, key:str, inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None) -> None:
|
||||
super().__init__(anp, key, SeleniumTabModel, inputs)
|
||||
self.browser:WireWebDriver = None
|
||||
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 __load(self:Self, url:str|Callable[[dict[str, Any|None]], str], shared:dict[str, Any|None] = {}) -> None:
|
||||
|
||||
if Check.is_function(url):
|
||||
url = url(shared)
|
||||
|
||||
self.browser.get(shared[url] if url in shared else url)
|
||||
|
||||
self.browser.set_script_timeout(10)
|
||||
self.browser.execute_async_script("""
|
||||
if(document.readyState == "complete")
|
||||
arguments[arguments.length - 1](200);
|
||||
else
|
||||
window.addEventListener("load", () => arguments[arguments.length - 1](200));
|
||||
""")
|
||||
|
||||
def open(self:Self, action:Open, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
if self.browser is None:
|
||||
self.browser = webdriver.Firefox()
|
||||
else:
|
||||
self.browser.switch_to.new_window("tab")
|
||||
self.tabs[action.key] = self.browser.current_window_handle
|
||||
self.tab_selected = action.key
|
||||
if action.url is not None:
|
||||
self.__load(action.url, shared)
|
||||
|
||||
def close(self:Self, action:Close, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
self.browser.close()
|
||||
del self.tabs[action.key]
|
||||
self.tab_selected = ""
|
||||
action.done = True
|
||||
|
||||
def switch(self:Self, action:Switch, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
if action.key in self.tabs:
|
||||
self.browser.switch_to.window(self.tabs[action.key])
|
||||
self.tab_selected = action.key
|
||||
action.done = True
|
||||
|
||||
def go(self:Self, action:Go, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
self.__load(action.url, shared)
|
||||
action.done = True
|
||||
|
||||
def get_selector(self:Self,
|
||||
selector:Selector,
|
||||
from_item:Optional[WebDriver|WebElement] = None,
|
||||
force_multiple:bool = False
|
||||
) -> list[WebElement]|WebElement|None:
|
||||
if self.tab_selected in self.tabs:
|
||||
|
||||
mode:ByType = (
|
||||
By.XPATH if isinstance(selector, XPath) else
|
||||
By.CSS_SELECTOR if isinstance(selector, CSS) else
|
||||
By.ID if isinstance(selector, ID) else
|
||||
By.CLASS_NAME if isinstance(selector, Class) else
|
||||
By.PARTIAL_LINK_TEXT if isinstance(selector, LinkText) else
|
||||
By.NAME if isinstance(selector, Name) else
|
||||
By.CSS_SELECTOR)
|
||||
|
||||
if from_item is None:
|
||||
from_item = self.browser
|
||||
|
||||
if force_multiple or selector.multiple:
|
||||
return from_item.find_elements(mode, selector.selector)
|
||||
return from_item.find_element(mode, selector.selector)
|
||||
return [] if force_multiple or selector.multiple else None
|
||||
|
||||
def get(self:Self, action:Get, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
|
||||
item:WebElement = last_item or self.browser if action.selector is None else self.get_selector(action.selector, last_item)
|
||||
multiple:bool = action.selector is not None and action.selector.multiple
|
||||
|
||||
if action.key not in shared or not isinstance(shared[action.key], list):
|
||||
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])
|
||||
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
|
||||
|
||||
def __download_from_cache(self:Self, url:str, timeout:Optional[int|float] = None) -> bytes|None:
|
||||
|
||||
request:Request
|
||||
date:float = timestamp()
|
||||
|
||||
while (
|
||||
timestamp() - date < (timeout or self.__download_timeout) and
|
||||
any(file.endswith(".part") for file in list_directory(self.__cache_directory))
|
||||
):
|
||||
if not self.is_working():
|
||||
return None
|
||||
sleep(.1)
|
||||
|
||||
for request in self.browser.requests:
|
||||
if request.url == url and request.response:
|
||||
return sw_decode(
|
||||
request.response.body,
|
||||
request.response.headers.get("Content-Encoding", "identity")
|
||||
)
|
||||
return None
|
||||
|
||||
def __download_by_anchor_label(self:Self, url:str) -> bytes|None:
|
||||
|
||||
# l:int = len(self.browser.window_handles)
|
||||
data:bytes|None = None
|
||||
|
||||
self.browser.execute_script("""
|
||||
|
||||
const anchor = document.createElement("a");
|
||||
|
||||
anchor.href = arguments[0];
|
||||
anchor.download = "";
|
||||
anchor.target = "_blank";
|
||||
anchor.click();
|
||||
anchor.remove();
|
||||
|
||||
""", url)
|
||||
|
||||
data = self.__download_from_cache(url)
|
||||
|
||||
# if l < len(self.browser.window_handles):
|
||||
# self.browser.switch_to.window(self.browser.window_handles[-1])
|
||||
# self.browser.close()
|
||||
# self.browser.switch_to.window(self.browser.window_handles[-1])
|
||||
|
||||
return data
|
||||
|
||||
def __download_by_url(self:Self, url:str) -> bytes|None:
|
||||
return self.browser.get(url)
|
||||
|
||||
def download(self:Self, action:Download, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
|
||||
url:str = action.url(shared) if Check.is_function(action.url) else action.url
|
||||
method:Callable[[str], bytes|None]
|
||||
|
||||
if url in shared:
|
||||
url = shared[url]
|
||||
|
||||
for method in (
|
||||
self.__download_from_cache,
|
||||
self.__download_by_anchor_label,
|
||||
# self.__download_by_url
|
||||
):
|
||||
|
||||
data:bytes|None = method(url)
|
||||
|
||||
if data:
|
||||
if action.key:
|
||||
shared[action.key] = data
|
||||
action.done = True
|
||||
break
|
||||
@ -5,7 +5,7 @@ from typing import Self, Any, Optional, Sequence, Callable
|
||||
from requests import get, Response
|
||||
from parsel import Selector as ParseSelector, SelectorList as ParseSelectorList
|
||||
from AnP.DSLs.ScrapDSL import (
|
||||
Execute, Selector, CSS, XPath, Name, LinkText, Class, ID, Open, Close, Switch,
|
||||
Selector, CSS, XPath, Name, LinkText, Class, ID, Open, Close, Switch,
|
||||
Go, Get, ForEach, Do, Condition, And, Or, Not, If, While, DoWhile, Download,
|
||||
Action
|
||||
)
|
||||
@ -35,13 +35,6 @@ class XMLScrapDriver(BrowserAbstract):
|
||||
except Exception as exception:
|
||||
return exception
|
||||
return None
|
||||
|
||||
def execute(self:Self, action:Execute, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
try:
|
||||
action.callback(shared)
|
||||
action.done = True
|
||||
except Exception as exception:
|
||||
action.exception = exception
|
||||
|
||||
def open(self:Self, action:Open, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
|
||||
@ -72,7 +65,7 @@ class XMLScrapDriver(BrowserAbstract):
|
||||
action.exception = self.__load(url)
|
||||
action.done = action.exception is None
|
||||
|
||||
def get_selector(self:Self, selector:str, from_item:Optional[Selector] = None, force_multiple:bool = False) -> ParseSelectorList|ParseSelector|None:
|
||||
def get_selector(self:Self, selector:str, from_item:Optional[Selector] = None, force_multiple:bool = False) -> Any|list[Any]|None:
|
||||
if self.tab_selected in self.tabs:
|
||||
|
||||
path:ParseSelector = self.tabs[self.tab_selected].selector
|
||||
@ -94,7 +87,7 @@ class XMLScrapDriver(BrowserAbstract):
|
||||
def get(self:Self, action:Get, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
|
||||
item:ParseSelector|ParseSelectorList|None = last_item if action.selector is None else self.get_selector(action.selector, last_item)
|
||||
multiple:bool = False if action.selector is None else action.selector.multiple
|
||||
multiple:bool = action.selector is not None and action.selector.multiple
|
||||
attribute:str = "::text" if action.attribute == "CDATA" else f"::attr({action.attribute})"
|
||||
|
||||
if action.reset and action.key in shared:
|
||||
@ -103,184 +96,11 @@ class XMLScrapDriver(BrowserAbstract):
|
||||
if multiple:
|
||||
if action.key not in shared or not isinstance(shared[action.key], list):
|
||||
shared[action.key] = []
|
||||
shared[action.key].extend([subitem.css(attribute).get() for subitem in item])
|
||||
shared[action.key].extend([self.get_value(action, subitem.css(attribute).get()) for subitem in item])
|
||||
else:
|
||||
shared[action.key] = None if item is None else item.css(attribute).get()
|
||||
shared[action.key] = None if item is None else self.get_value(action, item.css(attribute).get())
|
||||
|
||||
def foreach(self:Self, action:ForEach, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
|
||||
item:ParseSelector|Any|None
|
||||
done:bool = True
|
||||
|
||||
for item in (
|
||||
self.get_selector(action.items, last_item, True) if isinstance(action.items, Selector) else
|
||||
action.items):
|
||||
|
||||
if not self.anp.working():
|
||||
done = False
|
||||
break
|
||||
|
||||
if action.callback is not None:
|
||||
try:
|
||||
item = action.callback(item)
|
||||
except Exception as exception:
|
||||
action.exception = exception
|
||||
done = False
|
||||
break
|
||||
|
||||
subactions:Do = Do(*action.actions)
|
||||
|
||||
self.do(subactions, shared, item if isinstance(item, ParseSelector) else last_item)
|
||||
if not subactions.done or subactions.exception is not None:
|
||||
action.exception = subactions.exception
|
||||
done = False
|
||||
break
|
||||
|
||||
action.done = done
|
||||
|
||||
def _and(self:Self, action:And, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> bool:
|
||||
|
||||
condition:Callable[[dict[str, Any|None]], bool]|Selector|Condition
|
||||
|
||||
action.ok = True
|
||||
|
||||
for condition in action.conditions:
|
||||
if isinstance(condition, Condition):
|
||||
self.do(condition, shared, last_item)
|
||||
if not condition.ok:
|
||||
action.ok = False
|
||||
break
|
||||
elif isinstance(condition, Selector):
|
||||
if not self.get_selector(condition, last_item):
|
||||
action.ok = False
|
||||
break
|
||||
elif callable(condition):
|
||||
try:
|
||||
if not condition(shared):
|
||||
action.ok = False
|
||||
break
|
||||
except Exception as exception:
|
||||
action.exception = exception
|
||||
action.ok = False
|
||||
break
|
||||
|
||||
def _or(self:Self, action:Or, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> bool:
|
||||
|
||||
condition:Callable[[dict[str, Any|None]], bool]|Selector|Condition
|
||||
|
||||
action.ok = False
|
||||
|
||||
for condition in action.conditions:
|
||||
if isinstance(condition, Condition):
|
||||
self.do(condition, shared, last_item)
|
||||
if condition.ok:
|
||||
action.ok = True
|
||||
break
|
||||
elif isinstance(condition, Selector):
|
||||
if self.get_selector(condition, last_item):
|
||||
action.ok = True
|
||||
break
|
||||
elif callable(condition):
|
||||
try:
|
||||
if condition(shared):
|
||||
action.ok = True
|
||||
break
|
||||
except Exception as exception:
|
||||
action.exception = exception
|
||||
action.ok = False
|
||||
break
|
||||
|
||||
def _not(self:Self, action:Not, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> bool:
|
||||
if isinstance(action.condition, Condition):
|
||||
self.do(action.condition, shared, last_item)
|
||||
action.ok = not action.condition.ok
|
||||
elif isinstance(action.condition, Selector):
|
||||
action.ok = not self.get_selector(action.condition, last_item)
|
||||
elif callable(action.condition):
|
||||
try:
|
||||
action.ok = not action.condition(shared)
|
||||
except Exception as exception:
|
||||
action.exception = exception
|
||||
action.ok = False
|
||||
|
||||
def _if(self:Self, action:If, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
|
||||
if not isinstance(action.condition, Condition):
|
||||
action.condition = Condition((action.condition,))
|
||||
|
||||
self.do(action.condition, shared, last_item)
|
||||
|
||||
if action.condition.done and action.condition.exception is None:
|
||||
|
||||
subactions:Do = Do(*((action.if_actions if action.condition.ok else action.else_actions) or []))
|
||||
|
||||
self.do(subactions, shared, last_item)
|
||||
action.done = subactions.done and subactions.exception is None
|
||||
action.exception = subactions.exception
|
||||
|
||||
else:
|
||||
action.done = action.condition.done
|
||||
action.exception = action.condition.exception
|
||||
|
||||
def _while(self:Self, action:While, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
|
||||
if not isinstance(action.condition, Condition):
|
||||
action.condition = Condition((action.condition,))
|
||||
|
||||
while True:
|
||||
if not self.anp.working():
|
||||
action.done = False
|
||||
break
|
||||
|
||||
self.do(action.condition, shared, last_item)
|
||||
|
||||
if not action.condition.done or action.condition.exception is not None:
|
||||
action.done = action.condition.done
|
||||
action.exception = action.condition.exception
|
||||
break
|
||||
|
||||
if action.condition.ok:
|
||||
|
||||
subactions:Do = Do(*action.actions)
|
||||
|
||||
self.do(subactions, shared, last_item)
|
||||
if not subactions.done or subactions.exception is not None:
|
||||
action.exception = subactions.exception
|
||||
action.done = False
|
||||
break
|
||||
|
||||
else:
|
||||
action.done = True
|
||||
break
|
||||
|
||||
def do_while(self:Self, action:DoWhile, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
|
||||
if not isinstance(action.condition, Condition):
|
||||
action.condition = Condition((action.condition,))
|
||||
|
||||
while True:
|
||||
if not self.anp.working():
|
||||
action.done = False
|
||||
break
|
||||
|
||||
subactions:Do = Do(*action.actions)
|
||||
|
||||
self.do(subactions, shared, last_item)
|
||||
if not subactions.done or subactions.exception is not None:
|
||||
action.exception = subactions.exception
|
||||
action.done = False
|
||||
break
|
||||
|
||||
self.do(action.condition, shared, last_item)
|
||||
|
||||
if not action.condition.done or action.condition.exception is not None:
|
||||
action.done = action.condition.done
|
||||
action.exception = action.condition.exception
|
||||
break
|
||||
|
||||
if not action.condition.ok:
|
||||
action.done = True
|
||||
break
|
||||
action.done = True
|
||||
|
||||
def download(self:Self, action:Download, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
||||
if action.key or action.path:
|
||||
|
||||
19
Python/AnP/Models/SeleniumTabModel.py
Normal file
19
Python/AnP/Models/SeleniumTabModel.py
Normal file
@ -0,0 +1,19 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from typing import Self, Union
|
||||
from seleniumwire import webdriver
|
||||
from selenium.webdriver.remote.webdriver import WebDriver
|
||||
|
||||
WireWebDriver = Union[WebDriver, webdriver.Firefox]
|
||||
|
||||
class SeleniumTabModel:
|
||||
|
||||
def __init__(self:Self, key:str, url:str) -> None:
|
||||
self.key:str = key
|
||||
self.selector:WireWebDriver|None = None
|
||||
self.tabs:dict[str, Any] = {}
|
||||
|
||||
def set(self:Self, data:str) -> None:
|
||||
self.data = data
|
||||
# self.selector = Selector(text = data, type = "xml")
|
||||
Loading…
Reference in New Issue
Block a user