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

4.0.0b17(+)

Browse Source 
修复相对定位timeout失效问题；
相对定位timeout改为None；
增加listen.wait_silent()
This commit is contained in:
g1879 2023-12-03 23:18:27 +08:00
parent 364700df2c
commit bf245f7221
9 changed files with 69 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
DrissionPage/__init__.py
View File

@ -13,4 +13,4 @@ from ._configs.chromium_options import ChromiumOptions
from ._configs.session_options import SessionOptions
__all__ = ['ChromiumPage', 'ChromiumOptions', 'SessionOptions', 'SessionPage', 'WebPage', '__version__']
__version__ = '4.0.0b16'
__version__ = '4.0.0b17'

8
DrissionPage/_base/base.py
View File

@ -172,7 +172,7 @@ class DrissionElement(BaseElement):
return NoneElement(self.page, 'child()', {'filter_loc': filter_loc,
'index': index, 'ele_only': ele_only})
def prev(self, filter_loc='', index=1, timeout=0, ele_only=True):
def prev(self, filter_loc='', index=1, timeout=None, ele_only=True):
"""返回前面的一个兄弟元素，可用查询语法筛选，可指定返回筛选结果的第几个
:param filter_loc: 用于筛选的查询语法
:param index: 前面第几个查询结果，1开始
@ -192,7 +192,7 @@ class DrissionElement(BaseElement):
else:
return NoneElement(self.page, 'prev()', {'filter_loc': filter_loc, 'index': index, 'ele_only': ele_only})
def next(self, filter_loc='', index=1, timeout=0, ele_only=True):
def next(self, filter_loc='', index=1, timeout=None, ele_only=True):
"""返回后面的一个兄弟元素，可用查询语法筛选，可指定返回筛选结果的第几个
:param filter_loc: 用于筛选的查询语法
:param index: 后面第几个查询结果，1开始
@ -271,7 +271,7 @@ class DrissionElement(BaseElement):
nodes = self._ele(loc, timeout=timeout, single=False, relative=True)
return [e for e in nodes if not (isinstance(e, str) and sub('[ \n\t\r]', '', e) == '')]
def prevs(self, filter_loc='', timeout=0, ele_only=True):
def prevs(self, filter_loc='', timeout=None, ele_only=True):
"""返回前面全部兄弟元素或节点组成的列表，可用查询语法筛选
:param filter_loc: 用于筛选的查询语法
:param timeout: 查找节点的超时时间
@ -280,7 +280,7 @@ class DrissionElement(BaseElement):
"""
return self._get_brothers(filter_loc=filter_loc, direction='preceding', timeout=timeout, ele_only=ele_only)
def nexts(self, filter_loc='', timeout=0, ele_only=True):
def nexts(self, filter_loc='', timeout=None, ele_only=True):
"""返回后面全部兄弟元素或节点组成的列表，可用查询语法筛选
:param filter_loc: 用于筛选的查询语法
:param timeout: 查找节点的超时时间

65
DrissionPage/_base/base.pyi
View File

@ -83,62 +83,39 @@ class DrissionElement(BaseElement):
def parent(self, level_or_loc: Union[tuple, str, int] = 1, index: int = 1) -> Union[DrissionElement, None]: ...
def child(self,
filter_loc: Union[tuple, str, int] = '',
index: int = 1,
timeout: float = None,
ele_only: bool = True) -> Union[DrissionElement, str, NoneElement]: ...
def child(self, filter_loc: Union[tuple, str, int] = '', index: int = 1,
timeout: float = None, ele_only: bool = True) -> Union[DrissionElement, str, NoneElement]: ...
def prev(self,
filter_loc: Union[tuple, str, int] = '',
index: int = 1,
timeout: float = 0,
ele_only: bool = True) -> Union[DrissionElement, str, NoneElement]: ...
def prev(self, filter_loc: Union[tuple, str, int] = '', index: int = 1,
timeout: float = None, ele_only: bool = True) -> Union[DrissionElement, str, NoneElement]: ...
def next(self,
filter_loc: Union[tuple, str, int] = '',
index: int = 1,
timeout: float = 0,
ele_only: bool = True) -> Union[DrissionElement, str, NoneElement]: ...
def next(self, filter_loc: Union[tuple, str, int] = '', index: int = 1,
timeout: float = None, ele_only: bool = True) -> Union[DrissionElement, str, NoneElement]: ...
def before(self,
filter_loc: Union[tuple, str, int] = '',
index: int = 1,
timeout: float = None,
ele_only: bool = True) -> Union[DrissionElement, str, NoneElement]: ...
def before(self, filter_loc: Union[tuple, str, int] = '', index: int = 1,
timeout: float = None, ele_only: bool = True) -> Union[DrissionElement, str, NoneElement]: ...
def after(self,
filter_loc: Union[tuple, str, int] = '',
index: int = 1,
timeout: float = None,
ele_only: bool = True) -> Union[DrissionElement, str, NoneElement]: ...
def after(self, filter_loc: Union[tuple, str, int] = '', index: int = 1,
timeout: float = None, ele_only: bool = True) -> Union[DrissionElement, str, NoneElement]: ...
def children(self, filter_loc: Union[tuple, str] = '',
timeout: float = None,
def children(self, filter_loc: Union[tuple, str] = '', timeout: float = None,
ele_only: bool = True) -> List[Union[DrissionElement, str]]: ...
def prevs(self, filter_loc: Union[tuple, str] = '',
timeout: float = 0,
def prevs(self, filter_loc: Union[tuple, str] = '', timeout: float = None,
ele_only: bool = True) -> List[Union[DrissionElement, str]]: ...
def nexts(self, filter_loc: Union[tuple, str] = '',
timeout: float = 0,
def nexts(self, filter_loc: Union[tuple, str] = '', timeout: float = None,
ele_only: bool = True) -> List[Union[DrissionElement, str]]: ...
def befores(self, filter_loc: Union[tuple, str] = '',
timeout: float = None,
def befores(self, filter_loc: Union[tuple, str] = '', timeout: float = None,
ele_only: bool = True) -> List[Union[DrissionElement, str]]: ...
def afters(self, filter_loc: Union[tuple, str] = '',
timeout: float = None,
def afters(self, filter_loc: Union[tuple, str] = '', timeout: float = None,
ele_only: bool = True) -> List[Union[DrissionElement, str]]: ...
def _get_brothers(self, index: int = None,
filter_loc: Union[tuple, str] = '',
direction: str = 'following',
brother: bool = True,
timeout: float = 0.5,
ele_only: bool = True) -> List[Union[DrissionElement, str]]: ...
def _get_brothers(self, index: int = None, filter_loc: Union[tuple, str] = '',
direction: str = 'following', brother: bool = True,
timeout: float = 0.5, ele_only: bool = True) -> List[Union[DrissionElement, str]]: ...
# ----------------以下属性或方法由后代实现----------------
@property
@ -205,11 +182,7 @@ class BasePage(BaseParser):
def get_cookies(self, as_dict: bool = False, all_info: bool = False) -> Union[list, dict]: ...
@abstractmethod
def get(self,
url: str,
show_errmsg: bool = False,
retry: int = None,
interval: float = None): ...
def get(self, url: str, show_errmsg: bool = False, retry: int = None, interval: float = None): ...
def _ele(self, loc_or_ele, timeout: float = None, single: bool = True,
raise_err: bool = None, method: str = None): ...

4
DrissionPage/_base/chromium_driver.py
View File

@ -190,14 +190,10 @@ class ChromiumDriver(object):
event = self.event_queue.get_nowait()
function = self.event_handlers.get(event['method'])
if function:
# if self._debug:
# print(f'开始执行 {function.__name__}')
try:
function(**event['params'])
except:
pass
# if self._debug:
# print(f'执行 {function.__name__}完毕')
self.event_handlers.clear()
self.method_results.clear()

6
DrissionPage/_elements/chromium_element.py
View File

@ -724,7 +724,7 @@ class ChromiumElement(DrissionElement):
elif mode == 'css':
txt1 = ''
txt3 = ''
txt4 = '''path = '>' + el.tagName + ":nth-child(" + nth + ")" + path;'''
txt4 = '''path = '>' + el.tagName.toLowerCase() + ":nth-child(" + nth + ")" + path;'''
txt5 = '''return path.substr(1);'''
else:
@ -1150,8 +1150,8 @@ def find_by_xpath(ele, xpath, single, timeout, relative=True):
raise SyntaxError(f'查询语句错误：\n{r}')
end_time = perf_counter() + timeout
while (r['result']['subtype'] == 'null'
or r['result']['description'] == 'NodeList(0)') and perf_counter() < end_time:
while ((r['result']['subtype'] == 'null' or r['result']['description'] in ('NodeList(0)', 'Array(0)'))
and perf_counter() < end_time):
r = ele.page.run_cdp_loaded('Runtime.callFunctionOn', functionDeclaration=js, objectId=ele._obj_id,
returnByValue=False, awaitPromise=True, userGesture=True)

14
DrissionPage/_pages/chromium_frame.py
View File

@ -15,7 +15,7 @@ from .._units.scroller import FrameScroller
from .._units.setter import ChromiumFrameSetter
from .._units.states import FrameStates
from .._units.waiter import FrameWaiter
from ..errors import ContextLostError, ElementLostError, GetDocumentError, PageClosedError
from ..errors import ContextLostError, ElementLostError, GetDocumentError, PageClosedError, JavaScriptError
class ChromiumFrame(ChromiumBase):
@ -54,10 +54,11 @@ class ChromiumFrame(ChromiumBase):
obj_id = super().run_js('document;', as_expr=True)['objectId']
self.doc_ele = ChromiumElement(self, obj_id=obj_id)
end_time = perf_counter() + 5
while perf_counter() < end_time and self.url == 'about:blank':
sleep(.1)
self._rect = None
end_time = perf_counter() + 5
while perf_counter() < end_time:
if self.url is None or self.url == 'about:blank':
sleep(.1)
def __call__(self, loc_or_str, timeout=None):
"""在内部查找元素
@ -292,7 +293,10 @@ class ChromiumFrame(ChromiumBase):
@property
def url(self):
"""返回frame当前访问的url"""
return self.doc_ele.run_js('return this.location.href;')
try:
return self.doc_ele.run_js('return this.location.href;')
except JavaScriptError:
return None
@property
def html(self):

