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

重构ChromiumPage启动逻辑

Browse Source
This commit is contained in:
g1879 2023-10-19 16:20:44 +08:00
parent de6691adc2
commit 955f8c27ae
10 changed files with 83 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
DrissionPage/browser.py
View File

@ -8,18 +8,20 @@ from .chromium_driver import BrowserDriver
class Browser(object): class Browser(object):
BROWSERS = {} BROWSERS = {}
def __new__(cls, browser_id, page): def __new__(cls, address, browser_id, page):
""" """
:param browser_id: BrowserDriver对象 :param address: 浏览器地址
:param browser_id: 浏览器id
:param page: ChromiumPage对象 :param page: ChromiumPage对象
""" """
if browser_id in cls.BROWSERS: if browser_id in cls.BROWSERS:
return cls.BROWSERS[browser_id] return cls.BROWSERS[browser_id]
return object.__new__(cls) return object.__new__(cls)
def __init__(self, browser_id, page): def __init__(self, address, browser_id, page):
""" """
:param page: BrowserDriver对象 :param address: 浏览器地址
:param browser_id: 浏览器id
:param page: ChromiumPage对象 :param page: ChromiumPage对象
""" """
if hasattr(self, '_created'): if hasattr(self, '_created'):
@ -28,10 +30,11 @@ class Browser(object):
Browser.BROWSERS[browser_id] = self Browser.BROWSERS[browser_id] = self
self.page = page self.page = page
self.address = page.address self.address = address
self._driver = BrowserDriver(browser_id, 'browser', page.address) self._driver = BrowserDriver(browser_id, 'browser', address)
self.id = browser_id self.id = browser_id
self._frames = {} self._frames = {}
self._connected = False
self._process_id = None self._process_id = None
r = self.run_cdp('SystemInfo.getProcessInfo') r = self.run_cdp('SystemInfo.getProcessInfo')
@ -40,8 +43,6 @@ class Browser(object):
self._process_id = i['id'] self._process_id = i['id']
break break
self._dl_mgr = BrowserDownloadManager(self)
self.run_cdp('Target.setDiscoverTargets') self.run_cdp('Target.setDiscoverTargets')
self._driver.set_listener('Target.targetDestroyed', self._onTargetDestroyed) self._driver.set_listener('Target.targetDestroyed', self._onTargetDestroyed)
@ -53,6 +54,12 @@ class Browser(object):
if i == tab_id: if i == tab_id:
self._frames.pop(k) self._frames.pop(k)
def connect_to_page(self):
"""执行与page相关的逻辑"""
if not self._connected:
self._dl_mgr = BrowserDownloadManager(self)
self._connected = True
def run_cdp(self, cmd, **cmd_args): def run_cdp(self, cmd, **cmd_args):
"""执行Chrome DevTools Protocol语句 """执行Chrome DevTools Protocol语句
:param cmd: 协议项目 :param cmd: 协议项目

7
DrissionPage/browser.pyi
View File

@ -15,10 +15,11 @@ class Browser(object):
_frames: dict = ... _frames: dict = ...
_process_id: Optional[int] = ... _process_id: Optional[int] = ...
_dl_mgr: BrowserDownloadManager = ... _dl_mgr: BrowserDownloadManager = ...
_connected: bool = ...
def __new__(cls, browser_id: str, page: ChromiumPage): ... def __new__(cls, address: str, browser_id: str, page: ChromiumPage): ...
def __init__(self, browser_id: str, page: ChromiumPage): ... def __init__(self, address: str, browser_id: str, page: ChromiumPage): ...
def run_cdp(self, cmd, **cmd_args) -> dict: ... def run_cdp(self, cmd, **cmd_args) -> dict: ...
@ -43,6 +44,8 @@ class Browser(object):
def get_window_bounds(self) -> dict: ... def get_window_bounds(self) -> dict: ...
def connect_to_page(self) -> None: ...
def _onTargetDestroyed(self, **kwargs): ... def _onTargetDestroyed(self, **kwargs): ...
def quit(self) -> None: ... def quit(self) -> None: ...

28
DrissionPage/chromium_base.py
View File

