mirror of
https://gitee.com/g1879/DrissionPage.git
synced 2024-12-10 04:00:23 +08:00
Waiter重构到专属文件
This commit is contained in:
g1879
parent
3f999f066c
commit
b62c3cb6a1
@ -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):
|
||||
"""
|
||||
|
||||
@ -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): ...
|
||||
|
||||
|
||||
@ -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):
|
||||
"""
|
||||
|
||||
@ -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 = ...
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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): ...
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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):
|
||||
|
||||
242
DrissionPage/waiter.py
Normal file
242
DrissionPage/waiter.py
Normal file
@ -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)
|
||||
78
DrissionPage/waiter.pyi
Normal file
78
DrissionPage/waiter.pyi
Normal file
@ -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): ...
|
||||
Loading…
x
Reference in New Issue
Block a user