81 lines
3.0 KiB
Python
81 lines
3.0 KiB
Python
import random
|
|
import string
|
|
import typing
|
|
|
|
from selenium.common.exceptions import StaleElementReferenceException, TimeoutException
|
|
from selenium.webdriver.chrome.webdriver import WebDriver
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
from selenium.webdriver.support.wait import WebDriverWait
|
|
from selenium.webdriver.remote.webelement import WebElement
|
|
|
|
driver: WebDriver|None = None
|
|
attempts, timeout, interval = int(), float(), float()
|
|
|
|
def until(condition: typing.Callable[[WebDriver], bool], watch=True):
|
|
try:
|
|
WebDriverWait(driver, timeout).until(condition)
|
|
except (TimeoutException, StaleElementReferenceException):
|
|
pass
|
|
if watch: WebDriverWait(driver, timeout).until_not(condition)
|
|
|
|
def sleep(seconds: float):
|
|
from common.jsonrpc2.server import connection
|
|
assert connection is not None and connection.closed is None
|
|
|
|
try: WebDriverWait(driver, seconds, seconds).until(lambda _: False)
|
|
except: pass
|
|
|
|
def locate(selector: str, wait=True, condition=None) -> WebElement:
|
|
while True:
|
|
try:
|
|
locator = (By.CSS_SELECTOR, selector)
|
|
if not wait: return driver.find_element(*locator)
|
|
|
|
presence = EC.presence_of_element_located(locator)
|
|
element = WebDriverWait(driver, timeout).until(presence, 'Timeout')
|
|
driver.execute_script("arguments[0].scrollIntoView({ block: 'center', inline: 'nearest' });", element)
|
|
|
|
if condition is not None:
|
|
predicate = condition or EC.visibility_of_element_located
|
|
element = WebDriverWait(driver, timeout).until(predicate(locator), 'Timeout')
|
|
|
|
return element
|
|
except StaleElementReferenceException:
|
|
pass
|
|
|
|
def click(selector: str|WebElement, wait=True, condition=None):
|
|
predicate = condition if condition is not None else EC.element_to_be_clickable
|
|
identity = ''.join(random.choices(string.ascii_uppercase + string.digits, k=8))
|
|
element = locate(selector, wait, predicate) if isinstance(selector, str) else selector
|
|
counter = lambda: int(element.get_attribute(identity) or 0)
|
|
error = False
|
|
|
|
value = counter()
|
|
driver.execute_script("arguments[0].addEventListener('click', () => arguments[0].setAttribute(arguments[1], arguments[2] + 1));", element, identity, value)
|
|
|
|
for _ in range(attempts):
|
|
try:
|
|
if not error: element.click()
|
|
else: driver.execute_script("arguments[0].click();", element)
|
|
except StaleElementReferenceException:
|
|
break
|
|
except:
|
|
error = True
|
|
continue
|
|
try:
|
|
WebDriverWait(driver, interval).until(lambda _: counter() > value)
|
|
break
|
|
except TimeoutException: continue
|
|
except: break
|
|
|
|
def setup(a: WebDriver, b: int, c: float, d: float):
|
|
global driver, attempts, timeout, interval
|
|
driver = a
|
|
attempts = b
|
|
timeout = c
|
|
interval = d
|
|
|
|
from common.jsonrpc2.server import connection
|
|
until(lambda _: connection is not None, watch=False)
|