@ -43,7 +43,7 @@ class ChromiumBase(BasePage):
self._root_id = None # object id self._root_id = None # object id
self._debug = False self._debug = False
self._debug_recorder = None self._debug_recorder = None
self._tab_obj = None self._driver = None
self._set = None self._set = None
self._screencast = None self._screencast = None
self._actions = None self._actions = None
@ -103,18 +103,18 @@ class ChromiumBase(BasePage):
:return: None :return: None
""" """
self._is_loading = True self._is_loading = True
self._tab_obj = ChromiumDriver(tab_id=tab_id, tab_type='page', address=self.address) self._driver = ChromiumDriver(tab_id=tab_id, tab_type='page', address=self.address)
self._tab_obj.call_method('DOM.enable') self._driver.call_method('DOM.enable')
self._tab_obj.call_method('Page.enable') self._driver.call_method('Page.enable')
self._tab_obj.set_listener('Page.frameStoppedLoading', self._onFrameStoppedLoading) self._driver.set_listener('Page.frameStoppedLoading', self._onFrameStoppedLoading)
self._tab_obj.set_listener('Page.frameStartedLoading', self._onFrameStartedLoading) self._driver.set_listener('Page.frameStartedLoading', self._onFrameStartedLoading)
self._tab_obj.set_listener('DOM.documentUpdated', self._onDocumentUpdated) self._driver.set_listener('DOM.documentUpdated', self._onDocumentUpdated)
self._tab_obj.set_listener('Page.loadEventFired', self._onLoadEventFired) self._driver.set_listener('Page.loadEventFired', self._onLoadEventFired)
self._tab_obj.set_listener('Page.frameNavigated', self._onFrameNavigated) self._driver.set_listener('Page.frameNavigated', self._onFrameNavigated)
self._tab_obj.set_listener('Page.frameAttached', self._onFrameAttached) self._driver.set_listener('Page.frameAttached', self._onFrameAttached)
self._tab_obj.set_listener('Page.frameDetached', self._onFrameDetached) self._driver.set_listener('Page.frameDetached', self._onFrameDetached)
def _get_document(self): def _get_document(self):
"""刷新cdp使用的document数据""" """刷新cdp使用的document数据"""
@ -283,9 +283,9 @@ class ChromiumBase(BasePage):
@property @property
def driver(self): def driver(self):
"""返回用于控制浏览器的ChromiumDriver对象""" """返回用于控制浏览器的ChromiumDriver对象"""
if self._tab_obj is None: if self._driver is None:
raise RuntimeError('浏览器已关闭或链接已断开。') raise RuntimeError('浏览器已关闭或链接已断开。')
return self._tab_obj return self._driver
@property @property
def is_loading(self): def is_loading(self):
@ -738,7 +738,7 @@ class ChromiumBase(BasePage):
raise TypeError('必须传入定位符、iframe序号、id、name、ChromiumFrame对象其中之一。') raise TypeError('必须传入定位符、iframe序号、id、name、ChromiumFrame对象其中之一。')
def get_frames(self, loc=None, timeout=None): def get_frames(self, loc=None, timeout=None):
"""获取所有符号条件的frame对象 """获取所有符合条件的frame对象
:param loc: 定位符，为None时返回所有 :param loc: 定位符，为None时返回所有
:param timeout: 查找超时时间 :param timeout: 查找超时时间
:return: ChromiumFrame对象组成的列表 :return: ChromiumFrame对象组成的列表

2
DrissionPage/chromium_base.pyi
View File

@ -30,7 +30,7 @@ class ChromiumBase(BasePage):
self._browser: Browser = ... self._browser: Browser = ...
self._page: ChromiumPage = ... self._page: ChromiumPage = ...
self.address: str = ... self.address: str = ...
self._tab_obj: ChromiumDriver = ... self._driver: ChromiumDriver = ...
self._is_reading: bool = ... self._is_reading: bool = ...
self._timeouts: Timeout = ... self._timeouts: Timeout = ...
self._first_run: bool = ... self._first_run: bool = ...

4
DrissionPage/chromium_frame.py
View File

