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

功能完成

Browse Source
This commit is contained in:
g1879 2022-11-19 11:42:36 +08:00
parent 939a6a3144
commit c4f52a312a
3 changed files with 54 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
DrissionPage/chromium_element.py
View File

@ -1086,7 +1086,6 @@ class ChromiumBase(BasePage):
:param tab_id: 要控制的标签页id，不指定默认为激活的
:return: None
"""
self._is_loading = False
self._root_id = None
self.timeouts = Timeout(self)
self._control_session = Session()
@ -1100,7 +1099,6 @@ class ChromiumBase(BasePage):
self._set_options()
self._init_page(tab_id)
self._get_document()
self._first_run = False
def _init_page(self, tab_id: str = None) -> None:
"""新建页面、页面刷新、切换标签页后要进行的cdp参数初始化
@ -1109,23 +1107,23 @@ class ChromiumBase(BasePage):
"""
self._is_loading = True
if tab_id:
self._driver = Tab(id=tab_id, type='page',
webSocketDebuggerUrl=f'ws://{self.address}/devtools/page/{tab_id}')
self._tab_obj = Tab(id=tab_id, type='page',
webSocketDebuggerUrl=f'ws://{self.address}/devtools/page/{tab_id}')
self._driver.start()
self._driver.DOM.enable()
self._driver.Page.enable()
self._tab_obj.start()
self._tab_obj.DOM.enable()
self._tab_obj.Page.enable()
self._driver.Page.frameNavigated = self._onFrameNavigated
self._driver.Page.loadEventFired = self._onLoadEventFired
# self._driver.DOM.documentUpdated = self._onDocumentUpdated
self._tab_obj.Page.frameNavigated = self._onFrameNavigated
self._tab_obj.Page.loadEventFired = self._onLoadEventFired
# self._tab_obj.DOM.documentUpdated = self._onDocumentUpdated
def _get_document(self) -> None:
"""刷新cdp使用的document数据"""
# print('get doc')
self._wait_loading()
root_id = self._driver.DOM.getDocument()['root']['nodeId']
self._root_id = self._driver.DOM.resolveNode(nodeId=root_id)['object']['objectId']
root_id = self._tab_obj.DOM.getDocument()['root']['nodeId']
self._root_id = self._tab_obj.DOM.resolveNode(nodeId=root_id)['object']['objectId']
self._is_loading = False
def _wait_loading(self, timeout: float = None) -> bool:
@ -1146,7 +1144,9 @@ class ChromiumBase(BasePage):
elif self.page_load_strategy == 'none':
self.stop_loading()
return True
sleep(.1)
self.stop_loading()
return False
def _onLoadEventFired(self, **kwargs):
@ -1161,9 +1161,9 @@ class ChromiumBase(BasePage):
# print('nav')
self._is_loading = True
def _onDocumentUpdated(self, **kwargs):
# print('doc')
pass
# def _onDocumentUpdated(self, **kwargs):
# # print('doc')
# pass
def _set_options(self) -> None:
pass
@ -1181,22 +1181,25 @@ class ChromiumBase(BasePage):
@property
def driver(self) -> Tab:
"""返回用于控制浏览器的Tab对象"""
return self._driver
return self._tab_obj
@property
def _driver(self):
return self._tab_obj
@property
def _wait_driver(self) -> Tab:
"""返回用于控制浏览器的Tab对象，会先等待页面加载完毕"""
while self._is_loading:
# print('loading')
sleep(.1)
return self._driver
self._wait_loading()
return self._tab_obj
@property
def url(self) -> str:
"""返回当前页面url"""
tab_id = self._wait_driver.id # 用于WebPage时激活浏览器
json = loads(self._control_session.get(f'http://{self.address}/json').text)
return [i['url'] for i in json if i['id'] == tab_id][0]
return [i['url'] for i in json if i['id'] == self._tab_obj.id][0] # change_mode要调用，不能用_driver
@property
def html(self) -> str:
@ -1376,7 +1379,7 @@ class ChromiumBase(BasePage):
def _ele(self,
loc_or_ele: Union[Tuple[str, str], str, ChromiumElement, 'ChromiumFrame'],
timeout: float = None, single: bool = True, relative:bool=False) \
timeout: float = None, single: bool = True, relative: bool = False) \
-> Union[ChromiumElement, 'ChromiumFrame', None, List[Union[ChromiumElement, 'ChromiumFrame']]]:
"""执行元素查找
:param loc_or_ele: 定位符或元素对象

18
DrissionPage/chromium_page.py
View File

