From b62c3cb6a1b4b24566ad316f6b449bc25461e26a Mon Sep 17 00:00:00 2001 From: g1879 Date: Thu, 29 Jun 2023 17:14:25 +0800 Subject: [PATCH] =?UTF-8?q?Waiter=E9=87=8D=E6=9E=84=E5=88=B0=E4=B8=93?= =?UTF-8?q?=E5=B1=9E=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- DrissionPage/chromium_base.py | 101 +------------ DrissionPage/chromium_base.pyi | 26 +--- DrissionPage/chromium_element.py | 111 +------------- DrissionPage/chromium_element.pyi | 27 +--- DrissionPage/chromium_frame.py | 14 +- DrissionPage/chromium_frame.pyi | 9 +- DrissionPage/chromium_page.py | 30 +--- DrissionPage/chromium_page.pyi | 119 +++++++-------- DrissionPage/waiter.py | 242 ++++++++++++++++++++++++++++++ DrissionPage/waiter.pyi | 78 ++++++++++ 10 files changed, 387 insertions(+), 370 deletions(-) create mode 100644 DrissionPage/waiter.py create mode 100644 DrissionPage/waiter.pyi diff --git a/DrissionPage/chromium_base.py b/DrissionPage/chromium_base.py index abc4257..a046cc1 100644 --- a/DrissionPage/chromium_base.py +++ b/DrissionPage/chromium_base.py @@ -15,14 +15,15 @@ from requests import Session from .base import BasePage from .chromium_driver import ChromiumDriver from .chromium_element import ChromiumScroll, ChromiumElement, run_js, make_chromium_ele -from .commons.constants import HANDLE_ALERT_METHOD, ERROR, NoneElement, Settings +from .commons.constants import HANDLE_ALERT_METHOD, ERROR, NoneElement from .commons.locator import get_loc from .commons.tools import get_usable_path, clean_folder from .commons.web import set_browser_cookies from .errors import ContextLossError, ElementLossError, AlertExistsError, CDPError, TabClosedError, \ - NoRectError, BrowserConnectError, GetDocumentError, WaitTimeoutError + NoRectError, BrowserConnectError, GetDocumentError from .network_listener import NetworkListener from .session_element import make_session_ele +from .waiter import ChromiumBaseWaiter class ChromiumBase(BasePage): @@ -995,102 +996,6 @@ class ChromiumBaseSetter(object): self._page.run_cdp('Network.setExtraHTTPHeaders', headers=headers) -class ChromiumBaseWaiter(object): - def __init__(self, page_or_ele): - """ - :param page_or_ele: 页面对象或元素对象 - """ - self._driver = page_or_ele - - def ele_delete(self, loc_or_ele, timeout=None, raise_err=None): - """等待元素从DOM中删除 - :param loc_or_ele: 要等待的元素，可以是已有元素、定位符 - :param timeout: 超时时间，默认读取页面超时时间 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - ele = self._driver._ele(loc_or_ele, raise_err=False, timeout=0) - return ele.wait.delete(timeout, raise_err=raise_err) if ele else True - - def ele_display(self, loc_or_ele, timeout=None, raise_err=None): - """等待元素变成显示状态 - :param loc_or_ele: 要等待的元素，可以是已有元素、定位符 - :param timeout: 超时时间，默认读取页面超时时间 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - ele = self._driver._ele(loc_or_ele, raise_err=False, timeout=0) - return ele.wait.display(timeout, raise_err=raise_err) - - def ele_hidden(self, loc_or_ele, timeout=None, raise_err=None): - """等待元素变成隐藏状态 - :param loc_or_ele: 要等待的元素，可以是已有元素、定位符 - :param timeout: 超时时间，默认读取页面超时时间 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - ele = self._driver._ele(loc_or_ele, raise_err=False, timeout=0) - return ele.wait.hidden(timeout, raise_err=raise_err) - - def ele_load(self, loc, timeout=None, raise_err=None): - """等待元素加载到DOM - :param loc: 要等待的元素，输入定位符 - :param timeout: 超时时间，默认读取页面超时时间 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 成功返回元素对象，失败返回False - """ - ele = self._driver._ele(loc, raise_err=False, timeout=timeout) - if ele: - return True - if raise_err is True or Settings.raise_when_wait_failed is True: - raise WaitTimeoutError('等待元素加载失败。') - else: - return False - - def load_start(self, timeout=None, raise_err=None): - """等待页面开始加载 - :param timeout: 超时时间，为None时使用页面timeout属性 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - return self._loading(timeout=timeout, gap=.002, raise_err=raise_err) - - def load_complete(self, timeout=None, raise_err=None): - """等待页面开始加载 - :param timeout: 超时时间，为None时使用页面timeout属性 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - return self._loading(timeout=timeout, start=False, raise_err=raise_err) - - def upload_paths_inputted(self): - """等待自动填写上传文件路径""" - while self._driver._upload_list: - sleep(.01) - - def _loading(self, timeout=None, start=True, gap=.01, raise_err=None): - """等待页面开始加载或加载完成 - :param timeout: 超时时间，为None时使用页面timeout属性 - :param start: 等待开始还是结束 - :param gap: 间隔秒数 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - if timeout != 0: - if timeout is None or timeout is True: - timeout = self._driver.timeout - end_time = perf_counter() + timeout - while perf_counter() < end_time: - if self._driver.is_loading == start: - return True - sleep(gap) - - if raise_err is True or Settings.raise_when_wait_failed is True: - raise WaitTimeoutError('等待页面加载失败。') - else: - return False - - class ChromiumPageScroll(ChromiumScroll): def __init__(self, page): """ diff --git a/DrissionPage/chromium_base.pyi b/DrissionPage/chromium_base.pyi index 7b275a8..160609d 100644 --- a/DrissionPage/chromium_base.pyi +++ b/DrissionPage/chromium_base.pyi @@ -10,6 +10,7 @@ from DataRecorder import Recorder from requests import Session from requests.cookies import RequestsCookieJar +from .waiter import ChromiumBaseWaiter from .base import BasePage from .chromium_driver import ChromiumDriver from .chromium_element import ChromiumElement, ChromiumScroll @@ -213,31 +214,6 @@ class ChromiumBase(BasePage): timeout: float = None) -> Union[bool, None]: ... -class ChromiumBaseWaiter(object): - def __init__(self, page: ChromiumBase): - self._driver: ChromiumBase = ... - - def ele_delete(self, loc_or_ele: Union[str, tuple, ChromiumElement], timeout: float = None, - raise_err: bool = None) -> bool: ... - - def ele_display(self, loc_or_ele: Union[str, tuple, ChromiumElement], timeout: float = None, - raise_err: bool = None) -> bool: ... - - def ele_hidden(self, loc_or_ele: Union[str, tuple, ChromiumElement], timeout: float = None, - raise_err: bool = None) -> bool: ... - - def ele_load(self, loc: Union[str, tuple], timeout: float = None, - raise_err: bool = None) -> Union[bool, ChromiumElement]: ... - - def _loading(self, timeout: float = None, start: bool = True, gap: float = .01, raise_err: bool = None) -> bool: ... - - def load_start(self, timeout: float = None, raise_err: bool = None) -> bool: ... - - def load_complete(self, timeout: float = None, raise_err: bool = None) -> bool: ... - - def upload_paths_inputted(self) -> None: ... - - class ChromiumPageScroll(ChromiumScroll): def __init__(self, page: ChromiumBase): ... diff --git a/DrissionPage/chromium_element.py b/DrissionPage/chromium_element.py index 03fb5f9..68b63fe 100644 --- a/DrissionPage/chromium_element.py +++ b/DrissionPage/chromium_element.py @@ -14,8 +14,9 @@ from .commons.keys import keys_to_typing, keyDescriptionForString, keyDefinition from .commons.locator import get_loc from .commons.web import make_absolute_link, get_ele_txt, format_html, is_js_func, location_in_viewport, offset_scroll from .errors import ContextLossError, ElementLossError, JavaScriptError, NoRectError, ElementNotFoundError, \ - CDPError, NoResourceError, CanNotClickError, WaitTimeoutError + CDPError, NoResourceError, CanNotClickError from .session_element import make_session_ele +from .waiter import ChromiumElementWaiter class ChromiumElement(DrissionElement): @@ -2028,114 +2029,6 @@ class ChromiumSelect(object): self._ele.run_js('this.dispatchEvent(new UIEvent("change"));') -class ChromiumElementWaiter(object): - """等待元素在dom中某种状态，如删除、显示、隐藏""" - - def __init__(self, page, ele): - """等待元素在dom中某种状态，如删除、显示、隐藏 - :param page: 元素所在页面 - :param ele: 要等待的元素 - """ - self._page = page - self._ele = ele - - def delete(self, timeout=None, raise_err=None): - """等待元素从dom删除 - :param timeout: 超时时间，为None使用元素所在页面timeout属性 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - return self._wait_state('is_alive', False, timeout, raise_err) - - def display(self, timeout=None, raise_err=None): - """等待元素从dom显示 - :param timeout: 超时时间，为None使用元素所在页面timeout属性 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - return self._wait_state('is_displayed', True, timeout, raise_err) - - def hidden(self, timeout=None, raise_err=None): - """等待元素从dom隐藏 - :param timeout: 超时时间，为None使用元素所在页面timeout属性 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - return self._wait_state('is_displayed', False, timeout, raise_err) - - def covered(self, timeout=None, raise_err=None): - """等待当前元素被遮盖 - :param timeout:超时时间，为None使用元素所在页面timeout属性 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - return self._wait_state('is_covered', True, timeout, raise_err) - - def not_covered(self, timeout=None, raise_err=None): - """等待当前元素被遮盖 - :param timeout:超时时间，为None使用元素所在页面timeout属性 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - return self._wait_state('is_covered', False, timeout, raise_err) - - def enabled(self, timeout=None, raise_err=None): - """等待当前元素变成可用 - :param timeout:超时时间，为None使用元素所在页面timeout属性 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - return self._wait_state('is_enabled', True, timeout, raise_err) - - def disabled(self, timeout=None, raise_err=None): - """等待当前元素变成可用 - :param timeout:超时时间，为None使用元素所在页面timeout属性 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - return self._wait_state('is_enabled', False, timeout, raise_err) - - def disabled_or_delete(self, timeout=None, raise_err=None): - """等待当前元素变成不可用或从DOM移除 - :param timeout:超时时间，为None使用元素所在页面timeout属性 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - if timeout is None: - timeout = self._page.timeout - end_time = perf_counter() + timeout - while perf_counter() < end_time: - if not self._ele.states.is_enabled or not self._ele.states.is_alive: - return True - sleep(.05) - - if raise_err is True or Settings.raise_when_wait_failed is True: - raise WaitTimeoutError('等待元素隐藏或删除失败。') - else: - return False - - def _wait_state(self, attr, mode=False, timeout=None, raise_err=None): - """等待元素某个bool状态到达指定状态 - :param attr: 状态名称 - :param mode: True或False - :param timeout: 超时时间，为None使用元素所在页面timeout属性 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等待成功 - """ - if timeout is None: - timeout = self._page.timeout - end_time = perf_counter() + timeout - while perf_counter() < end_time: - if self._ele.states.__getattribute__(attr) == mode: - return True - sleep(.05) - - if raise_err is True or Settings.raise_when_wait_failed is True: - raise WaitTimeoutError('等待元素状态改变失败。') - else: - return False - - class Pseudo(object): def __init__(self, ele): """ diff --git a/DrissionPage/chromium_element.pyi b/DrissionPage/chromium_element.pyi index a218d56..ae78c42 100644 --- a/DrissionPage/chromium_element.pyi +++ b/DrissionPage/chromium_element.pyi @@ -12,6 +12,7 @@ from .chromium_frame import ChromiumFrame from .chromium_page import ChromiumPage from .commons.constants import NoneElement from .session_element import SessionElement +from .waiter import ChromiumElementWaiter from .web_page import WebPage @@ -554,32 +555,6 @@ class ChromiumSelect(object): def _dispatch_change(self) -> None: ... -class ChromiumElementWaiter(object): - def __init__(self, - page: ChromiumBase, - ele: ChromiumElement): - self._ele: ChromiumElement = ... - self._page: ChromiumBase = ... - - def delete(self, timeout: float = None, raise_err: bool = None) -> bool: ... - - def display(self, timeout: float = None, raise_err: bool = None) -> bool: ... - - def hidden(self, timeout: float = None, raise_err: bool = None) -> bool: ... - - def covered(self, timeout: float = None, raise_err: bool = None) -> bool: ... - - def not_covered(self, timeout: float = None, raise_err: bool = None) -> bool: ... - - def enabled(self, timeout: float = None, raise_err: bool = None) -> bool: ... - - def disabled(self, timeout: float = None, raise_err: bool = None) -> bool: ... - - def disabled_or_delete(self, timeout: float = None, raise_err: bool = None) -> bool: ... - - def _wait_state(self, attr: str, mode: bool = False, timeout: float = None, raise_err: bool = None) -> bool: ... - - class Pseudo(object): def __init__(self, ele: ChromiumElement): self._ele: ChromiumElement = ... diff --git a/DrissionPage/chromium_frame.py b/DrissionPage/chromium_frame.py index 6a9d70c..65a76ae 100644 --- a/DrissionPage/chromium_frame.py +++ b/DrissionPage/chromium_frame.py @@ -7,10 +7,11 @@ from re import search from threading import Thread from time import sleep, perf_counter -from .chromium_base import ChromiumBase, ChromiumPageScroll, ChromiumBaseSetter, ChromiumBaseWaiter -from .chromium_element import ChromiumElement, ChromiumElementWaiter +from .chromium_base import ChromiumBase, ChromiumPageScroll, ChromiumBaseSetter +from .chromium_element import ChromiumElement from .commons.tools import get_usable_path from .errors import ContextLossError +from .waiter import FrameWaiter class ChromiumFrame(ChromiumBase): @@ -660,12 +661,3 @@ class ChromiumFrameSetter(ChromiumBaseSetter): """ self._page._check_ok() self._page.frame_ele.set.attr(attr, value) - - -class FrameWaiter(ChromiumBaseWaiter, ChromiumElementWaiter): - def __init__(self, frame): - """ - :param frame: ChromiumFrame对象 - """ - super().__init__(frame) - super(ChromiumBaseWaiter, self).__init__(frame, frame.frame_ele) diff --git a/DrissionPage/chromium_frame.pyi b/DrissionPage/chromium_frame.pyi index 47dc8a1..0f56dab 100644 --- a/DrissionPage/chromium_frame.pyi +++ b/DrissionPage/chromium_frame.pyi @@ -6,8 +6,9 @@ from pathlib import Path from typing import Union, Tuple, List, Any -from .chromium_base import ChromiumBase, ChromiumPageScroll, ChromiumBaseSetter, ChromiumBaseWaiter -from .chromium_element import ChromiumElement, Locations, ChromiumElementStates, ChromiumElementWaiter +from .waiter import FrameWaiter +from .chromium_base import ChromiumBase, ChromiumPageScroll, ChromiumBaseSetter +from .chromium_element import ChromiumElement, Locations, ChromiumElementStates class ChromiumFrame(ChromiumBase): @@ -210,7 +211,3 @@ class ChromiumFrameSetter(ChromiumBaseSetter): _page: ChromiumFrame = ... def attr(self, attr: str, value: str) -> None: ... - - -class FrameWaiter(ChromiumBaseWaiter, ChromiumElementWaiter): - def __init__(self, frame: ChromiumFrame): ... diff --git a/DrissionPage/chromium_page.py b/DrissionPage/chromium_page.py index 81bde60..6dd7097 100644 --- a/DrissionPage/chromium_page.py +++ b/DrissionPage/chromium_page.py @@ -6,14 +6,14 @@ from platform import system from time import perf_counter, sleep -from .chromium_base import ChromiumBase, Timeout, ChromiumBaseSetter, ChromiumBaseWaiter +from .chromium_base import ChromiumBase, Timeout, ChromiumBaseSetter from .chromium_driver import ChromiumDriver from .chromium_tab import ChromiumTab from .commons.browser import connect_browser -from .commons.constants import Settings from .commons.tools import port_is_using from .configs.chromium_options import ChromiumOptions -from .errors import BrowserConnectError, WaitTimeoutError +from .errors import BrowserConnectError +from .waiter import ChromiumPageWaiter class ChromiumPage(ChromiumBase): @@ -363,30 +363,6 @@ class ChromiumPage(ChromiumBase): self._tab_obj.has_alert = True -class ChromiumPageWaiter(ChromiumBaseWaiter): - def __init__(self, page: ChromiumBase): - super().__init__(page) - self._listener = None - - def new_tab(self, timeout=None, raise_err=None): - """等待新标签页出现 - :param timeout: 等待超时时间，为None则使用页面对象timeout属性 - :param raise_err: 等待识别时是否报错，为None时根据Settings设置 - :return: 是否等到新标签页出现 - """ - timeout = timeout if timeout is not None else self._driver.timeout - end_time = perf_counter() + timeout - while perf_counter() < end_time: - if self._driver.tab_id != self._driver.latest_tab: - return True - sleep(.01) - - if raise_err is True or Settings.raise_when_wait_failed is True: - raise WaitTimeoutError('等待新标签页失败。') - else: - return False - - class ChromiumTabRect(object): def __init__(self, page): self._page = page diff --git a/DrissionPage/chromium_page.pyi b/DrissionPage/chromium_page.pyi index 2a9eda4..47115cb 100644 --- a/DrissionPage/chromium_page.pyi +++ b/DrissionPage/chromium_page.pyi @@ -4,19 +4,13 @@ @Contact : g1879@qq.com """ from os import popen -from pathlib import Path -from typing import Union, Tuple, List, Dict +from typing import Union, Tuple, List -from DownloadKit import DownloadKit -from DownloadKit.mission import Mission -from requests import Session - -from .chromium_base import ChromiumBase, ChromiumBaseSetter, ChromiumBaseWaiter +from .chromium_base import ChromiumBase, ChromiumBaseSetter from .chromium_driver import ChromiumDriver from .chromium_tab import ChromiumTab from .configs.chromium_options import ChromiumOptions -from .network_listener import NetworkListener -from .session_page import DownloadSetter +from .waiter import ChromiumPageWaiter class ChromiumPage(ChromiumBase): @@ -98,17 +92,6 @@ class ChromiumPage(ChromiumBase): def _on_alert_open(self, **kwargs): ... -class ChromiumPageWaiter(ChromiumBaseWaiter): - _driver: ChromiumPage = ... - _listener: Union[NetworkListener, None] = ... - - # def download_begin(self, timeout: float = 1.5) -> bool: ... - - # def download_finish(self, timeout: float = None) -> bool: ... - - def new_tab(self, timeout: float = None, raise_err: bool = None) -> bool: ... - - class ChromiumTabRect(object): def __init__(self, page: ChromiumPage): self._page: ChromiumPage = ... @@ -142,54 +125,54 @@ class ChromiumTabRect(object): def _get_browser_rect(self) -> dict: ... -class BaseDownloadSetter(DownloadSetter): - def __init__(self, page: ChromiumPage): - self._page: ChromiumPage = ... - self._behavior: str = ... - self._session: Session = ... - self._save_path: str = ... - self._rename: str = ... - self._waiting_download: bool = ... - self._download_begin: bool = ... - self._browser_missions: Dict[str, BrowserDownloadMission] = ... - self._browser_downloading_count: int = ... - self._show_msg: bool = ... - - @property - def session(self) -> Session: ... - - @property - def browser_missions(self) -> List[BrowserDownloadMission]: ... - - @property - def DownloadKit_missions(self) -> List[Mission]: ... - - @property - def _switched_DownloadKit(self) -> DownloadKit: ... - - def save_path(self, path: Union[str, Path]) -> None: ... - - def rename(self, name: str) -> None: ... - - def by_browser(self) -> None: ... - - def by_DownloadKit(self) -> None: ... - - def wait_download_begin(self, timeout: float = None) -> bool: ... - - def wait_download_finish(self, timeout: float = None) -> bool: ... - - def show_msg(self, on_off: bool = True) -> None: ... - - def _cookies_to_session(self) -> None: ... - - def _download_by_DownloadKit(self, **kwargs) -> None: ... - - def _download_will_begin(self, **kwargs) -> None: ... - - def _download_progress(self, **kwargs) -> None: ... - - def _wait_download_complete(self, mission: Mission) -> None: ... +# class BaseDownloadSetter(DownloadSetter): +# def __init__(self, page: ChromiumPage): +# self._page: ChromiumPage = ... +# self._behavior: str = ... +# self._session: Session = ... +# self._save_path: str = ... +# self._rename: str = ... +# self._waiting_download: bool = ... +# self._download_begin: bool = ... +# self._browser_missions: Dict[str, BrowserDownloadMission] = ... +# self._browser_downloading_count: int = ... +# self._show_msg: bool = ... +# +# @property +# def session(self) -> Session: ... +# +# @property +# def browser_missions(self) -> List[BrowserDownloadMission]: ... +# +# @property +# def DownloadKit_missions(self) -> List[Mission]: ... +# +# @property +# def _switched_DownloadKit(self) -> DownloadKit: ... +# +# def save_path(self, path: Union[str, Path]) -> None: ... +# +# def rename(self, name: str) -> None: ... +# +# def by_browser(self) -> None: ... +# +# def by_DownloadKit(self) -> None: ... +# +# def wait_download_begin(self, timeout: float = None) -> bool: ... +# +# def wait_download_finish(self, timeout: float = None) -> bool: ... +# +# def show_msg(self, on_off: bool = True) -> None: ... +# +# def _cookies_to_session(self) -> None: ... +# +# def _download_by_DownloadKit(self, **kwargs) -> None: ... +# +# def _download_will_begin(self, **kwargs) -> None: ... +# +# def _download_progress(self, **kwargs) -> None: ... +# +# def _wait_download_complete(self, mission: Mission) -> None: ... class BrowserDownloadMission(object): diff --git a/DrissionPage/waiter.py b/DrissionPage/waiter.py new file mode 100644 index 0000000..7bfd238 --- /dev/null +++ b/DrissionPage/waiter.py @@ -0,0 +1,242 @@ +# -*- coding:utf-8 -*- +from time import sleep, perf_counter + +from .commons.constants import Settings +from .errors import WaitTimeoutError + + +class ChromiumBaseWaiter(object): + def __init__(self, page_or_ele): + """ + :param page_or_ele: 页面对象或元素对象 + """ + self._driver = page_or_ele + + def ele_delete(self, loc_or_ele, timeout=None, raise_err=None): + """等待元素从DOM中删除 + :param loc_or_ele: 要等待的元素，可以是已有元素、定位符 + :param timeout: 超时时间，默认读取页面超时时间 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + ele = self._driver._ele(loc_or_ele, raise_err=False, timeout=0) + return ele.wait.delete(timeout, raise_err=raise_err) if ele else True + + def ele_display(self, loc_or_ele, timeout=None, raise_err=None): + """等待元素变成显示状态 + :param loc_or_ele: 要等待的元素，可以是已有元素、定位符 + :param timeout: 超时时间，默认读取页面超时时间 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + ele = self._driver._ele(loc_or_ele, raise_err=False, timeout=0) + return ele.wait.display(timeout, raise_err=raise_err) + + def ele_hidden(self, loc_or_ele, timeout=None, raise_err=None): + """等待元素变成隐藏状态 + :param loc_or_ele: 要等待的元素，可以是已有元素、定位符 + :param timeout: 超时时间，默认读取页面超时时间 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + ele = self._driver._ele(loc_or_ele, raise_err=False, timeout=0) + return ele.wait.hidden(timeout, raise_err=raise_err) + + def ele_load(self, loc, timeout=None, raise_err=None): + """等待元素加载到DOM + :param loc: 要等待的元素，输入定位符 + :param timeout: 超时时间，默认读取页面超时时间 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 成功返回元素对象，失败返回False + """ + ele = self._driver._ele(loc, raise_err=False, timeout=timeout) + if ele: + return True + if raise_err is True or Settings.raise_when_wait_failed is True: + raise WaitTimeoutError('等待元素加载失败。') + else: + return False + + def load_start(self, timeout=None, raise_err=None): + """等待页面开始加载 + :param timeout: 超时时间，为None时使用页面timeout属性 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + return self._loading(timeout=timeout, gap=.002, raise_err=raise_err) + + def load_complete(self, timeout=None, raise_err=None): + """等待页面开始加载 + :param timeout: 超时时间，为None时使用页面timeout属性 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + return self._loading(timeout=timeout, start=False, raise_err=raise_err) + + def upload_paths_inputted(self): + """等待自动填写上传文件路径""" + while self._driver._upload_list: + sleep(.01) + + def _loading(self, timeout=None, start=True, gap=.01, raise_err=None): + """等待页面开始加载或加载完成 + :param timeout: 超时时间，为None时使用页面timeout属性 + :param start: 等待开始还是结束 + :param gap: 间隔秒数 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + if timeout != 0: + if timeout is None or timeout is True: + timeout = self._driver.timeout + end_time = perf_counter() + timeout + while perf_counter() < end_time: + if self._driver.is_loading == start: + return True + sleep(gap) + + if raise_err is True or Settings.raise_when_wait_failed is True: + raise WaitTimeoutError('等待页面加载失败。') + else: + return False + + +class ChromiumPageWaiter(ChromiumBaseWaiter): + def __init__(self, page): + super().__init__(page) + # self._listener = None + + def new_tab(self, timeout=None, raise_err=None): + """等待新标签页出现 + :param timeout: 等待超时时间，为None则使用页面对象timeout属性 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等到新标签页出现 + """ + timeout = timeout if timeout is not None else self._driver.timeout + end_time = perf_counter() + timeout + while perf_counter() < end_time: + if self._driver.tab_id != self._driver.latest_tab: + return True + sleep(.01) + + if raise_err is True or Settings.raise_when_wait_failed is True: + raise WaitTimeoutError('等待新标签页失败。') + else: + return False + + +class ChromiumElementWaiter(object): + """等待元素在dom中某种状态，如删除、显示、隐藏""" + + def __init__(self, page, ele): + """等待元素在dom中某种状态，如删除、显示、隐藏 + :param page: 元素所在页面 + :param ele: 要等待的元素 + """ + self._page = page + self._ele = ele + + def delete(self, timeout=None, raise_err=None): + """等待元素从dom删除 + :param timeout: 超时时间，为None使用元素所在页面timeout属性 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + return self._wait_state('is_alive', False, timeout, raise_err) + + def display(self, timeout=None, raise_err=None): + """等待元素从dom显示 + :param timeout: 超时时间，为None使用元素所在页面timeout属性 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + return self._wait_state('is_displayed', True, timeout, raise_err) + + def hidden(self, timeout=None, raise_err=None): + """等待元素从dom隐藏 + :param timeout: 超时时间，为None使用元素所在页面timeout属性 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + return self._wait_state('is_displayed', False, timeout, raise_err) + + def covered(self, timeout=None, raise_err=None): + """等待当前元素被遮盖 + :param timeout:超时时间，为None使用元素所在页面timeout属性 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + return self._wait_state('is_covered', True, timeout, raise_err) + + def not_covered(self, timeout=None, raise_err=None): + """等待当前元素被遮盖 + :param timeout:超时时间，为None使用元素所在页面timeout属性 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + return self._wait_state('is_covered', False, timeout, raise_err) + + def enabled(self, timeout=None, raise_err=None): + """等待当前元素变成可用 + :param timeout:超时时间，为None使用元素所在页面timeout属性 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + return self._wait_state('is_enabled', True, timeout, raise_err) + + def disabled(self, timeout=None, raise_err=None): + """等待当前元素变成可用 + :param timeout:超时时间，为None使用元素所在页面timeout属性 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + return self._wait_state('is_enabled', False, timeout, raise_err) + + def disabled_or_delete(self, timeout=None, raise_err=None): + """等待当前元素变成不可用或从DOM移除 + :param timeout:超时时间，为None使用元素所在页面timeout属性 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + if timeout is None: + timeout = self._page.timeout + end_time = perf_counter() + timeout + while perf_counter() < end_time: + if not self._ele.states.is_enabled or not self._ele.states.is_alive: + return True + sleep(.05) + + if raise_err is True or Settings.raise_when_wait_failed is True: + raise WaitTimeoutError('等待元素隐藏或删除失败。') + else: + return False + + def _wait_state(self, attr, mode=False, timeout=None, raise_err=None): + """等待元素某个bool状态到达指定状态 + :param attr: 状态名称 + :param mode: True或False + :param timeout: 超时时间，为None使用元素所在页面timeout属性 + :param raise_err: 等待识别时是否报错，为None时根据Settings设置 + :return: 是否等待成功 + """ + if timeout is None: + timeout = self._page.timeout + end_time = perf_counter() + timeout + while perf_counter() < end_time: + if self._ele.states.__getattribute__(attr) == mode: + return True + sleep(.05) + + if raise_err is True or Settings.raise_when_wait_failed is True: + raise WaitTimeoutError('等待元素状态改变失败。') + else: + return False + + +class FrameWaiter(ChromiumBaseWaiter, ChromiumElementWaiter): + def __init__(self, frame): + """ + :param frame: ChromiumFrame对象 + """ + super().__init__(frame) + super(ChromiumBaseWaiter, self).__init__(frame, frame.frame_ele) diff --git a/DrissionPage/waiter.pyi b/DrissionPage/waiter.pyi new file mode 100644 index 0000000..41ea5e9 --- /dev/null +++ b/DrissionPage/waiter.pyi @@ -0,0 +1,78 @@ +# -*- coding:utf-8 -*- +""" +@Author : g1879 +@Contact : g1879@qq.com +""" +from typing import Union + +from .chromium_base import ChromiumBase +from .chromium_element import ChromiumElement +from .chromium_frame import ChromiumFrame +from .chromium_page import ChromiumPage + + +class ChromiumBaseWaiter(object): + def __init__(self, page: ChromiumBase): + self._driver: ChromiumBase = ... + + def ele_delete(self, loc_or_ele: Union[str, tuple, ChromiumElement], timeout: float = None, + raise_err: bool = None) -> bool: ... + + def ele_display(self, loc_or_ele: Union[str, tuple, ChromiumElement], timeout: float = None, + raise_err: bool = None) -> bool: ... + + def ele_hidden(self, loc_or_ele: Union[str, tuple, ChromiumElement], timeout: float = None, + raise_err: bool = None) -> bool: ... + + def ele_load(self, loc: Union[str, tuple], timeout: float = None, + raise_err: bool = None) -> Union[bool, ChromiumElement]: ... + + def _loading(self, timeout: float = None, start: bool = True, gap: float = .01, raise_err: bool = None) -> bool: ... + + def load_start(self, timeout: float = None, raise_err: bool = None) -> bool: ... + + def load_complete(self, timeout: float = None, raise_err: bool = None) -> bool: ... + + def upload_paths_inputted(self) -> None: ... + + +class ChromiumPageWaiter(ChromiumBaseWaiter): + _driver: ChromiumPage = ... + + # _listener: Union[NetworkListener, None] = ... + + # def download_begin(self, timeout: float = 1.5) -> bool: ... + + # def download_finish(self, timeout: float = None) -> bool: ... + + def new_tab(self, timeout: float = None, raise_err: bool = None) -> bool: ... + + +class ChromiumElementWaiter(object): + def __init__(self, + page: ChromiumBase, + ele: ChromiumElement): + self._ele: ChromiumElement = ... + self._page: ChromiumBase = ... + + def delete(self, timeout: float = None, raise_err: bool = None) -> bool: ... + + def display(self, timeout: float = None, raise_err: bool = None) -> bool: ... + + def hidden(self, timeout: float = None, raise_err: bool = None) -> bool: ... + + def covered(self, timeout: float = None, raise_err: bool = None) -> bool: ... + + def not_covered(self, timeout: float = None, raise_err: bool = None) -> bool: ... + + def enabled(self, timeout: float = None, raise_err: bool = None) -> bool: ... + + def disabled(self, timeout: float = None, raise_err: bool = None) -> bool: ... + + def disabled_or_delete(self, timeout: float = None, raise_err: bool = None) -> bool: ... + + def _wait_state(self, attr: str, mode: bool = False, timeout: float = None, raise_err: bool = None) -> bool: ... + + +class FrameWaiter(ChromiumBaseWaiter, ChromiumElementWaiter): + def __init__(self, frame: ChromiumFrame): ...