@ -110,7 +110,7 @@ class ChromiumFrame(ChromiumBase):
self._debug = debug self._debug = debug
else: else:
self._is_diff_domain = True self._is_diff_domain = True
self._tab_obj.stop() self._driver.stop()
super().__init__(self.address, self.frame_id, self._target_page.timeout) super().__init__(self.address, self.frame_id, self._target_page.timeout)
obj_id = super().run_js('document;', as_expr=True)['objectId'] obj_id = super().run_js('document;', as_expr=True)['objectId']
self.doc_ele = ChromiumElement(self, obj_id=obj_id) self.doc_ele = ChromiumElement(self, obj_id=obj_id)
@ -118,7 +118,7 @@ class ChromiumFrame(ChromiumBase):
def _check_ok(self): def _check_ok(self):
"""用于应付同域异域之间跳转导致元素丢失问题""" """用于应付同域异域之间跳转导致元素丢失问题"""
if self._tab_obj._stopped.is_set(): if self._driver._stopped.is_set():
self._reload() self._reload()
try: try:

84
DrissionPage/chromium_page.py
View File

@ -10,11 +10,9 @@ from requests import get
from .browser import Browser from .browser import Browser
from .chromium_base import ChromiumBase, Timeout from .chromium_base import ChromiumBase, Timeout
from .chromium_driver import ChromiumDriver
from .chromium_tab import ChromiumTab from .chromium_tab import ChromiumTab
from .commons.browser import connect_browser from .commons.browser import connect_browser
from .configs.chromium_options import ChromiumOptions from .configs.chromium_options import ChromiumOptions
from .errors import BrowserConnectError
from .setter import ChromiumPageSetter from .setter import ChromiumPageSetter
from .waiter import ChromiumPageWaiter from .waiter import ChromiumPageWaiter
@ -22,41 +20,48 @@ from .waiter import ChromiumPageWaiter
class ChromiumPage(ChromiumBase): class ChromiumPage(ChromiumBase):
"""用于管理浏览器的类""" """用于管理浏览器的类"""
def __init__(self, addr_driver_opts=None, tab_id=None, timeout=None): def __init__(self, addr_or_opts=None, tab_id=None, timeout=None, addr_driver_opts=None):
""" """
:param addr_driver_opts: 浏览器地址:端口、ChromiumDriver对象或ChromiumOptions对象 :param addr_or_opts: 浏览器地址:端口或ChromiumOptions对象
:param tab_id: 要控制的标签页id，不指定默认为激活的 :param tab_id: 要控制的标签页id，不指定默认为激活的
:param timeout: 超时时间 :param timeout: 超时时间
""" """
if addr_driver_opts:
addr_or_opts = addr_driver_opts
self._page = self self._page = self
super().__init__(addr_driver_opts, tab_id) address = self._handle_options(addr_or_opts)
self._run_browser()
super().__init__(address, tab_id)
self.set.timeouts(implicit=timeout) self.set.timeouts(implicit=timeout)
self._page_init()
def _set_start_options(self, addr_driver_opts, none): def _handle_options(self, addr_or_opts):
"""设置浏览器启动属性 """设置浏览器启动属性
:param addr_driver_opts: 'ip:port'、ChromiumOptions :param addr_or_opts: 'ip:port'、ChromiumOptions
:param none: 用于后代继承 :return: 返回浏览器地址
:return: None
""" """
if not addr_driver_opts or isinstance(addr_driver_opts, ChromiumOptions): if not addr_or_opts:
self._driver_options = addr_driver_opts or ChromiumOptions(addr_driver_opts) self._driver_options = ChromiumOptions(addr_or_opts)
elif isinstance(addr_or_opts, ChromiumOptions):
self._driver_options = addr_or_opts
# 接收浏览器地址和端口 # 接收浏览器地址和端口
elif isinstance(addr_driver_opts, str): elif isinstance(addr_or_opts, str):
self._driver_options = ChromiumOptions() self._driver_options = ChromiumOptions()
self._driver_options.debugger_address = addr_driver_opts self._driver_options.debugger_address = addr_or_opts
# 接收传递过来的ChromiumDriver，浏览器
elif isinstance(addr_driver_opts, ChromiumDriver):
self._driver_options = ChromiumOptions(read_file=False)
self._driver_options.debugger_address = addr_driver_opts.address
self._tab_obj = addr_driver_opts
else: else:
raise TypeError('只能接收ChromiumDriver或ChromiumOptions类型参数。') raise TypeError('只能接收ip:port格式或ChromiumOptions类型参数。')
self.address = self._driver_options.debugger_address.replace('localhost', return self._driver_options.debugger_address
'127.0.0.1').lstrip('http://').lstrip('https://')
def _run_browser(self):
"""连接浏览器"""
connect_browser(self._driver_options)
ws = get(f'http://{self._driver_options.debugger_address}/json/version',
headers={'Connection': 'close'}).json()['webSocketDebuggerUrl']
self._browser = Browser(self._driver_options.debugger_address, ws.split('/')[-1], self)
def _set_runtime_settings(self): def _set_runtime_settings(self):
"""设置运行时用到的属性""" """设置运行时用到的属性"""
@ -69,40 +74,17 @@ class ChromiumPage(ChromiumBase):
self._page_load_strategy = self._driver_options.page_load_strategy self._page_load_strategy = self._driver_options.page_load_strategy
self._download_path = str(Path(self._driver_options.download_path).absolute()) self._download_path = str(Path(self._driver_options.download_path).absolute())
def _connect_browser(self, tab_id=None):
"""连接浏览器，在第一次时运行
:param tab_id: 要控制的标签页id，不指定默认为激活的
:return: None
"""
self._chromium_init()
if not self._tab_obj: # 不是传入driver的情况
connect_browser(self._driver_options)
if not tab_id:
json = get(f'http://{self.address}/json', headers={'Connection': 'close'}).json()
tab_id = [i['id'] for i in json if i['type'] == 'page']
if not tab_id:
raise BrowserConnectError('浏览器连接失败，可能是浏览器版本原因。')
tab_id = tab_id[0]
self._driver_init(tab_id)
self._page_init()
self._get_document()
self._first_run = False
def _page_init(self): def _page_init(self):
"""浏览器相关设置""" """浏览器相关设置"""
ws = get(f'http://{self.address}/json/version', headers={'Connection': 'close'}).json()['webSocketDebuggerUrl']
self._browser = Browser(ws.split('/')[-1], self)
self._alert = Alert() self._alert = Alert()
self._tab_obj.set_listener('Page.javascriptDialogOpening', self._on_alert_open) self._driver.set_listener('Page.javascriptDialogOpening', self._on_alert_open)
self._tab_obj.set_listener('Page.javascriptDialogClosed', self._on_alert_close) self._driver.set_listener('Page.javascriptDialogClosed', self._on_alert_close)
self._rect = None self._rect = None
self._main_tab = self.tab_id self._main_tab = self.tab_id
self._browser.connect_to_page()
@property @property
def browser(self): def browser(self):
"""返回用于控制浏览器cdp的driver""" """返回用于控制浏览器cdp的driver"""
@ -327,7 +309,7 @@ class ChromiumPage(ChromiumBase):
self._alert.defaultPrompt = None self._alert.defaultPrompt = None
self._alert.response_accept = kwargs.get('result') self._alert.response_accept = kwargs.get('result')
self._alert.response_text = kwargs['userInput'] self._alert.response_text = kwargs['userInput']
self._tab_obj.has_alert = False self._driver.has_alert = False
def _on_alert_open(self, **kwargs): def _on_alert_open(self, **kwargs):
"""alert出现时触发的方法""" """alert出现时触发的方法"""
@ -337,7 +319,7 @@ class ChromiumPage(ChromiumBase):
self._alert.defaultPrompt = kwargs.get('defaultPrompt', None) self._alert.defaultPrompt = kwargs.get('defaultPrompt', None)
self._alert.response_accept = None self._alert.response_accept = None
self._alert.response_text = None self._alert.response_text = None
self._tab_obj.has_alert = True self._driver.has_alert = True
class ChromiumTabRect(object): class ChromiumTabRect(object):