@ -55,7 +55,6 @@ class ChromiumPage(ChromiumBase):
tab_id = [i['id'] for i in json if i['type'] == 'page'][0]
self._init_page(tab_id)
self._get_document()
self._first_run = False
# 接收浏览器地址和端口
elif isinstance(addr_tab_opts, str):
@ -69,22 +68,21 @@ class ChromiumPage(ChromiumBase):
tab_id = [i['id'] for i in json if i['type'] == 'page'][0]
self._init_page(tab_id)
self._get_document()
self._first_run = False
# 接收传递过来的Tab，浏览器
elif isinstance(addr_tab_opts, Tab):
self._driver = addr_tab_opts
self._tab_obj = addr_tab_opts
self.address = search(r'ws://(.*?)/dev', addr_tab_opts._websocket_url).group(1)
self.process = None
self.options = DriverOptions(read_file=False)
self._set_options()
self._init_page(tab_id)
self._get_document()
self._first_run = False
else:
raise TypeError('只能接收Tab或DriverOptions类型参数。')
self._first_run = False
self._main_tab = self.tab_id
def _init_page(self, tab_id: str = None) -> None:
@ -94,8 +92,8 @@ class ChromiumPage(ChromiumBase):
"""
super()._init_page(tab_id)
self._driver.Page.javascriptDialogOpening = self._on_alert_open
self._driver.Page.javascriptDialogClosed = self._on_alert_close
self._tab_obj.Page.javascriptDialogOpening = self._on_alert_open
self._tab_obj.Page.javascriptDialogClosed = self._on_alert_close
def _set_options(self) -> None:
self.set_timeouts(page_load=self.options.timeouts['pageLoad'] / 1000,
@ -111,7 +109,7 @@ class ChromiumPage(ChromiumBase):
@property
def tabs(self) -> list:
"""返回所有标签页id"""
d = self._wait_driver
self._driver
json = loads(self._control_session.get(f'http://{self.address}/json').text)
return [i['id'] for i in json if i['type'] == 'page']
@ -119,7 +117,7 @@ class ChromiumPage(ChromiumBase):
def process_id(self) -> Union[None, int]:
"""返回浏览器进程id"""
try:
return self._wait_driver.SystemInfo.getProcessInfo()['id']
return self._driver.SystemInfo.getProcessInfo()['id']
except Exception:
return None
@ -300,8 +298,8 @@ class ChromiumPage(ChromiumBase):
def quit(self) -> None:
"""关闭浏览器"""
self._driver.Browser.close()
self._driver.stop()
self._tab_obj.Browser.close()
self._tab_obj.stop()
def _on_alert_close(self, **kwargs):
"""alert关闭时触发的方法"""

27
DrissionPage/web_page.py
View File

@ -1,4 +1,5 @@
# -*- coding:utf-8 -*-
from time import sleep
from typing import Union, Tuple, List
from DownloadKit import DownloadKit
@ -65,7 +66,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
def url(self) -> Union[str, None]:
"""返回当前url"""
if self._mode == 'd':
return super(SessionPage, self).url if self._has_driver else None
return super(SessionPage, self).url if self._tab_obj else None
elif self._mode == 's':
return self._session_url
@ -114,6 +115,19 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
return self._session
@property
def driver(self) -> Tab:
"""返回纯粹的Tab对象"""
return self._tab_obj
@property
def _wait_driver(self) -> Tab:
"""返回用于控制浏览器的Tab对象，会先等待页面加载完毕"""
while self._is_loading:
sleep(.1)
# self._wait_loading()
return self._driver
@property
def _driver(self) -> Tab:
"""返回纯粹的Tab对象，调用时切换到d模式，并连接浏览器"""
@ -214,8 +228,9 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
# s模式转d模式
if self._mode == 'd':
if not self._has_driver:
d = self.driver
if self._tab_obj is None:
self._connect_browser(self._driver_options, self._setting_tab_id)
self._url = None if not self._has_driver else super(SessionPage, self).url
self._has_driver = True
@ -274,7 +289,8 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
return self._get_driver_cookies(as_dict)
def _get_driver_cookies(self, as_dict: bool = False):
cookies = super(SessionPage, self)._wait_driver.Network.getCookies()['cookies']
cookies = self._tab_obj.Network.getCookies()['cookies']
# cookies = super(WebPage, self)._wait_driver.Network.getCookies()['cookies']
if as_dict:
return {cookie['name']: cookie['value'] for cookie in cookies}
else:
@ -292,7 +308,8 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
'name': cookie['name'],
'domain': cookie['domain']}
result_cookies.append(c)
super(SessionPage, self)._wait_driver.Network.setCookies(cookies=result_cookies)
# super(WebPage, self)._wait_driver.Network.setCookies(cookies=result_cookies)
self._tab_obj.Network.setCookies(cookies=result_cookies)
# 添加cookie到session
if set_session: