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.1.0.0b2s_ele()和s_eles()增加timeout参数；优化连接浏览器失败提示

Browse Source
This commit is contained in:
g1879 2024-07-09 17:36:23 +08:00
parent d758932910
commit 2ea2be782f
9 changed files with 107 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
DrissionPage/__init__.py
View File

@ -14,4 +14,4 @@ from ._pages.chromium_page import ChromiumPage
from ._pages.mix_page import MixPage from ._pages.mix_page import MixPage
from ._pages.mix_page import MixPage as WebPage from ._pages.mix_page import MixPage as WebPage
__version__ = '4.1.0.0b1' __version__ = '4.1.0.0b2'

28
DrissionPage/_elements/chromium_element.py
View File

@ -162,7 +162,7 @@ class ChromiumElement(DrissionElement):
return self._rect return self._rect
@property @property
def shadow_root(self): def sr(self):
"""返回当前元素的shadow_root元素对象""" """返回当前元素的shadow_root元素对象"""
info = self.owner._run_cdp('DOM.describeNode', backendNodeId=self._backend_id)['node'] info = self.owner._run_cdp('DOM.describeNode', backendNodeId=self._backend_id)['node']
if not info.get('shadowRoots', None): if not info.get('shadowRoots', None):
@ -171,9 +171,9 @@ class ChromiumElement(DrissionElement):
return ShadowRoot(self, backend_id=info['shadowRoots'][0]['backendNodeId']) return ShadowRoot(self, backend_id=info['shadowRoots'][0]['backendNodeId'])
@property @property
def sr(self): def shadow_root(self):
"""返回当前元素的shadow_root元素对象""" """返回当前元素的shadow_root元素对象"""
return self.shadow_root return self.sr
@property @property
def scroll(self): def scroll(self):
@ -564,22 +564,25 @@ class ChromiumElement(DrissionElement):
""" """
return self._ele(locator, timeout=timeout, index=None) return self._ele(locator, timeout=timeout, index=None)
def s_ele(self, locator=None, index=1): def s_ele(self, locator=None, index=1, timeout=None):
"""查找一个符合条件的元素，以SessionElement形式返回 """查找一个符合条件的元素，以SessionElement形式返回
:param locator: 元素的定位信息，可以是loc元组，或查询字符串 :param locator: 元素的定位信息，可以是loc元组，或查询字符串
:param index: 获取第几个，从1开始，可传入负数获取倒数第几个 :param index: 获取第几个，从1开始，可传入负数获取倒数第几个
:param timeout: 查找元素超时时间（秒），默认与元素所在页面等待时间一致
:return: SessionElement对象或属性、文本 :return: SessionElement对象或属性、文本
""" """
return (make_session_ele(self, locator, index=index, method='s_ele()') return (make_session_ele(self, locator, index=index, method='s_ele()')
if self.ele(locator, index=index) if self.ele(locator, index=index, timeout=timeout)
else NoneElement(self, method='s_ele()', args={'locator': locator, 'index': index})) else NoneElement(self, method='s_ele()', args={'locator': locator, 'index': index}))
def s_eles(self, locator=None): def s_eles(self, locator=None, timeout=None):
"""查找所有符合条件的元素，以SessionElement列表形式返回 """查找所有符合条件的元素，以SessionElement列表形式返回
:param locator: 定位符 :param locator: 定位符
:param timeout: 查找元素超时时间（秒），默认与元素所在页面等待时间一致
:return: SessionElement或属性、文本组成的列表 :return: SessionElement或属性、文本组成的列表
""" """
return make_session_ele(self, locator, index=None) if self.ele(locator) else SessionElementsList() return (make_session_ele(self, locator, index=None)
if self.ele(locator, timeout=timeout) else SessionElementsList())
def _find_elements(self, locator, timeout=None, index=1, relative=False, raise_err=None): def _find_elements(self, locator, timeout=None, index=1, relative=False, raise_err=None):
"""返回当前元素下级符合条件的子元素、属性或节点文本，默认返回第一个 """返回当前元素下级符合条件的子元素、属性或节点文本，默认返回第一个
@ -1152,22 +1155,25 @@ class ShadowRoot(BaseElement):
""" """
return self._ele(locator, timeout=timeout, index=None) return self._ele(locator, timeout=timeout, index=None)
def s_ele(self, locator=None, index=1): def s_ele(self, locator=None, index=1, timeout=None):
"""查找一个符合条件的元素以SessionElement形式返回，处理复杂页面时效率很高 """查找一个符合条件的元素以SessionElement形式返回，处理复杂页面时效率很高
:param locator: 元素的定位信息，可以是loc元组，或查询字符串 :param locator: 元素的定位信息，可以是loc元组，或查询字符串
:param index: 获取第几个，从1开始，可传入负数获取倒数第几个 :param index: 获取第几个，从1开始，可传入负数获取倒数第几个
:param timeout: 查找元素超时时间（秒），默认与元素所在页面等待时间一致
:return: SessionElement对象或属性、文本 :return: SessionElement对象或属性、文本
""" """
return (make_session_ele(self, locator, index=index, method='s_ele()') return (make_session_ele(self, locator, index=index, method='s_ele()')
if self.ele(locator, index=index) if self.ele(locator, index=index, timeout=timeout)
else NoneElement(self, method='s_ele()', args={'locator': locator, 'index': index})) else NoneElement(self, method='s_ele()', args={'locator': locator, 'index': index}))
def s_eles(self, locator): def s_eles(self, locator, timeout=None):
"""查找所有符合条件的元素以SessionElement列表形式返回，处理复杂页面时效率很高 """查找所有符合条件的元素以SessionElement列表形式返回，处理复杂页面时效率很高
:param locator: 元素的定位信息，可以是loc元组，或查询字符串 :param locator: 元素的定位信息，可以是loc元组，或查询字符串
:param timeout: 查找元素超时时间（秒），默认与元素所在页面等待时间一致
:return: SessionElement对象 :return: SessionElement对象
""" """
return make_session_ele(self, locator, index=None) if self.ele(locator) else SessionElementsList() return (make_session_ele(self, locator, index=None)
if self.ele(locator, timeout=timeout) else SessionElementsList())
def _find_elements(self, locator, timeout=None, index=1, relative=False, raise_err=None): def _find_elements(self, locator, timeout=None, index=1, relative=False, raise_err=None):
"""返回当前元素下级符合条件的子元素、属性或节点文本，默认返回第一个 """返回当前元素下级符合条件的子元素、属性或节点文本，默认返回第一个