28
DrissionPage/_units/listener.py
View File

@ -25,6 +25,7 @@ class Listener(object):
self._address = page.address
self._target_id = page._target_id
self._driver = None
self._running_requests = 0
self._caught = None # 临存捕捉到的数据
self._request_ids = None # 暂存须要拦截的请求id
@ -86,6 +87,7 @@ class Listener(object):
self._request_ids = {}
self._extra_info_ids = {}
self._caught = Queue(maxsize=0)
self._running_requests = 0
self._set_callback()
@ -180,6 +182,27 @@ class Listener(object):
self._request_ids = {}
self._extra_info_ids = {}
self._caught.queue.clear()
self._running_requests = 0
def wait_silent(self, timeout=None):
"""等待所有请求结束
:param timeout: 超时，为None时无限等待
:return: 返回是否等待成功
"""
if not self.listening:
raise RuntimeError('监听未启动，用listen.start()启动。')
if timeout is None:
while self._running_requests > 0:
sleep(.1)
return True
end_time = perf_counter() + timeout
while perf_counter() < end_time:
if self._running_requests <= 0:
return True
sleep(.1)
else:
return False
def _to_target(self, target_id, address, page):
"""切换监听的页面对象
@ -212,6 +235,7 @@ class Listener(object):
def _requestWillBeSent(self, **kwargs):
"""接收到请求时的回调函数"""
self._running_requests += 1
p = None
if not self._targets:
if not self._method or kwargs['request']['method'] in self._method:
@ -236,6 +260,7 @@ class Listener(object):
def _requestWillBeSentExtraInfo(self, **kwargs):
"""接收到请求额外信息时的回调函数"""
self._running_requests += 1
self._extra_info_ids.setdefault(kwargs['requestId'], {})['request'] = kwargs
def _response_received(self, **kwargs):
@ -247,6 +272,7 @@ class Listener(object):
def _responseReceivedExtraInfo(self, **kwargs):
"""接收到返回额外信息时的回调函数"""
self._running_requests -= 1
r = self._extra_info_ids.get(kwargs['requestId'], None)
if r:
obj = r.get('obj', None)
@ -261,6 +287,7 @@ class Listener(object):
def _loading_finished(self, **kwargs):
"""请求完成时处理方法"""
self._running_requests -= 1
rid = kwargs['requestId']
packet = self._request_ids.get(rid)
if packet:
@ -295,6 +322,7 @@ class Listener(object):
def _loading_failed(self, **kwargs):
"""请求失败时的回调方法"""
self._running_requests -= 1
r_id = kwargs['requestId']
dp = self._request_ids.get(r_id, None)
if dp:

5
DrissionPage/_units/listener.pyi
View File

@ -26,6 +26,7 @@ class Listener(object):
self._request_ids: dict = ...
self._extra_info_ids: dict = ...
self.listening: bool = ...
self._running_requests: int = ...
@property
def targets(self) -> Optional[set]: ...
@ -47,6 +48,8 @@ class Listener(object):
def clear(self) -> None: ...
def wait_silent(self, timeout=None) -> bool: ...
def _to_target(self, target_id: str, address: str, page: ChromiumBase) -> None: ...
def start(self, targets: Union[str, List[str], Tuple, bool, None] = None, is_regex: bool = False,
@ -72,7 +75,7 @@ class Listener(object):
class FrameListener(Listener):
def __init__(self, page: ChromiumFrame, is_diff: bool):
def __init__(self, page: ChromiumFrame):
self._page: ChromiumFrame = ...
self._is_diff: bool = ...

2
setup.py
View File

@ -6,7 +6,7 @@ with open("README.md", "r", encoding='utf-8') as fh:
setup(
name="DrissionPage",
version="4.0.0b16",
version="4.0.0b17",
author="g1879",
author_email="g1879@qq.com",
description="Python based web automation tool. It can control the browser and send and receive data packets.",