blackteay/DrissionPage
1
0
Fork 0
mirror of https://gitee.com/g1879/DrissionPage.git synced 2024-12-10 04:00:23 +08:00
Code Issues Projects Releases Wiki Activity

增加ele.states.has_rect和ele.wait.stop_moving()；common删除FlowViewer

Browse Source
This commit is contained in:
g1879 2023-10-23 17:44:19 +08:00
parent 06a215d93a
commit 90c715aeae
9 changed files with 69 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
DrissionPage/_elements/chromium_element.py
View File

@ -18,7 +18,7 @@ from .._units.clicker import Clicker
from .._units.setter import ChromiumElementSetter
from .._units.waiter import ChromiumElementWaiter
from ..errors import ContextLossError, ElementLossError, JavaScriptError, ElementNotFoundError, \
CDPError, NoResourceError
CDPError, NoResourceError, NoRectError
class ChromiumElement(DrissionElement):
@ -1465,7 +1465,7 @@ class ChromiumElementStates(object):
@property
def is_in_viewport(self):
"""返回元素是否出现在视口中，以元素可以接受点击的点为判断"""
"""返回元素是否出现在视口中，以元素click_point为判断"""
x, y = self._ele.locations.click_point
return location_in_viewport(self._ele.page, x, y) if x else False
@ -1491,6 +1491,14 @@ class ChromiumElementStates(object):
return False
@property
def has_rect(self):
"""返回元素是否拥有位置和大小，没有返回False，有返回大小元组"""
try:
return self._ele.size
except NoRectError:
return False
class ShadowRootStates(object):
def __init__(self, ele):

3
DrissionPage/_elements/chromium_element.pyi
View File

@ -239,6 +239,9 @@ class ChromiumElementStates(object):
@property
def is_covered(self) -> bool: ...
@property
def has_rect(self) -> Union[bool, Tuple[int, int]]: ...
class ChromiumShadowRoot(BaseElement):

4
DrissionPage/_pages/chromium_base.py
View File

@ -255,6 +255,8 @@ class ChromiumBase(BasePage):
def _onFileChooserOpened(self, **kwargs):
"""文件选择框打开时执行"""
if self._upload_list:
if 'backendNodeId' not in kwargs:
raise TypeError('该输入框无法接管，请改用对<input>元素输入路径的方法设置。')
files = self._upload_list if kwargs['mode'] == 'selectMultiple' else self._upload_list[:1]
self.run_cdp('DOM.setFileInputFiles', files=files, backendNodeId=kwargs['backendNodeId'])
@ -419,7 +421,7 @@ class ChromiumBase(BasePage):
return self._actions
@property
def listener(self):
def listen(self):
"""返回用于聆听数据包的对象"""
if self._listener is None:
self._listener = NetworkListener(self)

2
DrissionPage/_pages/chromium_base.pyi
View File

@ -154,7 +154,7 @@ class ChromiumBase(BasePage):
def actions(self) -> ActionChains: ...
@property
def listener(self) -> NetworkListener: ...
def listen(self) -> NetworkListener: ...
def run_js(self, script: str, *args: Any, as_expr: bool = False) -> Any: ...

15
DrissionPage/_units/network_listener.py
View File

@ -11,6 +11,7 @@ from time import perf_counter, sleep
from requests.structures import CaseInsensitiveDict
from .._base.chromium_driver import ChromiumDriver
from ..errors import CDPError
@ -22,7 +23,7 @@ class NetworkListener(object):
:param page: ChromiumBase对象
"""
self._page = page
self._driver = page.driver
self._driver = ChromiumDriver(page.tab_id, 'page', page.address)
self._driver.call_method('Network.enable')
self._caught = None # 临存捕捉到的数据
@ -65,7 +66,7 @@ class NetworkListener(object):
else:
raise TypeError('method参数只能是str、list、tuple、set类型。')
def listen(self, targets=None, is_regex=False, method=None):
def start(self, targets=None, is_regex=False, method=None):
"""拦截目标请求，每次拦截前清空结果
:param targets: 要匹配的数据包url特征，可用list等传入多个，为True时获取所有
:param is_regex: 设置的target是否正则表达式
@ -180,10 +181,10 @@ class NetworkListener(object):
def _requestWillBeSent(self, **kwargs):
"""接收到请求时的回调函数"""
if not self._targets:
self._request_ids[kwargs['requestId']] = DataPacket(self._page.tab_id, None, kwargs)
self._request_ids[kwargs['requestId']] = DataPacket(self._driver.id, None, kwargs)
if kwargs['request'].get('hasPostData', None) and not kwargs['request'].get('postData', None):
self._request_ids[kwargs['requestId']]._raw_post_data = \
self._page.run_cdp('Network.getRequestPostData', requestId=kwargs['requestId'])['postData']
self._driver.call_method('Network.getRequestPostData', requestId=kwargs['requestId'])['postData']
return
@ -191,11 +192,11 @@ class NetworkListener(object):
if ((self._is_regex and search(target, kwargs['request']['url'])) or
(not self._is_regex and target in kwargs['request']['url'])) and (
not self._method or kwargs['request']['method'] in self._method):
self._request_ids[kwargs['requestId']] = DataPacket(self._page.tab_id, target, kwargs)
self._request_ids[kwargs['requestId']] = DataPacket(self._driver.id, target, kwargs)
if kwargs['request'].get('hasPostData', None) and not kwargs['request'].get('postData', None):
self._request_ids[kwargs['requestId']]._raw_post_data = \
self._page.run_cdp('Network.getRequestPostData', requestId=kwargs['requestId'])['postData']
self._driver.call_method('Network.getRequestPostData', requestId=kwargs['requestId'])['postData']
break
@ -212,7 +213,7 @@ class NetworkListener(object):
dp = self._request_ids.get(request_id)
if dp:
try:
r = self._page.run_cdp('Network.getResponseBody', requestId=request_id)
r = self._driver.call_method('Network.getResponseBody', requestId=request_id)
body = r['body']
is_base64 = r['base64Encoded']
except CDPError:

4
DrissionPage/_units/network_listener.pyi
View File

@ -42,8 +42,8 @@ class NetworkListener(object):
def clear(self) -> None: ...
def listen(self, targets: Union[str, List[str], Tuple, bool, None] = None, is_regex: bool = False,
method: Union[str, list, tuple, set] = None) \
def start(self, targets: Union[str, List[str], Tuple, bool, None] = None, is_regex: bool = False,
method: Union[str, list, tuple, set] = None) \
-> Union[DataPacket, Dict[str, List[DataPacket]], False]: ...
def _requestWillBeSent(self, **kwargs) -> None: ...

46
DrissionPage/_units/waiter.py
View File

@ -2,7 +2,7 @@
from time import sleep, perf_counter
from .._commons.constants import Settings
from ..errors import WaitTimeoutError
from ..errors import WaitTimeoutError, NoRectError
class ChromiumBaseWaiter(object):
@ -132,7 +132,7 @@ class ChromiumBaseWaiter(object):
:param timeout: 超时时间，为None无限等待
:param fix_count: 是否必须满足总数要求，发生超时，为True返回False，为False返回已捕捉到的数据包
:return: count为1时返回数据包对象，大于1时返回列表，超时且fix_count为True时返回False"""
return self._driver.listener.wait(count, timeout, fix_count)
return self._driver.listen.wait(count, timeout, fix_count)
def _change(self, arg, text, exclude=False, timeout=None, raise_err=None):
"""等待指定属性变成包含或不包含指定文本
@ -309,7 +309,7 @@ class ChromiumElementWaiter(object):
def covered(self, timeout=None, raise_err=None):
"""等待当前元素被遮盖
:param timeout:超时时间，为None使用元素所在页面timeout属性
:param timeout: 超时时间，为None使用元素所在页面timeout属性
:param raise_err: 等待失败时是否报错，为None时根据Settings设置
:return: 是否等待成功
"""
@ -317,7 +317,7 @@ class ChromiumElementWaiter(object):
def not_covered(self, timeout=None, raise_err=None):
"""等待当前元素被遮盖
:param timeout:超时时间，为None使用元素所在页面timeout属性
:param timeout: 超时时间，为None使用元素所在页面timeout属性
:param raise_err: 等待失败时是否报错，为None时根据Settings设置
:return: 是否等待成功
"""
@ -325,7 +325,7 @@ class ChromiumElementWaiter(object):
def enabled(self, timeout=None, raise_err=None):
"""等待当前元素变成可用
:param timeout:超时时间，为None使用元素所在页面timeout属性
:param timeout: 超时时间，为None使用元素所在页面timeout属性
:param raise_err: 等待失败时是否报错，为None时根据Settings设置
:return: 是否等待成功
"""
@ -333,7 +333,7 @@ class ChromiumElementWaiter(object):
def disabled(self, timeout=None, raise_err=None):
"""等待当前元素变成可用
:param timeout:超时时间，为None使用元素所在页面timeout属性
:param timeout: 超时时间，为None使用元素所在页面timeout属性
:param raise_err: 等待失败时是否报错，为None时根据Settings设置
:return: 是否等待成功
"""
@ -341,7 +341,7 @@ class ChromiumElementWaiter(object):
def disabled_or_delete(self, timeout=None, raise_err=None):
"""等待当前元素变成不可用或从DOM移除
:param timeout:超时时间，为None使用元素所在页面timeout属性
:param timeout: 超时时间，为None使用元素所在页面timeout属性
:param raise_err: 等待失败时是否报错，为None时根据Settings设置
:return: 是否等待成功
"""
@ -358,6 +358,38 @@ class ChromiumElementWaiter(object):
else:
return False
def stop_moving(self, gap=.1, timeout=None, raise_err=None):
"""等待当前元素停止运动
:param gap: 检测间隔时间
: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:
try:
size = self._ele.states.has_rect
location = self._ele.location
break
except NoRectError:
pass
else:
raise NoRectError
while perf_counter() < end_time:
sleep(gap)
if self._ele.size == size and location == self._ele.location:
return True
size = self._ele.size
location = self._ele.location
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: 状态名称

2
DrissionPage/_units/waiter.pyi
View File

@ -92,6 +92,8 @@ class ChromiumElementWaiter(object):
def disabled_or_delete(self, timeout: float = None, raise_err: bool = None) -> bool: ...
def stop_moving(self, gap: float = .1, 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: ...

5
DrissionPage/common.py
View File

@ -2,13 +2,10 @@
"""
@Author : g1879
@Contact : g1879@qq.com
实用工具
"""
from FlowViewer import Listener, RequestMan
from ._elements.session_element import make_session_ele
from ._units.action_chains import ActionChains
from ._commons.keys import Keys
from ._commons.by import By
from ._commons.constants import Settings
from ._commons.tools import wait_until