9
DrissionPage/chromium_page.pyi
View File

@ -7,7 +7,6 @@ from typing import Union, Tuple, List, Optional
from .browser import Browser from .browser import Browser
from .chromium_base import ChromiumBase from .chromium_base import ChromiumBase
from .chromium_driver import ChromiumDriver
from .chromium_tab import ChromiumTab from .chromium_tab import ChromiumTab
from .configs.chromium_options import ChromiumOptions from .configs.chromium_options import ChromiumOptions
from .setter import ChromiumPageSetter from .setter import ChromiumPageSetter
@ -17,7 +16,7 @@ from .waiter import ChromiumPageWaiter
class ChromiumPage(ChromiumBase): class ChromiumPage(ChromiumBase):
def __init__(self, def __init__(self,
addr_driver_opts: Union[str, int, ChromiumOptions, ChromiumDriver] = None, addr_or_opts: Union[str, int, ChromiumOptions] = None,
tab_id: str = None, tab_id: str = None,
timeout: float = None): timeout: float = None):
self._driver_options: ChromiumOptions = ... self._driver_options: ChromiumOptions = ...
@ -26,11 +25,9 @@ class ChromiumPage(ChromiumBase):
self._browser: Browser = ... self._browser: Browser = ...
self._rect: ChromiumTabRect = ... self._rect: ChromiumTabRect = ...
def _connect_browser(self, def _handle_options(self, addr_or_opts) -> str: ...
addr_driver_opts: Union[str, ChromiumDriver] = None,
tab_id: str = None) -> None: ...
def _set_start_options(self, addr_driver_opts: Union[str, ChromiumDriver], none) -> None: ... def _run_browser(self) -> None: ...
def _page_init(self) -> None: ... def _page_init(self) -> None: ...

