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

删除wait.new_frame；增强iframe和查找元素稳定性

Browse Source
This commit is contained in:
g1879 2023-03-28 18:15:38 +08:00
parent f81a4e439c
commit 7eb1aac778
5 changed files with 30 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
DrissionPage/base.py
View File

@ -5,12 +5,13 @@
"""
from abc import abstractmethod
from re import sub
from time import sleep
from urllib.parse import quote
from .commons.constants import Settings, NoneElement
from .commons.locator import get_loc
from .commons.web import format_html
from .errors import ElementNotFoundError
from .errors import ElementNotFoundError, ContextLossError
class BaseParser(object):
@ -71,7 +72,13 @@ class BaseElement(BaseParser):
pass
def _ele(self, loc_or_str, timeout=None, single=True, relative=False, raise_err=None):
r = self._find_elements(loc_or_str, timeout=timeout, single=single, relative=relative, raise_err=raise_err)
while True:
try:
r = self._find_elements(loc_or_str, timeout=timeout, single=single,
relative=relative, raise_err=raise_err)
break
except ContextLossError:
sleep(.1)
if not single or raise_err is False:
return r
if not r and (Settings.raise_ele_not_found or raise_err is True):
@ -412,7 +419,14 @@ class BasePage(BaseParser):
def _ele(self, loc_or_ele, timeout=None, single=True, raise_err=None):
if not loc_or_ele:
raise ElementNotFoundError
r = self._find_elements(loc_or_ele, timeout=timeout, single=single, raise_err=raise_err)
while True:
try:
r = self._find_elements(loc_or_ele, timeout=timeout, single=single, raise_err=raise_err)
break
except ContextLossError:
sleep(.1)
if not single or raise_err is False:
return r
if not r and (Settings().raise_ele_not_found is True or raise_err is True):

23
DrissionPage/chromium_base.py
View File

@ -1061,29 +1061,6 @@ class ChromiumBaseWaiter(object):
"""
return self._loading(timeout=timeout, start=False)
def new_frame(self, timeout=None):
"""等待新frame加载到dom
:param timeout: 超时时间
:return: 是否等待成功
"""
timeout = timeout if timeout is not None else self._driver.timeout
self._new_frame = False
self._driver.driver.Page.frameAttached = self._on_frame_attached
result = False
end_time = perf_counter() + timeout
while perf_counter() < end_time:
if self._new_frame:
result = True
break
sleep(.1)
self._driver.driver.Page.frameAttached = None
self._new_frame = False
return result
def _on_frame_attached(self):
self._new_frame = True
def upload_paths_inputted(self):
"""等待自动填写上传文件路径"""
while self._driver._upload_list:

5
DrissionPage/chromium_base.pyi
View File

@ -217,7 +217,6 @@ class ChromiumBase(BasePage):
class ChromiumBaseWaiter(object):
def __init__(self, page: ChromiumBase):
self._driver: ChromiumBase = ...
self._new_frame: bool = ...
def ele_delete(self, loc_or_ele: Union[str, tuple, ChromiumElement], timeout: float = None) -> bool: ...
@ -231,10 +230,6 @@ class ChromiumBaseWaiter(object):
def load_complete(self, timeout: float = None) -> bool: ...
def new_frame(self, timeout: float = None) -> bool: ...
def _on_frame_attached(self): ...
def data_package(self, target: str, timeout: float = None) -> Union[ResponseData, None]: ...
def upload_paths_inputted(self) -> None: ...

14
DrissionPage/chromium_element.py
View File

@ -1252,7 +1252,7 @@ def find_by_xpath(ele, xpath, single, timeout, relative=True):
type_txt = '9' if single else '7'
node_txt = 'this.contentDocument' if ele.tag in FRAME_ELEMENT and not relative else 'this'
js = make_js_for_find_ele_by_xpath(xpath, type_txt, node_txt)
r = ele.page.run_cdp('Runtime.callFunctionOn',
r = ele.page.run_cdp_loaded('Runtime.callFunctionOn',
functionDeclaration=js, objectId=ele.ids.obj_id, returnByValue=False, awaitPromise=True,
userGesture=True)
if r['result']['type'] == 'string':
@ -1261,7 +1261,7 @@ def find_by_xpath(ele, xpath, single, timeout, relative=True):
if 'exceptionDetails' in r:
if 'The result is not a node set' in r['result']['description']:
js = make_js_for_find_ele_by_xpath(xpath, '1', node_txt)
r = ele.page.run_cdp('Runtime.callFunctionOn',
r = ele.page.run_cdp_loaded('Runtime.callFunctionOn',
functionDeclaration=js, objectId=ele.ids.obj_id, returnByValue=False,
awaitPromise=True,
userGesture=True)
@ -1272,7 +1272,7 @@ def find_by_xpath(ele, xpath, single, timeout, relative=True):
end_time = perf_counter() + timeout
while (r['result']['subtype'] == 'null'
or r['result']['description'] == 'NodeList(0)') and perf_counter() < end_time:
r = ele.page.run_cdp('Runtime.callFunctionOn',
r = ele.page.run_cdp_loaded('Runtime.callFunctionOn',
functionDeclaration=js, objectId=ele.ids.obj_id, returnByValue=False, awaitPromise=True,
userGesture=True)
@ -1283,7 +1283,7 @@ def find_by_xpath(ele, xpath, single, timeout, relative=True):
if r['result']['description'] == 'NodeList(0)':
return []
else:
r = ele.page.run_cdp('Runtime.getProperties', objectId=r['result']['objectId'], ownProperties=True)['result']
r = ele.page.run_cdp_loaded('Runtime.getProperties', objectId=r['result']['objectId'], ownProperties=True)['result']
return [make_chromium_ele(ele.page, obj_id=i['value']['objectId'])
if i['value']['type'] == 'object' else i['value']['value']
for i in r[:-1]]
@ -1301,14 +1301,14 @@ def find_by_css(ele, selector, single, timeout):
find_all = '' if single else 'All'
node_txt = 'this.contentDocument' if ele.tag in ('iframe', 'frame', 'shadow-root') else 'this'
js = f'function(){{return {node_txt}.querySelector{find_all}("{selector}");}}'
r = ele.page.run_cdp('Runtime.callFunctionOn',
r = ele.page.run_cdp_loaded('Runtime.callFunctionOn',
functionDeclaration=js, objectId=ele.ids.obj_id, returnByValue=False, awaitPromise=True,
userGesture=True)
end_time = perf_counter() + timeout
while ('exceptionDetails' in r or r['result']['subtype'] == 'null'
or r['result']['description'] == 'NodeList(0)') and perf_counter() < end_time:
r = ele.page.run_cdp('Runtime.callFunctionOn',
r = ele.page.run_cdp_loaded('Runtime.callFunctionOn',
functionDeclaration=js, objectId=ele.ids.obj_id, returnByValue=False, awaitPromise=True,
userGesture=True)
@ -1322,7 +1322,7 @@ def find_by_css(ele, selector, single, timeout):
if r['result']['description'] == 'NodeList(0)':
return []
else:
r = ele.page.run_cdp('Runtime.getProperties', objectId=r['result']['objectId'], ownProperties=True)['result']
r = ele.page.run_cdp_loaded('Runtime.getProperties', objectId=r['result']['objectId'], ownProperties=True)['result']
return [make_chromium_ele(ele.page, obj_id=i['value']['objectId']) for i in r]

9
DrissionPage/chromium_frame.py
View File

@ -7,6 +7,7 @@ from re import search
from time import sleep, perf_counter
from warnings import warn
from errors import ContextLossError
from .commons.tools import get_usable_path
from .chromium_base import ChromiumBase, ChromiumPageScroll, ChromiumBaseSetter, ChromiumBaseWaiter
from .chromium_element import ChromiumElement, ChromiumElementWaiter
@ -131,7 +132,7 @@ class ChromiumFrame(ChromiumBase):
def _onFrameNavigated(self, **kwargs):
"""页面跳转时触发"""
if kwargs['frame']['frameId'] == self.frame_id and self._first_run is False and self._is_loading:
if kwargs['frame']['id'] == self.frame_id and self._first_run is False and self._is_loading:
self._is_loading = True
if self._debug:
@ -274,8 +275,10 @@ class ChromiumFrame(ChromiumBase):
while True:
try:
return self.doc_ele.run_js('return this.readyState;')
except:
sleep(.1)
except ContextLossError:
node = self.run_cdp('DOM.describeNode', backendNodeId=self.frame_ele.ids.backend_id)['node']
doc = ChromiumElement(self.page, backend_id=node['contentDocument']['backendNodeId'])
return doc.run_js('return this.readyState;')
@property
def scroll(self):