12
DrissionPage/_elements/chromium_element.pyi
View File

@ -209,9 +209,12 @@ class ChromiumElement(DrissionElement):
def s_ele(self, def s_ele(self,
locator: Union[Tuple[str, str], str] = None, locator: Union[Tuple[str, str], str] = None,
index: int = 1) -> SessionElement: ... index: int = 1,
timeout: float = None) -> SessionElement: ...
def s_eles(self, locator: Union[Tuple[str, str], str] = None) -> SessionElementsList: ... def s_eles(self,
locator: Union[Tuple[str, str], str] = None,
timeout: float = None) -> SessionElementsList: ...
def _find_elements(self, def _find_elements(self,
locator: Union[Tuple[str, str], str], locator: Union[Tuple[str, str], str],
@ -339,9 +342,10 @@ class ShadowRoot(BaseElement):
def s_ele(self, def s_ele(self,
locator: Union[Tuple[str, str], str] = None, locator: Union[Tuple[str, str], str] = None,
index: int = 1) -> SessionElement: ... index: int = 1,
timeout: float = None) -> SessionElement: ...
def s_eles(self, locator: Union[Tuple[str, str], str]) -> SessionElementsList: ... def s_eles(self, locator: Union[Tuple[str, str], str], timeout: float = None) -> SessionElementsList: ...
def _find_elements(self, def _find_elements(self,
locator: Union[Tuple[str, str], str], locator: Union[Tuple[str, str], str],

28
DrissionPage/_functions/browser.py
View File

@ -29,9 +29,17 @@ def connect_browser(option):
browser_path = option.browser_path browser_path = option.browser_path
ip, port = address.split(':') ip, port = address.split(':')
if ip != '127.0.0.1' or port_is_using(ip, port) or option.is_existing_only: using = port_is_using(ip, port)
test_connect(ip, port) if ip != '127.0.0.1' or using or option.is_existing_only:
if test_connect(ip, port):
return True return True
elif ip != '127.0.0.1':
raise BrowserConnectError(f'\n{address}浏览器连接失败。')
elif using:
raise BrowserConnectError(f'\n{address}浏览器连接失败，请检查{port}端口是否浏览器，'
f'且已添加\'--remote-debugging-port={port}\'启动项。')
else: # option.is_existing_only
raise BrowserConnectError(f'\n{address}浏览器连接失败，请确认浏览器已启动。')
# ----------创建浏览器进程---------- # ----------创建浏览器进程----------
args, user_path = get_launch_args(option) args, user_path = get_launch_args(option)
@ -49,7 +57,12 @@ def connect_browser(option):
raise FileNotFoundError('无法找到浏览器可执行文件路径，请手动配置。') raise FileNotFoundError('无法找到浏览器可执行文件路径，请手动配置。')
_run_browser(port, browser_path, args) _run_browser(port, browser_path, args)
test_connect(ip, port) if not test_connect(ip, port):
raise BrowserConnectError(f'\n{address}浏览器连接失败。\n请确认：\n'
f'1、用户文件夹没有和已打开的浏览器冲突\n'
f'2、如为无界面系统，请添加\'--headless=new\'启动参数\n'
f'3、如果是Linux系统，尝试添加\'--no-sandbox\'启动参数\n'
f'可使用ChromiumOptions设置端口和用户文件夹路径。')
return False return False
@ -186,18 +199,13 @@ def test_connect(ip, port, timeout=30):
if tab['type'] in ('page', 'webview'): if tab['type'] in ('page', 'webview'):
r.close() r.close()
s.close() s.close()
return return True
r.close() r.close()
except Exception: except Exception:
sleep(.2) sleep(.2)
s.close() s.close()
raise BrowserConnectError(f'\n{ip}:{port}浏览器无法链接。\n请确认：\n1、该端口为浏览器\n' return False
f'2、已添加\'--remote-debugging-port={port}\'启动项\n'
f'3、用户文件夹没有和已打开的浏览器冲突\n'
f'4、如为无界面系统，请添加\'--headless=new\'参数\n'
f'5、如果是Linux系统，可能还要添加\'--no-sandbox\'启动参数\n'
f'可使用ChromiumOptions设置端口和用户文件夹路径。')
def _run_browser(port, path: str, args) -> Popen: def _run_browser(port, path: str, args) -> Popen:

2
DrissionPage/_functions/browser.pyi
View File

@ -22,7 +22,7 @@ def set_prefs(opt: ChromiumOptions) -> None: ...
def set_flags(opt: ChromiumOptions) -> None: ... def set_flags(opt: ChromiumOptions) -> None: ...
def test_connect(ip: str, port: Union[int, str], timeout: float = 30) -> None: ... def test_connect(ip: str, port: Union[int, str], timeout: float = 30) -> bool: ...
def get_chrome_path(ini_path: str) -> Union[str, None]: ... def get_chrome_path(ini_path: str) -> Union[str, None]: ...

45
DrissionPage/_pages/chromium_base.py
View File

@ -38,12 +38,12 @@ __ERROR__ = 'error'
class ChromiumBase(BasePage): class ChromiumBase(BasePage):
"""标签页、frame、页面基类""" """标签页、Frame、Page基类"""
def __init__(self, browser, tab_id=None): def __init__(self, browser, target_id=None):
""" """
:param browser: Chromium :param browser: Chromium
:param tab_id: 要控制的标签页id，不指定默认为激活的 :param target_id: 要控制的target id，不指定默认为激活的标签页
""" """
super().__init__() super().__init__()
self._browser = browser self._browser = browser
@ -68,19 +68,19 @@ class ChromiumBase(BasePage):
self._listener = None self._listener = None
self._d_set_runtime_settings() self._d_set_runtime_settings()
self._connect_browser(tab_id) self._connect_browser(target_id)
def _d_set_runtime_settings(self): def _d_set_runtime_settings(self):
pass pass
def _connect_browser(self, tab_id=None): def _connect_browser(self, target_id=None):
"""连接浏览器，在第一次时运行 """连接浏览器，在第一次时运行
:param tab_id: 要控制的标签页id，不指定默认为激活的 :param target_id: 要控制的target id，不指定默认为激活的标签页
:return: None :return: None
""" """
self._is_reading = False self._is_reading = False
if not tab_id: if not target_id:
tabs = self.browser._driver.get(f'http://{self.browser.address}/json').json() tabs = self.browser._driver.get(f'http://{self.browser.address}/json').json()
tabs = [(i['id'], i['url']) for i in tabs tabs = [(i['id'], i['url']) for i in tabs
if i['type'] in ('page', 'webview') and not i['url'].startswith('devtools://')] if i['type'] in ('page', 'webview') and not i['url'].startswith('devtools://')]
@ -89,30 +89,30 @@ class ChromiumBase(BasePage):
for k, t in enumerate(tabs): for k, t in enumerate(tabs):
if t[1] == 'chrome://privacy-sandbox-dialog/notice': if t[1] == 'chrome://privacy-sandbox-dialog/notice':
dialog = k dialog = k
elif not tab_id: elif not target_id:
tab_id = t[0] target_id = t[0]
if tab_id and dialog is not None: if target_id and dialog is not None:
break break
if dialog is not None: if dialog is not None:
close_privacy_dialog(self, tabs[dialog][0]) close_privacy_dialog(self, tabs[dialog][0])
else: else:
tab_id = tabs[0][0] target_id = tabs[0][0]
self._driver_init(tab_id) self._driver_init(target_id)
if self._js_ready_state == 'complete' and self._ready_state is None: if self._js_ready_state == 'complete' and self._ready_state is None:
self._get_document() self._get_document()
self._ready_state = 'complete' self._ready_state = 'complete'
def _driver_init(self, tab_id): def _driver_init(self, target_id):
"""新建页面、页面刷新、切换标签页后要进行的cdp参数初始化 """新建页面、页面刷新、切换标签页后要进行的cdp参数初始化
:param tab_id: 要跳转到的标签页id :param target_id: 要跳转到的target id
:return: None :return: None
""" """
self._is_loading = True self._is_loading = True
self._driver = self.browser._get_driver(tab_id, self) self._driver = self.browser._get_driver(target_id, self)
self._alert = Alert() self._alert = Alert()
self._driver.set_callback('Page.javascriptDialogOpening', self._on_alert_open, immediate=True) self._driver.set_callback('Page.javascriptDialogOpening', self._on_alert_open, immediate=True)
@ -529,7 +529,7 @@ class ChromiumBase(BasePage):
"""获取一个符合条件的元素对象 """获取一个符合条件的元素对象
:param locator: 定位符或元素对象 :param locator: 定位符或元素对象
:param index: 获取第几个元素，从1开始，可传入负数获取倒数第几个 :param index: 获取第几个元素，从1开始，可传入负数获取倒数第几个
:param timeout: 查找超时时间（秒） :param timeout: 查找超时时间（秒），默认与页面等待时间一致
:return: ChromiumElement对象 :return: ChromiumElement对象
""" """
return self._ele(locator, timeout=timeout, index=index, method='ele()') return self._ele(locator, timeout=timeout, index=index, method='ele()')
@ -537,27 +537,30 @@ class ChromiumBase(BasePage):
def eles(self, locator, timeout=None): def eles(self, locator, timeout=None):
"""获取所有符合条件的元素对象 """获取所有符合条件的元素对象
:param locator: 定位符或元素对象 :param locator: 定位符或元素对象
:param timeout: 查找超时时间（秒） :param timeout: 查找超时时间（秒），默认与页面等待时间一致
:return: ChromiumElement对象组成的列表 :return: ChromiumElement对象组成的列表
""" """
return self._ele(locator, timeout=timeout, index=None) return self._ele(locator, timeout=timeout, index=None)
def s_ele(self, locator=None, index=1): def s_ele(self, locator=None, index=1, timeout=None):
"""查找一个符合条件的元素以SessionElement形式返回，处理复杂页面时效率很高 """查找一个符合条件的元素以SessionElement形式返回，处理复杂页面时效率很高
:param locator: 元素的定位信息，可以是loc元组，或查询字符串 :param locator: 元素的定位信息，可以是loc元组，或查询字符串
:param index: 获取第几个，从1开始，可传入负数获取倒数第几个 :param index: 获取第几个，从1开始，可传入负数获取倒数第几个
:param timeout: 查找元素超时时间（秒），默认与页面等待时间一致
:return: SessionElement对象或属性、文本 :return: SessionElement对象或属性、文本
""" """
return (NoneElement(self, method='s_ele()', args={'locator': locator, 'index': index}) return (NoneElement(self, method='s_ele()', args={'locator': locator, 'index': index})
if locator and not self.wait.eles_loaded(locator) if locator and not self.wait.eles_loaded(locator, timeout=timeout)
else make_session_ele(self, locator, index=index, method='s_ele()')) else make_session_ele(self, locator, index=index, method='s_ele()'))
def s_eles(self, locator): def s_eles(self, locator, timeout=None):
"""查找所有符合条件的元素以SessionElement列表形式返回 """查找所有符合条件的元素以SessionElement列表形式返回
:param locator: 元素的定位信息，可以是loc元组，或查询字符串 :param locator: 元素的定位信息，可以是loc元组，或查询字符串
:param timeout: 查找元素超时时间（秒），默认与页面等待时间一致
:return: SessionElement对象组成的列表 :return: SessionElement对象组成的列表
""" """
return make_session_ele(self, locator, index=None) if self.wait.eles_loaded(locator) else SessionElementsList() return (make_session_ele(self, locator, index=None)
if self.wait.eles_loaded(locator, timeout=timeout) else SessionElementsList())
def _find_elements(self, locator, timeout=None, index=1, relative=False, raise_err=None): def _find_elements(self, locator, timeout=None, index=1, relative=False, raise_err=None):
"""执行元素查找 """执行元素查找

15
DrissionPage/_pages/chromium_base.pyi
View File

@ -63,9 +63,9 @@ class ChromiumBase(BasePage):
self._rect: TabRect = ... self._rect: TabRect = ...
self._type: str = ... self._type: str = ...
def _connect_browser(self, tab_id: str = None) -> None: ... def _connect_browser(self, target_id: str = None) -> None: ...
def _driver_init(self, tab_id: str) -> None: ... def _driver_init(self, target_id: str) -> None: ...
def _get_document(self, timeout: float = 10) -> bool: ... def _get_document(self, timeout: float = 10) -> bool: ...
@ -196,9 +196,12 @@ class ChromiumBase(BasePage):
def s_ele(self, def s_ele(self,
locator: Union[Tuple[str, str], str] = None, locator: Union[Tuple[str, str], str] = None,
index: int = 1) -> SessionElement: ... index: int = 1,
timeout: float = None) -> SessionElement: ...
def s_eles(self, locator: Union[Tuple[str, str], str]) -> SessionElementsList: ... def s_eles(self,
locator: Union[Tuple[str, str], str],
timeout: float = None) -> SessionElementsList: ...
def _find_elements(self, def _find_elements(self,
locator: Union[Tuple[str, str], str, ChromiumElement, ChromiumFrame], locator: Union[Tuple[str, str], str, ChromiumElement, ChromiumFrame],
@ -224,7 +227,9 @@ class ChromiumBase(BasePage):
insert_to: Union[ChromiumElement, str, Tuple[str, str], None] = None, insert_to: Union[ChromiumElement, str, Tuple[str, str], None] = None,
before: Union[ChromiumElement, str, Tuple[str, str], None] = None) -> ChromiumElement: ... before: Union[ChromiumElement, str, Tuple[str, str], None] = None) -> ChromiumElement: ...
def get_frame(self, loc_ind_ele: Union[str, int, tuple, ChromiumFrame], timeout: float = None) -> ChromiumFrame: ... def get_frame(self,
loc_ind_ele: Union[str, int, tuple, ChromiumFrame, ChromiumElement],
timeout: float = None) -> ChromiumFrame: ...
def get_frames(self, locator: Union[str, tuple] = None, timeout: float = None) -> List[ChromiumFrame]: ... def get_frames(self, locator: Union[str, tuple] = None, timeout: float = None) -> List[ChromiumFrame]: ...

20
DrissionPage/_pages/chromium_frame.py
View File

@ -38,7 +38,7 @@ class ChromiumFrame(ChromiumBase):
if self._is_inner_frame(): if self._is_inner_frame():
self._is_diff_domain = False self._is_diff_domain = False
self.doc_ele = ChromiumElement(self._target_page, backend_id=node['contentDocument']['backendNodeId']) self.doc_ele = ChromiumElement(self._target_page, backend_id=node['contentDocument']['backendNodeId'])
super().__init__(owner.browser, owner.tab_id) super().__init__(owner.browser, owner.driver.id)
else: else:
self._is_diff_domain = True self._is_diff_domain = True
delattr(self, '_frame_id') delattr(self, '_frame_id')
@ -74,16 +74,16 @@ class ChromiumFrame(ChromiumBase):
self._download_path = self._target_page.download_path self._download_path = self._target_page.download_path
self._load_mode = self._target_page._load_mode if not self._is_diff_domain else 'normal' self._load_mode = self._target_page._load_mode if not self._is_diff_domain else 'normal'
def _driver_init(self, tab_id, is_init=True): def _driver_init(self, target_id, is_init=True):
"""避免出现服务器500错误 """避免出现服务器500错误
:param tab_id: 要跳转到的标签页id :param target_id: 要跳转到的target id
:return: None :return: None
""" """
try: try:
super()._driver_init(tab_id) super()._driver_init(target_id)
except: except:
self.browser._driver.get(f'http://{self._browser.address}/json') self.browser._driver.get(f'http://{self._browser.address}/json')
super()._driver_init(tab_id) super()._driver_init(target_id)
self._driver.set_callback('Inspector.detached', self._onInspectorDetached, immediate=True) self._driver.set_callback('Inspector.detached', self._onInspectorDetached, immediate=True)
self._driver.set_callback('Page.frameDetached', None) self._driver.set_callback('Page.frameDetached', None)
self._driver.set_callback('Page.frameDetached', self._onFrameDetached, immediate=True) self._driver.set_callback('Page.frameDetached', self._onFrameDetached, immediate=True)
@ -310,6 +310,16 @@ class ChromiumFrame(ChromiumBase):
def download_path(self): def download_path(self):
return self._download_path return self._download_path
@property
def sr(self):
"""返回iframe的shadow-root元素对象"""
return self.frame_ele.sr
@property
def shadow_root(self):
"""返回iframe的shadow-root元素对象"""
return self.frame_ele.sr
@property @property
def _js_ready_state(self): def _js_ready_state(self):
"""返回当前页面加载状态，'loading' 'interactive' 'complete'""" """返回当前页面加载状态，'loading' 'interactive' 'complete'"""

10
DrissionPage/_pages/chromium_frame.pyi
View File

@ -10,7 +10,7 @@ from typing import Union, Tuple, List, Any, Optional
from .chromium_base import ChromiumBase from .chromium_base import ChromiumBase
from .tabs import ChromiumTab, MixTab from .tabs import ChromiumTab, MixTab
from .._elements.chromium_element import ChromiumElement from .._elements.chromium_element import ChromiumElement, ShadowRoot
from .._functions.elements import ChromiumElementsList from .._functions.elements import ChromiumElementsList
from .._units.listener import FrameListener from .._units.listener import FrameListener
from .._units.rect import FrameRect from .._units.rect import FrameRect
@ -52,7 +52,7 @@ class ChromiumFrame(ChromiumBase):
def _d_set_runtime_settings(self) -> None: ... def _d_set_runtime_settings(self) -> None: ...
def _driver_init(self, tab_id: str) -> None: ... def _driver_init(self, target_id: str, is_init: bool = True) -> None: ...
def _reload(self) -> None: ... def _reload(self) -> None: ...
@ -128,6 +128,12 @@ class ChromiumFrame(ChromiumBase):
@property @property
def download_path(self) -> str: ... def download_path(self) -> str: ...
@property
def sr(self) -> Union[None, ShadowRoot]: ...
@property
def shadow_root(self) -> Union[None, ShadowRoot]: ...
def refresh(self) -> None: ... def refresh(self) -> None: ...
def property(self, name: str) -> Union[str, None]: ... def property(self, name: str) -> Union[str, None]: ...