4
DrissionPage/chromium_tab.py
View File

@ -109,7 +109,7 @@ class WebPageTab(SessionPage, ChromiumTab):
@property @property
def _browser_url(self): def _browser_url(self):
"""返回浏览器当前url""" """返回浏览器当前url"""
return super(SessionPage, self).url if self._tab_obj else None return super(SessionPage, self).url if self._driver else None
@property @property
def title(self): def title(self):
@ -279,7 +279,7 @@ class WebPageTab(SessionPage, ChromiumTab):
# s模式转d模式 # s模式转d模式
if self._mode == 'd': if self._mode == 'd':
if self._tab_obj is None: if self._driver is None:
self._connect_browser(self.page._driver_options) self._connect_browser(self.page._driver_options)
self._url = None if not self._has_driver else super(SessionPage, self).url self._url = None if not self._has_driver else super(SessionPage, self).url

14
DrissionPage/web_page.py
View File

@ -39,7 +39,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
self.address = None self.address = None
self._session = None self._session = None
self._tab_obj = None self._driver = None
self._driver_options = None self._driver_options = None
self._session_options = None self._session_options = None
self._response = None self._response = None
@ -62,7 +62,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
""" """
# 浏览器配置 # 浏览器配置
if isinstance(dr_opt, ChromiumDriver): if isinstance(dr_opt, ChromiumDriver):
self._tab_obj = dr_opt self._driver = dr_opt
self._driver_options = ChromiumOptions() self._driver_options = ChromiumOptions()
self._driver_options.debugger_address = dr_opt.address self._driver_options.debugger_address = dr_opt.address
dr_opt = False dr_opt = False
@ -141,7 +141,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
@property @property
def _browser_url(self): def _browser_url(self):
"""返回浏览器当前url""" """返回浏览器当前url"""
return super(SessionPage, self).url if self._tab_obj else None return super(SessionPage, self).url if self._driver else None
@property @property
def title(self): def title(self):
@ -311,7 +311,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
# s模式转d模式 # s模式转d模式
if self._mode == 'd': if self._mode == 'd':
if self._tab_obj is None: if self._driver is None:
self._connect_browser(self._driver_options) self._connect_browser(self._driver_options)
self._url = None if not self._has_driver else super(SessionPage, self).url self._url = None if not self._has_driver else super(SessionPage, self).url
@ -393,8 +393,8 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
self.driver.call_method('Browser.close') self.driver.call_method('Browser.close')
except Exception: except Exception:
pass pass
self._tab_obj.stop() self._driver.stop()
self._tab_obj = None self._driver = None
self._has_driver = None self._has_driver = None
def close_session(self): def close_session(self):
@ -430,5 +430,5 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
self._has_session = None self._has_session = None
if self._has_driver: if self._has_driver:
super(SessionPage, self).quit() super(SessionPage, self).quit()
self._tab_obj = None self._driver = None
self._has_driver = None self._has_driver = None

2
DrissionPage/web_page.pyi
View File

@ -36,7 +36,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
self._driver_options: Union[ChromiumOptions, None] = ... self._driver_options: Union[ChromiumOptions, None] = ...
self._DownloadKit: DownloadKit = ... self._DownloadKit: DownloadKit = ...
self._download_path: str = ... self._download_path: str = ...
self._tab_obj: ChromiumDriver = ... self._driver: ChromiumDriver = ...
self._frames: dict = ... self._frames: dict = ...
def __call__(self, def __call__(self,