#wip(py): Selenium Driver.
This commit is contained in:
parent
26a06d93bb
commit
c5b945efca
@ -7,10 +7,14 @@ 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.common.service import Service
|
||||
from selenium.webdriver.common.options import ArgOptions as Options
|
||||
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 selenium.webdriver.firefox.service import Service as ServiceFirefox
|
||||
from selenium.webdriver.firefox.options import Options as OptionsFirefox
|
||||
from AnP.DSLs.ScrapDSL import (
|
||||
Open, Close, Switch, Go, Selector, XPath, CSS, Name, LinkText, Class, ID,
|
||||
Get, Download, Action, Click, Move, Position, MouseAction
|
||||
@ -26,19 +30,26 @@ class SeleniumDriver(BrowserAbstract):
|
||||
|
||||
def __init__(self:Self, anp:AnPInterface, key:str, inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None) -> None:
|
||||
super().__init__(anp, key, inputs)
|
||||
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 is None or self.browser.quit()
|
||||
self.tabs.clear()
|
||||
self.tab_selected = ""
|
||||
self.__cache_directory:str = self.anp.settings.get(("selenium_cache_directory", "cache_directory"), inputs)
|
||||
self.__download_timeout:int = self.anp.settings.get(("selenium_download_timeout", "download_timeout"), inputs, 60)
|
||||
self.__driver_type:str = self.anp.settings.get(("selenium_driver_type", "driver_type"), inputs, "firefox")
|
||||
self.__driver_path:str = self.anp.settings.get(("selenium_driver_path", "driver_path"), inputs)
|
||||
self.__binary_path:str = self.anp.settings.get(("selenium_binary_path", "binary_path"), inputs)
|
||||
self.__downloads_path:str = self.anp.settings.get(("selenium_downloads_path", "downloads_path"), inputs)
|
||||
self.__mimetypes_for_download:Sequence[str]|None = self.anp.settings.get(("selenium_mimetypes_for_download", "mimetypes_for_download"), inputs)
|
||||
self.__headless:bool = self.anp.settings.get(("selenium_headless", "headless"), inputs, True)
|
||||
self.__domains:Sequence[str]|None = self.anp.settings.get(("selenium_domains", "domains"), inputs)
|
||||
|
||||
self.anp.files.prepare_directory(self.__cache_directory)
|
||||
self.anp.files.prepare_directory(self.__downloads_path)
|
||||
|
||||
def __load(self:Self,
|
||||
action:Action,
|
||||
key:str|None,
|
||||
raw_url:str|Callable[[dict[str, Any|None]], str],
|
||||
shared:dict[str, Any|None]
|
||||
shared:dict[str, Any|None],
|
||||
box:BrowserBoxModel
|
||||
) -> None:
|
||||
|
||||
url:str = (
|
||||
@ -50,10 +61,12 @@ class SeleniumDriver(BrowserAbstract):
|
||||
if url is not None:
|
||||
try:
|
||||
|
||||
self.browser.get(url)
|
||||
browser:WebDriver = box.get_base()
|
||||
|
||||
browser.get(url)
|
||||
|
||||
self.browser.set_script_timeout(10)
|
||||
self.browser.execute_async_script("""
|
||||
browser.set_script_timeout(10)
|
||||
browser.execute_async_script("""
|
||||
if(document.readyState == "complete")
|
||||
arguments[arguments.length - 1](200);
|
||||
else
|
||||
@ -67,26 +80,95 @@ class SeleniumDriver(BrowserAbstract):
|
||||
else:
|
||||
action.done = False
|
||||
|
||||
def __set_selected(self:Self, key:str, box:BrowserBoxModel, id:str) -> None:
|
||||
box.variables[key] = id
|
||||
box.variables["__selected__"] = key
|
||||
@staticmethod
|
||||
def __interceptor_proxy(request:Request, domains:tuple[str, ...]) -> None:
|
||||
|
||||
domain:str = request.url.split("/")[2]
|
||||
|
||||
def __get_selected(self:Self, key:str, box:BrowserBoxModel) -> str|None:
|
||||
if "__selected__" in box.variables and box.variables["__selected__"] in box.variables:
|
||||
return box.variables[key]
|
||||
return None
|
||||
if domain not in domains:
|
||||
request.abort()
|
||||
|
||||
def __build_firefox(self:Self) -> tuple[type[WebDriver], OptionsFirefox, ServiceFirefox]:
|
||||
|
||||
driver:type[webdriver.Firefox] = webdriver.Firefox
|
||||
service:ServiceFirefox = ServiceFirefox(executable_path = self.__driver_path) if self.__driver_path else ServiceFirefox()
|
||||
options:OptionsFirefox = OptionsFirefox()
|
||||
|
||||
options.set_preference("browser.download.folderList", 2)
|
||||
options.set_preference("browser.download.manager.showWhenStarting", False)
|
||||
self.__downloads_path and options.set_preference("browser.download.dir", self.__downloads_path)
|
||||
options.set_preference("browser.download.manager.showWhenStarting", False)
|
||||
options.set_preference("browser.download.useDownloadDir", True)
|
||||
options.set_preference("pdfjs.disabled", True)
|
||||
self.__mimetypes_for_download and options.set_preference(
|
||||
"browser.helperApps.neverAsk.saveToDisk",
|
||||
",".join(self.__mimetypes_for_download)
|
||||
)
|
||||
options.set_preference("dom.block_multiple_popups", False)
|
||||
options.set_preference("dom.disable_open_during_load", False)
|
||||
|
||||
if not self.__binary_path and self.anp.files.exists("/home"): # For being looking for local profiles for Snap Firefox.
|
||||
|
||||
user:str
|
||||
done:bool = False
|
||||
|
||||
for user in self.anp.files.list_directory_items("/home"):
|
||||
if (
|
||||
user not in ("..", ".", "lost+found") and
|
||||
self.anp.files.is_directory(f"/home/{user}") and
|
||||
self.anp.files.exists(f"/home/{user}/snap/firefox/common/.mozilla/firefox")
|
||||
):
|
||||
|
||||
profile:str
|
||||
|
||||
for profile in self.anp.files.list_directory_items(f"/home/{user}/snap/firefox/common/.mozilla/firefox"):
|
||||
if profile.endswith(".default"):
|
||||
print(f"Using Firefox profile: /home/{user}/snap/firefox/common/.mozilla/firefox/{profile}")
|
||||
options.add_argument(f"-profile")
|
||||
options.add_argument(f"/home/{user}/snap/firefox/common/.mozilla/firefox/{profile}")
|
||||
done = True
|
||||
break
|
||||
|
||||
if done:
|
||||
break
|
||||
|
||||
return driver, service, options
|
||||
|
||||
def open(self:Self, action:Open, shared:dict[str, Any|None], box:BrowserBoxModel) -> None:
|
||||
try:
|
||||
if self.browser is None:
|
||||
self.browser = webdriver.Firefox()
|
||||
box.set_base(self.browser)
|
||||
|
||||
browser:WebDriver|None = box.get_base()
|
||||
|
||||
if browser is None:
|
||||
|
||||
driver:type[WebDriver]
|
||||
service:Service
|
||||
options:Options
|
||||
|
||||
if self.__driver_type == "firefox":
|
||||
driver, service, options = self.__build_firefox()
|
||||
|
||||
if self.__binary_path:
|
||||
options.binary_location = self.__binary_path
|
||||
|
||||
if self.__headless:
|
||||
options.add_argument("--headless")
|
||||
self.__domains or options.add_argument("--ignore-certificate-errors")
|
||||
|
||||
box.set_base(driver(service = service, options = options))
|
||||
browser = box.get_base()
|
||||
|
||||
if self.__domains:
|
||||
if self.__driver_type == "firefox":
|
||||
browser.request_interceptor = lambda request:self.__interceptor_proxy(request, self.__domains)
|
||||
|
||||
else:
|
||||
self.browser.switch_to.new_window("tab")
|
||||
box.set(action.key, self.browser)
|
||||
box.variables[action.key] = self.browser.current_window_handle
|
||||
box.variables["__selected__"] = action.key
|
||||
self.__load(action, action.url, shared, box)
|
||||
browser.switch_to.new_window("tab")
|
||||
|
||||
box.set(action.key, browser.current_window_handle)
|
||||
|
||||
self.__load(action, action.key, action.url, shared, box)
|
||||
|
||||
except Exception as exception:
|
||||
action.done = False
|
||||
action.exception = exception
|
||||
@ -94,13 +176,18 @@ class SeleniumDriver(BrowserAbstract):
|
||||
def close(self:Self, action:Close, shared:dict[str, Any|None], box:BrowserBoxModel) -> None:
|
||||
try:
|
||||
|
||||
current:str = box.get_last()
|
||||
tab:str = box.get(action.key)
|
||||
|
||||
self.browser.switch_to.window(tab)
|
||||
self.browser.close()
|
||||
box.close(tab)
|
||||
self.browser.switch_to.window(current)
|
||||
current:str|None = box.get(box.get_selected())
|
||||
tab:str|None = box.get(action.key)
|
||||
browser:WebDriver = box.get_base()
|
||||
|
||||
|
||||
if tab is not None:
|
||||
if current != tab:
|
||||
browser.switch_to.window(tab)
|
||||
browser.close()
|
||||
box.close(tab)
|
||||
if len(box.tabs):
|
||||
browser.switch_to.window(current if current != tab else box.get(box.get_selected()))
|
||||
|
||||
action.done = True
|
||||
|
||||
@ -113,21 +200,27 @@ class SeleniumDriver(BrowserAbstract):
|
||||
tab:str = box.get(action.key)
|
||||
|
||||
if tab is not None:
|
||||
self.browser.switch_to.window(tab)
|
||||
self.switch(tab)
|
||||
|
||||
browser:WebDriver = box.get_base()
|
||||
|
||||
browser.switch_to.window(tab)
|
||||
box.switch(tab)
|
||||
action.done = True
|
||||
|
||||
def go(self:Self, action:Go, shared:dict[str, Any|None], box:BrowserBoxModel) -> None:
|
||||
self.__load(action, action.url, shared)
|
||||
self.__load(action, action.key, action.url, shared, box)
|
||||
|
||||
def __move_to(self:Self, element:WebElement|Position|MouseAction, box:BrowserBoxModel) -> None:
|
||||
|
||||
browser:WebDriver = box.get_base()
|
||||
|
||||
def __move_to(self:Self, element:WebElement|Position|MouseAction) -> None:
|
||||
if isinstance(element, WebElement):
|
||||
self.browser.execute_script("arguments[0].scrollIntoView({block: 'center'});", element)
|
||||
ActionChains(self.browser).move_to_element(element).perform()
|
||||
browser.execute_script("arguments[0].scrollIntoView({block: 'center'});", element)
|
||||
ActionChains(browser).move_to_element(element).perform()
|
||||
elif isinstance(element, Position):
|
||||
ActionChains(self.browser).move_by_offset(element.x, element.y).perform()
|
||||
ActionChains(browser).move_by_offset(element.x, element.y).perform()
|
||||
elif isinstance(element, MouseAction):
|
||||
ActionChains(self.browser).move_by_offset(*element.get()).perform()
|
||||
ActionChains(browser).move_by_offset(*element.get()).perform()
|
||||
|
||||
def get_selector(self:Self,
|
||||
selector:Selector,
|
||||
@ -144,7 +237,7 @@ class SeleniumDriver(BrowserAbstract):
|
||||
By.PARTIAL_LINK_TEXT if isinstance(selector, LinkText) else
|
||||
By.NAME if isinstance(selector, Name) else
|
||||
By.CSS_SELECTOR)
|
||||
from_item:WebElement|WebDriver = box.get_base()
|
||||
from_item:WebElement|WebDriver = box.get_last()
|
||||
|
||||
if force_multiple or selector.multiple:
|
||||
return from_item.find_elements(mode, selector.selector)
|
||||
@ -154,7 +247,7 @@ class SeleniumDriver(BrowserAbstract):
|
||||
def get(self:Self, action:Get, shared:dict[str, Any|None], box:BrowserBoxModel) -> None:
|
||||
|
||||
last_item:WebElement|None = box.get_last()
|
||||
item:WebElement = last_item or self.browser if action.selector is None else self.get_selector(action.selector, last_item)
|
||||
item:WebElement = last_item or box.get_base() if action.selector is None else self.get_selector(action.selector, box)
|
||||
multiple:bool = action.selector is not None and action.selector.multiple
|
||||
|
||||
if action.key not in shared or not isinstance(shared[action.key], list):
|
||||
@ -165,14 +258,14 @@ class SeleniumDriver(BrowserAbstract):
|
||||
element:WebElement
|
||||
|
||||
for element in item:
|
||||
self.__move_to(element)
|
||||
self.__move_to(element, box)
|
||||
# ActionChains(self.__browser).move_to_element(element).perform()
|
||||
shared[action.key].append(element.get_attribute(
|
||||
"textContent" if action.attribute == "CDATA" else
|
||||
action.attribute))
|
||||
|
||||
else:
|
||||
self.__move_to(item)
|
||||
self.__move_to(item, box)
|
||||
# ActionChains(self.__browser).move_to_element(item).perform()
|
||||
shared[action.key] = (item.get_attribute(
|
||||
"textContent" if action.attribute == "CDATA" else
|
||||
@ -184,6 +277,7 @@ class SeleniumDriver(BrowserAbstract):
|
||||
|
||||
request:Request
|
||||
date:float = timestamp()
|
||||
browser:WebDriver = box.get_base()
|
||||
|
||||
while (
|
||||
timestamp() - date < (timeout or self.__download_timeout) and
|
||||
@ -193,7 +287,7 @@ class SeleniumDriver(BrowserAbstract):
|
||||
return None
|
||||
sleep(.1)
|
||||
|
||||
for request in self.browser.requests:
|
||||
for request in browser.requests:
|
||||
if request.url == url and request.response:
|
||||
return sw_decode(
|
||||
request.response.body,
|
||||
@ -205,8 +299,9 @@ class SeleniumDriver(BrowserAbstract):
|
||||
|
||||
l:int = len(self.browser.window_handles)
|
||||
data:bytes|None = None
|
||||
browser:WebDriver = box.get_base()
|
||||
|
||||
self.browser.execute_script("""
|
||||
browser.execute_script("""
|
||||
|
||||
const anchor = document.createElement("a");
|
||||
|
||||
@ -220,15 +315,18 @@ class SeleniumDriver(BrowserAbstract):
|
||||
|
||||
data = self.__download_from_cache(url, box)
|
||||
|
||||
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(box.get_last())
|
||||
if l < len(browser.window_handles):
|
||||
browser.switch_to.window(browser.window_handles[-1])
|
||||
browser.close()
|
||||
browser.switch_to.window(box.get(box.get_selected()))
|
||||
|
||||
return data
|
||||
|
||||
def __download_by_url(self:Self, url:str, box:BrowserBoxModel) -> bytes|None:
|
||||
return self.browser.get(url)
|
||||
|
||||
browser:WebDriver = box.get_base()
|
||||
|
||||
return browser.get(url)
|
||||
|
||||
def download(self:Self, action:Download, shared:dict[str, Any|None], box:BrowserBoxModel) -> None:
|
||||
|
||||
@ -270,8 +368,12 @@ class SeleniumDriver(BrowserAbstract):
|
||||
if isinstance(action.position, Sequence) and len(action.position) == 2 and all(isinstance(coord, (int, float)) for coord in action.position):
|
||||
action.position = Position(action.position[0], action.position[1])
|
||||
if isinstance(action.position, Position):
|
||||
ActionChains(self.browser).move_by_offset(action.position.x, action.position.y).perform()
|
||||
|
||||
browser:WebDriver = box.get_base()
|
||||
|
||||
ActionChains(browser).move_by_offset(action.position.x, action.position.y).perform()
|
||||
action.done = True
|
||||
|
||||
else:
|
||||
action.done = False
|
||||
else:
|
||||
@ -279,7 +381,7 @@ class SeleniumDriver(BrowserAbstract):
|
||||
Position(element.location["x"], element.location["y"]),
|
||||
action.margin,
|
||||
Position(element.size["width"], element.size["height"])
|
||||
))
|
||||
), box)
|
||||
action.done = True
|
||||
|
||||
def click(self:Self, action:Click, shared:dict[str, Any|None], box:BrowserBoxModel) -> None:
|
||||
@ -288,7 +390,11 @@ class SeleniumDriver(BrowserAbstract):
|
||||
|
||||
self.move(move_action, shared, box)
|
||||
if move_action.done:
|
||||
ActionChains(self.browser).click().perform()
|
||||
|
||||
browser:WebDriver = box.get_base()
|
||||
|
||||
ActionChains(browser).click().perform()
|
||||
action.done = True
|
||||
|
||||
else:
|
||||
action.done = False
|
||||
@ -43,6 +43,11 @@ class BrowserBoxModel:
|
||||
self.tabs[key] = BrowserTabsModel(item)
|
||||
self.tab_selected = key
|
||||
|
||||
def get(self:Self, key:str) -> Any|None:
|
||||
if key in self.tabs:
|
||||
return self.tabs[key].item
|
||||
return None
|
||||
|
||||
def change(self:Self, key:str) -> bool:
|
||||
if key in self.tabs:
|
||||
self.tab_selected = key
|
||||
|
||||
Loading…
Reference in New Issue
Block a user