From 9f49f874cabe405f9c45889e313774b11fcd177d Mon Sep 17 00:00:00 2001 From: g1879 Date: Fri, 28 Jun 2024 15:39:09 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AF=B9Page=E5=AF=B9=E8=B1=A1=E8=BF=9B?= =?UTF-8?q?=E8=A1=8C=E8=A7=A3=E8=80=A6=EF=BC=8C=E6=9C=AA=E5=AE=8C=E6=88=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- DrissionPage/__init__.py | 14 +- DrissionPage/_base/base.py | 7 +- DrissionPage/_base/base.pyi | 2 +- DrissionPage/_base/browser.py | 403 ++++++++++++++++---- DrissionPage/_base/browser.pyi | 121 +++++- DrissionPage/_configs/chromium_options.py | 7 +- DrissionPage/_configs/session_options.py | 8 +- DrissionPage/_elements/chromium_element.pyi | 2 +- DrissionPage/_pages/chromium_base.py | 15 +- DrissionPage/_pages/chromium_base.pyi | 6 +- DrissionPage/_pages/chromium_frame.py | 27 +- DrissionPage/_pages/chromium_frame.pyi | 13 +- DrissionPage/_pages/chromium_page.py | 28 +- DrissionPage/_pages/chromium_page.pyi | 14 +- DrissionPage/_pages/chromium_tab.py | 57 ++- DrissionPage/_pages/chromium_tab.pyi | 30 +- DrissionPage/_pages/web_page.py | 18 +- DrissionPage/_pages/web_page.pyi | 12 +- DrissionPage/_units/clicker.py | 24 +- DrissionPage/_units/clicker.pyi | 6 +- DrissionPage/_units/cookies_setter.pyi | 4 +- DrissionPage/_units/downloader.py | 41 +- DrissionPage/_units/downloader.pyi | 12 +- DrissionPage/_units/rect.py | 2 +- DrissionPage/_units/rect.pyi | 4 +- DrissionPage/_units/screencast.py | 4 +- DrissionPage/_units/setter.py | 181 ++++++--- DrissionPage/_units/setter.pyi | 52 ++- DrissionPage/_units/waiter.py | 78 ++++ DrissionPage/_units/waiter.pyi | 12 + DrissionPage/items.py | 4 +- 31 files changed, 853 insertions(+), 355 deletions(-) diff --git a/DrissionPage/__init__.py b/DrissionPage/__init__.py index e173e0d..66b3008 100644 --- a/DrissionPage/__init__.py +++ b/DrissionPage/__init__.py @@ -5,13 +5,13 @@ @Copyright: (c) 2024 by g1879, Inc. All Rights Reserved. @License : BSD 3-Clause. """ -from ._pages.chromium_page import ChromiumPage -from ._pages.session_page import SessionPage -from ._pages.web_page import WebPage - -# 启动配置类 +from ._base.browser import Browser from ._configs.chromium_options import ChromiumOptions from ._configs.session_options import SessionOptions +from ._pages.session_page import SessionPage -__all__ = ['ChromiumPage', 'ChromiumOptions', 'SessionOptions', 'SessionPage', 'WebPage', '__version__'] -__version__ = '4.0.5.4' +# from ._pages.chromium_page import ChromiumPage +# from ._pages.web_page import WebPage + +__all__ = ['Browser', 'ChromiumOptions', 'SessionOptions', 'SessionPage', '__version__'] +__version__ = '4.0.5.3' diff --git a/DrissionPage/_base/base.py b/DrissionPage/_base/base.py index 1fcbd40..c30293e 100644 --- a/DrissionPage/_base/base.py +++ b/DrissionPage/_base/base.py @@ -54,7 +54,7 @@ class BaseElement(BaseParser): def __init__(self, owner=None): self.owner = owner - self.page = owner._page if owner else None + # self.page = owner._page if owner else None self._type = 'BaseElement' # ----------------以下属性或方法由后代实现---------------- @@ -366,11 +366,6 @@ class BasePage(BaseParser): """返回查找元素时等待的秒数""" return self._timeout - @timeout.setter - def timeout(self, second): - """设置查找元素时等待的秒数""" - self._timeout = second - @property def url_available(self): """返回当前访问的url有效性""" diff --git a/DrissionPage/_base/base.pyi b/DrissionPage/_base/base.pyi index 8de4f35..764ccb3 100644 --- a/DrissionPage/_base/base.pyi +++ b/DrissionPage/_base/base.pyi @@ -59,7 +59,7 @@ class BaseElement(BaseParser): def __init__(self, owner: BasePage = None): self.owner: BasePage = ... - self.page: Union[ChromiumPage, SessionPage, WebPage] = ... + # self.page: Union[ChromiumPage, SessionPage, WebPage] = ... # ----------------以下属性或方法由后代实现---------------- @property diff --git a/DrissionPage/_base/browser.py b/DrissionPage/_base/browser.py index 00649d5..1ae5c87 100644 --- a/DrissionPage/_base/browser.py +++ b/DrissionPage/_base/browser.py @@ -7,50 +7,80 @@ """ from pathlib import Path from shutil import rmtree -from time import perf_counter, sleep +from threading import Lock +from time import sleep, perf_counter +from requests import Session from websocket import WebSocketBadStatusException from .driver import BrowserDriver, Driver +from .._configs.chromium_options import ChromiumOptions +from .._configs.session_options import SessionOptions +from .._functions.browser import connect_browser +from .._functions.settings import Settings +from .._functions.tools import PortFinder from .._functions.tools import raise_error +from .._pages.chromium_base import Timeout +from .._pages.chromium_tab import ChromiumTab, MixTab from .._units.downloader import DownloadManager +from .._units.setter import BrowserSetter +from .._units.waiter import BrowserWaiter +from ..errors import BrowserConnectError from ..errors import PageDisconnectedError __ERROR__ = 'error' class Browser(object): - BROWSERS = {} + _BROWSERS = {} - def __new__(cls, address, browser_id, page): + def __new__(cls, addr_or_opts=None, session_options=None): """ - :param address: 浏览器地址 - :param browser_id: 浏览器id - :param page: ChromiumPage对象 + :param addr_or_opts: 浏览器地址:端口、ChromiumOptions对象或端口数字（int） + :param session_options: 使用双模Tab时使用的默认Session配置，为True使用ini文件配置 """ - if browser_id in cls.BROWSERS: - return cls.BROWSERS[browser_id] - return object.__new__(cls) + opt = handle_options(addr_or_opts) + is_exist, browser_id = run_browser(opt) + if browser_id in cls._BROWSERS: + r = cls._BROWSERS[browser_id] + return r + r = object.__new__(cls) + r._chromium_options = opt + r._is_exist = is_exist + r.id = browser_id + r.address = opt.address + cls._BROWSERS[browser_id] = r + return r - def __init__(self, address, browser_id, page): + def __init__(self, addr_or_opts=None, session_options=None): """ - :param address: 浏览器地址 - :param browser_id: 浏览器id - :param page: ChromiumPage对象 + :param addr_or_opts: 浏览器地址:端口、ChromiumOptions对象或端口数字（int） + :param session_options: 使用双模Tab时使用的默认Session配置，为True使用ini文件配置 """ if hasattr(self, '_created'): return self._created = True - Browser.BROWSERS[browser_id] = self - self.page = page - self.address = address - self._driver = BrowserDriver(browser_id, 'browser', address, self) - self.id = browser_id + self._type = 'Browser' + self._driver = BrowserDriver(self.id, 'browser', self.address, self) + self.version = self.run_cdp('Browser.getVersion')['product'] + self._frames = {} self._drivers = {} self._all_drivers = {} - self._connected = False + self._lock = Lock() + + self._set = None + self._wait = None + self._timeouts = Timeout(page_load=self._chromium_options.timeouts['page_load'], + script=self._chromium_options.timeouts['script'], + base=self._chromium_options.timeouts['base']) + if self._chromium_options.timeouts['base'] is not None: + self._timeout = self._chromium_options.timeouts['base'] + self._load_mode = self._chromium_options.load_mode + self._download_path = str(Path(self._chromium_options.download_path).absolute()) + self.retry_times = self._chromium_options.retry_times + self.retry_interval = self._chromium_options.retry_interval self._process_id = None try: @@ -65,6 +95,9 @@ class Browser(object): self.run_cdp('Target.setDiscoverTargets', discover=True) self._driver.set_callback('Target.targetDestroyed', self._onTargetDestroyed) self._driver.set_callback('Target.targetCreated', self._onTargetCreated) + self._dl_mgr = DownloadManager(self) + + self._session_options = SessionOptions() if session_options is True else session_options def _get_driver(self, tab_id, owner=None): """新建并返回指定tab id的Driver @@ -104,12 +137,6 @@ class Browser(object): self._drivers.pop(tab_id, None) self._all_drivers.pop(tab_id, None) - def connect_to_page(self): - """执行与page相关的逻辑""" - if not self._connected: - self._dl_mgr = DownloadManager(self) - self._connected = True - def run_cdp(self, cmd, **cmd_args): """执行Chrome DevTools Protocol语句 :param cmd: 协议项目 @@ -121,8 +148,37 @@ class Browser(object): return r if __ERROR__ not in r else raise_error(r, ignore) @property - def driver(self): - return self._driver + def process_id(self): + """返回浏览器进程id""" + return self._process_id + + @property + def timeout(self): + """返回timeouts设置""" + return self._timeouts.base + + @property + def timeouts(self): + """返回timeouts设置""" + return self._timeouts + + @property + def download_path(self): + """返回默认下载路径""" + return self._download_path + + @property + def set(self): + if self._set is None: + self._set = BrowserSetter(self) + return self._set + + @property + def wait(self): + """返回用于等待的对象""" + if self._wait is None: + self._wait = BrowserWaiter(self) + return self._wait @property def tabs_count(self): @@ -138,16 +194,154 @@ class Browser(object): and not i['url'].startswith('devtools://')] @property - def process_id(self): - """返回浏览器进程id""" - return self._process_id + def latest_tab(self): + """返回最新的标签页，最新标签页指最后创建或最后被激活的 + 当Settings.singleton_tab_obj==True时返回Tab对象，否则返回tab id""" + return self.get_tab(self.tab_ids[0], as_id=not Settings.singleton_tab_obj) - def find_tabs(self, title=None, url=None, tab_type=None): + def new_tab(self, url=None, new_window=False, background=False, + new_context=False): + """新建一个标签页 + :param url: 新标签页跳转到的网址 + :param new_window: 是否在新窗口打开标签页 + :param background: 是否不激活新标签页，如new_window为True则无效 + :param new_context: 是否创建新的上下文 + :return: 新标签页对象 + """ + return self._new_tab(ChromiumTab, url=url, new_window=new_window, + background=background, new_context=new_context) + + def new_mix_tab(self, url=None, new_window=False, background=False, + new_context=False): + """新建一个标签页 + :param url: 新标签页跳转到的网址 + :param new_window: 是否在新窗口打开标签页 + :param background: 是否不激活新标签页，如new_window为True则无效 + :param new_context: 是否创建新的上下文 + :return: 新标签页对象 + """ + return self._new_tab(MixTab, url=url, new_window=new_window, + background=background, new_context=new_context) + + def _new_tab(self, obj, url=None, new_window=False, background=False, + new_context=False): + """新建一个标签页 + :param obj: 要创建的Tab类型 + :param url: 新标签页跳转到的网址 + :param new_window: 是否在新窗口打开标签页 + :param background: 是否不激活新标签页，如new_window为True则无效 + :param new_context: 是否创建新的上下文 + :return: 新标签页对象 + """ + tab = None + if new_context: + tab = self.run_cdp('Target.createBrowserContext')['browserContextId'] + + kwargs = {'url': ''} + if new_window: + kwargs['newWindow'] = True + if background: + kwargs['background'] = True + if tab: + kwargs['browserContextId'] = tab + + tab = self.run_cdp('Target.createTarget', **kwargs)['targetId'] + while tab not in self._drivers: + sleep(.1) + tab = obj(self, tab) + if url: + tab.get(url) + return tab + + def get_tab(self, id_or_num=None, title=None, url=None, tab_type='page', as_id=False): + """获取一个标签页对象，id_or_num不为None时，后面几个参数无效 + :param id_or_num: 要获取的标签页id或序号，序号从1开始，可传入负数获取倒数第几个，不是视觉排列顺序，而是激活顺序 + :param title: 要匹配title的文本，模糊匹配，为None则匹配所有 + :param url: 要匹配url的文本，模糊匹配，为None则匹配所有 + :param tab_type: tab类型，可用列表输入多个，如 'page', 'iframe' 等，为None则匹配所有 + :param as_id: 是否返回标签页id而不是标签页对象，dual_mode=False时无效 + :return: Tab对象 + """ + return self._get_tab(id_or_num=id_or_num, title=title, url=url, tab_type=tab_type, as_id=as_id) + + def get_tabs(self, title=None, url=None, tab_type='page', as_id=False): """查找符合条件的tab，返回它们组成的列表，title和url是与关系 :param title: 要匹配title的文本 :param url: 要匹配url的文本 :param tab_type: tab类型，可用列表输入多个 - :return: dict格式的tab信息列表列表 + :param as_id: 是否返回标签页id而不是标签页对象，dual_mode=False时无效 + :return: Tab对象列表 + """ + return self._get_tabs(title=title, url=url, tab_type=tab_type, as_id=as_id) + + def get_mix_tab(self, id_or_num=None, title=None, url=None, tab_type='page', as_id=False): + """获取一个标签页对象，id_or_num不为None时，后面几个参数无效 + :param id_or_num: 要获取的标签页id或序号，序号从1开始，可传入负数获取倒数第几个，不是视觉排列顺序，而是激活顺序 + :param title: 要匹配title的文本，模糊匹配，为None则匹配所有 + :param url: 要匹配url的文本，模糊匹配，为None则匹配所有 + :param tab_type: tab类型，可用列表输入多个，如 'page', 'iframe' 等，为None则匹配所有 + :param as_id: 是否返回标签页id而不是标签页对象，dual_mode=False时无效 + :return: Tab对象 + """ + return self._get_tab(id_or_num=id_or_num, title=title, url=url, tab_type=tab_type, dual_mode=True, as_id=as_id) + + def get_mix_tabs(self, title=None, url=None, tab_type='page', as_id=False): + """查找符合条件的tab，返回它们组成的列表，title和url是与关系 + :param title: 要匹配title的文本 + :param url: 要匹配url的文本 + :param tab_type: tab类型，可用列表输入多个 + :param as_id: 是否返回标签页id而不是标签页对象，dual_mode=False时无效 + :return: Tab对象列表 + """ + return self._get_tabs(title=title, url=url, tab_type=tab_type, dual_mode=True, as_id=as_id) + + def _get_tab(self, id_or_num=None, title=None, url=None, tab_type='page', + dual_mode=False, as_id=False): + """获取一个标签页对象，id_or_num不为None时，后面几个参数无效 + :param id_or_num: 要获取的标签页id或序号，序号从1开始，可传入负数获取倒数第几个，不是视觉排列顺序，而是激活顺序 + :param title: 要匹配title的文本，模糊匹配，为None则匹配所有 + :param url: 要匹配url的文本，模糊匹配，为None则匹配所有 + :param tab_type: tab类型，可用列表输入多个，如 'page', 'iframe' 等，为None则匹配所有 + :param dual_mode: 是否返回可切换模式的Tab对象 + :param as_id: 是否返回标签页id而不是标签页对象，dual_mode=False时无效 + :return: Tab对象 + """ + if id_or_num is not None: + if isinstance(id_or_num, str): + id_or_num = id_or_num + elif isinstance(id_or_num, int): + id_or_num = self.tab_ids[id_or_num - 1 if id_or_num > 0 else id_or_num] + elif isinstance(id_or_num, ChromiumTab): + if as_id: + return id_or_num.tab_id + elif Settings.singleton_tab_obj: + return id_or_num + else: + return self._get_tab(id_or_num.tab_id) # todo: 循环调用 + + elif title == url == tab_type is None: + id_or_num = self.tab_ids[0] + + else: + tabs = self._get_tabs(title=title, url=url, tab_type=tab_type, as_id=True) + if tabs: + id_or_num = tabs[0] + else: + return None + + if as_id: + return id_or_num + with self._lock: + return MixTab(self, id_or_num) if dual_mode else ChromiumTab(self, id_or_num) + + def _get_tabs(self, title=None, url=None, tab_type='page', dual_mode=False, as_id=False): + """查找符合条件的tab，返回它们组成的列表，title和url是与关系 + :param title: 要匹配title的文本 + :param url: 要匹配url的文本 + :param tab_type: tab类型，可用列表输入多个 + :param dual_mode: 是否返回可切换模式的Tab对象 + :param as_id: 是否返回标签页id而不是标签页对象，dual_mode=False时无效 + :return: Tab对象列表 """ tabs = self._driver.get(f'http://{self.address}/json').json() # 不要改用cdp @@ -158,24 +352,49 @@ class Browser(object): elif tab_type is not None: raise TypeError('tab_type只能是set、list、tuple、str、None。') - return [i for i in tabs if ((title is None or title in i['title']) and (url is None or url in i['url']) + tabs = [i for i in tabs if ((title is None or title in i['title']) and (url is None or url in i['url']) and (tab_type is None or i['type'] in tab_type))] + if as_id: + return [tab['id'] for tab in tabs] + with self._lock: + if dual_mode: + return [MixTab(self, tab['id']) for tab in tabs] + else: + return [ChromiumTab(self, tab['id']) for tab in tabs] - def close_tab(self, tab_id): - """关闭标签页 - :param tab_id: 标签页id + def close_tabs(self, tabs_or_ids=None, others=False): + """关闭传入的标签页，默认关闭当前页。可传入多个 + :param tabs_or_ids: 要关闭的标签页对象或id，可传入列表或元组，为None时关闭最后操作的 + :param others: 是否关闭指定标签页之外的 :return: None """ - self._onTargetDestroyed(targetId=tab_id) - self.driver.run('Target.closeTarget', targetId=tab_id) + all_tabs = set(self.tab_ids) + if isinstance(tabs_or_ids, str): + tabs = {tabs_or_ids} + elif isinstance(tabs_or_ids, ChromiumTab): + tabs = {tabs_or_ids.tab_id} + elif tabs_or_ids is None: + tabs = {self.tab_ids[0]} + elif isinstance(tabs_or_ids, (list, tuple)): + tabs = set(i.tab_id if isinstance(i, ChromiumTab) else i for i in tabs_or_ids) + else: + raise TypeError('tabs_or_ids参数只能传入标签页对象或id。') - def stop_driver(self, driver): - """停止一个Driver - :param driver: Driver对象 - :return: None - """ - driver.stop() - self._all_drivers.get(driver.id, set()).discard(driver) + if others: + tabs = all_tabs - tabs + + end_len = len(set(all_tabs) - set(tabs)) + if end_len <= 0: + self.quit() + return + + for tab in tabs: + self._onTargetDestroyed(targetId=tab) + self._driver.run('Target.closeTarget', targetId=tab) + sleep(.2) + end_time = perf_counter() + 3 + while self.tabs_count != end_len and perf_counter() < end_time: + sleep(.1) def activate_tab(self, tab_id): """使标签页变为活动状态 @@ -184,37 +403,6 @@ class Browser(object): """ self.run_cdp('Target.activateTarget', targetId=tab_id) - def get_window_bounds(self, tab_id=None): - """返回浏览器窗口位置和大小信息 - :param tab_id: 标签页id - :return: 窗口大小字典 - """ - return self.run_cdp('Browser.getWindowForTarget', targetId=tab_id or self.id)['bounds'] - - def new_tab(self, new_window=False, background=False, new_context=False): - """新建一个标签页 - :param new_window: 是否在新窗口打开标签页 - :param background: 是否不激活新标签页，如new_window为True则无效 - :param new_context: 是否创建新的上下文 - :return: 新标签页id - """ - bid = None - if new_context: - bid = self.run_cdp('Target.createBrowserContext')['browserContextId'] - - kwargs = {'url': ''} - if new_window: - kwargs['newWindow'] = True - if background: - kwargs['background'] = True - if bid: - kwargs['browserContextId'] = bid - - tid = self.run_cdp('Target.createTarget', **kwargs)['targetId'] - while tid not in self._drivers: - sleep(.1) - return tid - def reconnect(self): """断开重连""" self._driver.stop() @@ -234,7 +422,7 @@ class Browser(object): self.run_cdp('Browser.close') except PageDisconnectedError: pass - self.driver.stop() + self._driver.stop() drivers = list(self._all_drivers.values()) for tab in drivers: @@ -276,10 +464,9 @@ class Browser(object): break def _on_disconnect(self): - self.page._on_disconnect() - Browser.BROWSERS.pop(self.id, None) - if self.page._chromium_options.is_auto_port and self.page._chromium_options.user_data_path: - path = Path(self.page._chromium_options.user_data_path) + Browser._BROWSERS.pop(self.id, None) + if self._chromium_options.is_auto_port and self._chromium_options.user_data_path: + path = Path(self._chromium_options.user_data_path) end_time = perf_counter() + 7 while perf_counter() < end_time: if not path.exists(): @@ -290,3 +477,57 @@ class Browser(object): except (PermissionError, FileNotFoundError, OSError): pass sleep(.03) + + +def handle_options(addr_or_opts): + """设置浏览器启动属性 + :param addr_or_opts: 'ip:port'、ChromiumOptions、Driver + :return: 返回ChromiumOptions对象 + """ + if not addr_or_opts: + _chromium_options = ChromiumOptions(addr_or_opts) + if _chromium_options.is_auto_port: + port, path = PortFinder(_chromium_options.tmp_path).get_port(_chromium_options.is_auto_port) + _chromium_options.set_address(f'127.0.0.1:{port}') + _chromium_options.set_user_data_path(path) + _chromium_options.auto_port(scope=_chromium_options.is_auto_port) + + elif isinstance(addr_or_opts, ChromiumOptions): + if addr_or_opts.is_auto_port: + port, path = PortFinder(addr_or_opts.tmp_path).get_port(addr_or_opts.is_auto_port) + addr_or_opts.set_address(f'127.0.0.1:{port}') + addr_or_opts.set_user_data_path(path) + addr_or_opts.auto_port(scope=addr_or_opts.is_auto_port) + _chromium_options = addr_or_opts + + elif isinstance(addr_or_opts, str): + _chromium_options = ChromiumOptions() + _chromium_options.set_address(addr_or_opts) + + elif isinstance(addr_or_opts, int): + _chromium_options = ChromiumOptions() + _chromium_options.set_local_port(addr_or_opts) + + else: + raise TypeError('只能接收ip:port格式或ChromiumOptions类型参数。') + + return _chromium_options + + +def run_browser(chromium_options): + """连接浏览器""" + is_exist = connect_browser(chromium_options) + try: + s = Session() + s.trust_env = False + ws = s.get(f'http://{chromium_options.address}/json/version', headers={'Connection': 'close'}) + if not ws: + raise BrowserConnectError('\n浏览器连接失败，如使用全局代理，须设置不代理127.0.0.1地址。') + browser_id = ws.json()['webSocketDebuggerUrl'].split('/')[-1] + ws.close() + s.close() + except KeyError: + raise BrowserConnectError('浏览器版本太旧或此浏览器不支持接管。') + except: + raise BrowserConnectError('\n浏览器连接失败，如使用全局代理，须设置不代理127.0.0.1地址。') + return is_exist, browser_id diff --git a/DrissionPage/_base/browser.pyi b/DrissionPage/_base/browser.pyi index 170f88b..9973f82 100644 --- a/DrissionPage/_base/browser.pyi +++ b/DrissionPage/_base/browser.pyi @@ -5,36 +5,71 @@ @Copyright: (c) 2024 by g1879, Inc. All Rights Reserved. @License : BSD 3-Clause. """ -from typing import List, Optional, Union, Set, Dict +from threading import Lock +from typing import List, Optional, Set, Dict, Union, Tuple from .driver import BrowserDriver, Driver -from .._pages.chromium_page import ChromiumPage +from .._configs.chromium_options import ChromiumOptions +from .._configs.session_options import SessionOptions +from .._pages.chromium_base import Timeout, ChromiumBase +from .._pages.chromium_tab import ChromiumTab, MixTab from .._units.downloader import DownloadManager +from .._units.setter import BrowserSetter +from .._units.waiter import BrowserWaiter class Browser(object): - BROWSERS: dict = ... - page: ChromiumPage = ... - _driver: BrowserDriver = ... id: str = ... address: str = ... + version: str = ... + retry_times: int = ... + retry_interval: float = ... + + _BROWSERS: dict = ... + _chromium_options: ChromiumOptions = ... + _session_options: SessionOptions = ... + _driver: BrowserDriver = ... _frames: dict = ... _drivers: Dict[str, Driver] = ... _all_drivers: Dict[str, Set[Driver]] = ... _process_id: Optional[int] = ... _dl_mgr: DownloadManager = ... - _connected: bool = ... + _lock: Lock = ... - def __new__(cls, address: str, browser_id: str, page: ChromiumPage): ... + _set: Optional[BrowserSetter] = ... + _wait: Optional[BrowserWaiter] = ... + _timeouts: Timeout = ... + _load_mode: str = ... + _download_path: str = ... - def __init__(self, address: str, browser_id: str, page: ChromiumPage): ... + def __new__(cls, + addr_or_opts: Union[str, int, ChromiumOptions] = None, + session_options: Optional[SessionOptions] = None): ... + + def __init__(self, addr_or_opts: Union[str, int, ChromiumOptions] = None, + session_options: Optional[SessionOptions] = None): ... def _get_driver(self, tab_id: str, owner=None) -> Driver: ... def run_cdp(self, cmd, **cmd_args) -> dict: ... @property - def driver(self) -> BrowserDriver: ... + def process_id(self) -> Optional[int]: ... + + @property + def timeout(self) -> float: ... + + @property + def timeouts(self) -> Timeout: ... + + @property + def download_path(self) -> str: ... + + @property + def set(self) -> BrowserSetter: ... + + @property + def wait(self) -> BrowserWaiter: ... @property def tabs_count(self) -> int: ... @@ -43,25 +78,75 @@ class Browser(object): def tab_ids(self) -> List[str]: ... @property - def process_id(self) -> Optional[int]: ... + def latest_tab(self) -> Union[ChromiumTab, str]: ... - def find_tabs(self, title: str = None, url: str = None, - tab_type: Union[str, list, tuple] = None) -> List[dict]: ... + def close_tabs(self, + tabs_or_ids: Union[str, ChromiumTab, List[Union[str, ChromiumTab]], + Tuple[Union[str, ChromiumTab]]] = None, + others: bool = False) -> None: ... - def close_tab(self, tab_id: str) -> None: ... + def get_tab(self, + id_or_num: Union[str, int] = None, + title: str = None, + url: str = None, + tab_type: str = 'page', + as_id: bool = False) -> Union[ChromiumTab, str]: ... - def stop_driver(self, driver: Driver) -> None: ... + def get_tabs(self, + title: str = None, + url: str = None, + tab_type: str = 'page', + as_id: bool = False) -> List[ChromiumTab, str]: ... + + def get_mix_tab(self, + id_or_num: Union[str, int] = None, + title: str = None, + url: str = None, + tab_type: str = 'page') -> Union[MixTab, str]: ... + + def get_mix_tabs(self, + title: str = None, + url: str = None, + tab_type: str = 'page') -> List[MixTab, str]: ... + + def _get_tab(self, + id_or_num: Union[str, int] = None, + title: str = None, + url: str = None, + tab_type: str = 'page', + dual_mode: bool = False, + as_id: bool = False) -> Union[ChromiumTab, str]: ... + + def _get_tabs(self, + title: str = None, + url: str = None, + tab_type: str = 'page', + dual_mode: bool = False, + as_id: bool = False) -> List[ChromiumTab, str]: ... def activate_tab(self, tab_id: str) -> None: ... - def get_window_bounds(self, tab_id: str = None) -> dict: ... + def _new_tab(self, + obj, + url: str = None, + new_window: bool = False, + background: bool = False, + new_context: bool = False) -> Union[ChromiumTab, MixTab]: ... - def new_tab(self, new_window: bool = False, background: bool = False, new_context: bool = False) -> str: ... + def new_tab(self, + url: str = None, + new_window: bool = False, + background: bool = False, + new_context: bool = False) -> ChromiumTab: ... + + def new_mix_tab(self, + url: str = None, + new_window: bool = False, + background: bool = False, + new_context: bool = False) -> MixTab: ... def reconnect(self) -> None: ... - def connect_to_page(self) -> None: ... - def _onTargetCreated(self, **kwargs) -> None: ... def _onTargetDestroyed(self, **kwargs) -> None: ... diff --git a/DrissionPage/_configs/chromium_options.py b/DrissionPage/_configs/chromium_options.py index 86e64d2..acb7295 100644 --- a/DrissionPage/_configs/chromium_options.py +++ b/DrissionPage/_configs/chromium_options.py @@ -36,7 +36,7 @@ class ChromiumOptions(object): om = OptionsManager(ini_path) options = om.chromium_options - self._download_path = om.paths.get('download_path', None) or None + self._download_path = om.paths.get('download_path', '.') or '.' self._tmp_path = om.paths.get('tmp_path', None) or None self._arguments = options.get('arguments', []) self._browser_path = options.get('browser_path', '') @@ -282,14 +282,13 @@ class ChromiumOptions(object): self._prefs = {} return self - def set_timeouts(self, base=None, page_load=None, script=None, implicit=None): + def set_timeouts(self, base=None, page_load=None, script=None): """设置超时时间，单位为秒 :param base: 默认超时时间 :param page_load: 页面加载超时时间 :param script: 脚本运行超时时间 :return: 当前对象 """ - base = base if base is not None else implicit if base is not None: self._timeouts['base'] = base if page_load is not None: @@ -450,7 +449,7 @@ class ChromiumOptions(object): :param path: 下载路径 :return: 当前对象 """ - self._download_path = str(path) + self._download_path = '.' if path is None else str(path) return self def set_tmp_path(self, path): diff --git a/DrissionPage/_configs/session_options.py b/DrissionPage/_configs/session_options.py index a533e4d..25ea8cd 100644 --- a/DrissionPage/_configs/session_options.py +++ b/DrissionPage/_configs/session_options.py @@ -24,7 +24,7 @@ class SessionOptions(object): :param ini_path: ini文件路径 """ self.ini_path = None - self._download_path = None + self._download_path = '.' self._timeout = 10 self._del_set = set() # 记录要从ini文件删除的参数 @@ -83,7 +83,7 @@ class SessionOptions(object): self.set_proxies(om.proxies.get('http', None), om.proxies.get('https', None)) self._timeout = om.timeouts.get('base', 10) - self._download_path = om.paths.get('download_path', None) or None + self._download_path = om.paths.get('download_path', '.') or '.' others = om.others self._retry_times = others.get('retry_times', 3) @@ -100,7 +100,7 @@ class SessionOptions(object): :param path: 下载路径 :return: 返回当前对象 """ - self._download_path = str(path) + self._download_path = '.' if path is None else str(path) return self @property @@ -419,7 +419,7 @@ class SessionOptions(object): return session_options_to_dict(self) def make_session(self): - """根据内在的配置生成Session对象，ua从对象中分离""" + """根据内在的配置生成Session对象，headers从对象中分离""" s = Session() h = CaseInsensitiveDict(self.headers) if self.headers else CaseInsensitiveDict() diff --git a/DrissionPage/_elements/chromium_element.pyi b/DrissionPage/_elements/chromium_element.pyi index 1be5553..a629c85 100644 --- a/DrissionPage/_elements/chromium_element.pyi +++ b/DrissionPage/_elements/chromium_element.pyi @@ -183,7 +183,7 @@ class ChromiumElement(DrissionElement): def select(self) -> SelectElement: ... @property - def value(self) -> None: ... + def value(self) -> str: ... def check(self, uncheck: bool = False, by_js: bool = False) -> None: ... diff --git a/DrissionPage/_pages/chromium_base.py b/DrissionPage/_pages/chromium_base.py index 96a1173..08ec207 100644 --- a/DrissionPage/_pages/chromium_base.py +++ b/DrissionPage/_pages/chromium_base.py @@ -38,11 +38,10 @@ __ERROR__ = 'error' class ChromiumBase(BasePage): """标签页、frame、页面基类""" - def __init__(self, address, tab_id=None, timeout=None): + def __init__(self, address, tab_id=None): """ :param address: 浏览器 ip:port :param tab_id: 要控制的标签页id，不指定默认为激活的 - :param timeout: 超时时间（秒） """ super().__init__() self._is_loading = None @@ -71,8 +70,6 @@ class ChromiumBase(BasePage): self._d_set_start_options(address) self._d_set_runtime_settings() self._connect_browser(tab_id) - if timeout is not None: - self.timeout = timeout def _d_set_start_options(self, address): """设置浏览器启动属性 @@ -93,7 +90,7 @@ class ChromiumBase(BasePage): self._is_reading = False if not tab_id: - tabs = self.browser.driver.get(f'http://{self.address}/json').json() + tabs = self.browser._driver.get(f'http://{self.address}/json').json() tabs = [(i['id'], i['url']) for i in tabs if i['type'] in ('page', 'webview') and not i['url'].startswith('devtools://')] dialog = None @@ -861,7 +858,8 @@ class ChromiumBase(BasePage): def disconnect(self): """断开与页面的连接，不关闭页面""" if self._driver: - self.browser.stop_driver(self._driver) + self._driver.stop() + self.browser._all_drivers.get(self._driver.id, set()).discard(self._driver) def reconnect(self, wait=0): """断开与页面原来的页面，重新建立连接 @@ -1119,15 +1117,12 @@ class ChromiumBase(BasePage): class Timeout(object): """用于保存d模式timeout信息的类""" - def __init__(self, page, base=None, page_load=None, script=None, implicit=None): + def __init__(self, base=None, page_load=None, script=None): """ - :param page: ChromiumBase页面 :param base: 默认超时时间 :param page_load: 页面加载超时时间 :param script: js超时时间 """ - self._page = page - base = base if base is not None else implicit self.base = 10 if base is None else base self.page_load = 30 if page_load is None else page_load self.script = 30 if script is None else script diff --git a/DrissionPage/_pages/chromium_base.pyi b/DrissionPage/_pages/chromium_base.pyi index 17b14bb..beaadad 100644 --- a/DrissionPage/_pages/chromium_base.pyi +++ b/DrissionPage/_pages/chromium_base.pyi @@ -32,8 +32,7 @@ PIC_TYPE = Literal['jpg', 'jpeg', 'png', 'webp', True] class ChromiumBase(BasePage): def __init__(self, address: Union[str, int], - tab_id: str = None, - timeout: float = None): + tab_id: str = None): self._browser: Browser = ... self._page: ChromiumPage = ... self.tab: Union[ChromiumPage, ChromiumTab] = ... @@ -267,8 +266,7 @@ class ChromiumBase(BasePage): class Timeout(object): - def __init__(self, page: ChromiumBase, base=None, page_load=None, script=None): - self._page: ChromiumBase = ... + def __init__(self, base=None, page_load=None, script=None): self.base: float = ... self.page_load: float = ... self.script: float = ... diff --git a/DrissionPage/_pages/chromium_frame.py b/DrissionPage/_pages/chromium_frame.py index f3e1ab5..aa537c8 100644 --- a/DrissionPage/_pages/chromium_frame.py +++ b/DrissionPage/_pages/chromium_frame.py @@ -27,14 +27,17 @@ class ChromiumFrame(ChromiumBase): :param ele: frame所在元素 :param info: frame所在元素信息 """ - if owner._type in ('ChromiumPage', 'WebPage'): - self._page = self._target_page = self.tab = owner - self._browser = owner.browser - else: # Tab、Frame - self._page = owner.page - self._browser = self._page.browser - self._target_page = owner - self.tab = owner.tab if owner._type == 'ChromiumFrame' else owner + self.tab = owner.tab + self._browser = owner.browser + self._target_page = owner + # if owner._type in ('ChromiumPage', 'WebPage'): + # self._target_page = self.tab = owner + # self._browser = owner.browser + # else: # ChromiumTab、Frame + # # self._page = owner.page + # self._browser = self._page.browser + # self._target_page = owner + # self.tab = owner.tab if owner._type == 'ChromiumFrame' else owner self.address = owner.address self._tab_id = owner.tab_id @@ -250,10 +253,10 @@ class ChromiumFrame(ChromiumBase): """返回cdp中的node id""" return self.frame_ele._node_id - @property - def page(self): - """返回所属Page对象""" - return self._page + # @property + # def page(self): + # """返回所属Page对象""" + # return self._page @property def owner(self): diff --git a/DrissionPage/_pages/chromium_frame.pyi b/DrissionPage/_pages/chromium_frame.pyi index b0c1a83..9e55d22 100644 --- a/DrissionPage/_pages/chromium_frame.pyi +++ b/DrissionPage/_pages/chromium_frame.pyi @@ -9,7 +9,7 @@ from pathlib import Path from typing import Union, Tuple, List, Any, Optional from .chromium_base import ChromiumBase -from .chromium_page import ChromiumPage +# from .chromium_page import ChromiumPage from .chromium_tab import ChromiumTab from .web_page import WebPage from .._elements.chromium_element import ChromiumElement @@ -25,13 +25,12 @@ from .._units.waiter import FrameWaiter class ChromiumFrame(ChromiumBase): def __init__(self, - owner: Union[ChromiumPage, WebPage, ChromiumTab, ChromiumFrame], + owner: Union[ChromiumTab, ChromiumFrame], ele: ChromiumElement, info: dict = None): self._target_page: ChromiumBase = ... - self._page: ChromiumPage = ... - self.tab: Union[ChromiumPage, ChromiumTab] = ... - self._tab_id: str = ... + self.tab: ChromiumTab = ... + # self._tab_id: str = ... self._set: ChromiumFrameSetter = ... self._frame_ele: ChromiumElement = ... self._backend_id: int = ... @@ -66,8 +65,8 @@ class ChromiumFrame(ChromiumBase): def _onInspectorDetached(self, **kwargs): ... - @property - def page(self) -> Union[ChromiumPage, WebPage]: ... + # @property + # def page(self) -> Union[ChromiumPage, WebPage]: ... @property def owner(self) -> ChromiumBase: ... diff --git a/DrissionPage/_pages/chromium_page.py b/DrissionPage/_pages/chromium_page.py index 7bd7277..b4fea5d 100644 --- a/DrissionPage/_pages/chromium_page.py +++ b/DrissionPage/_pages/chromium_page.py @@ -34,19 +34,16 @@ class ChromiumPage(ChromiumBase): :param tab_id: 要控制的标签页id，不指定默认为激活的 :param timeout: 超时时间（秒） """ - opt = handle_options(addr_or_opts) - is_exist, browser_id = run_browser(opt) - if browser_id in cls._PAGES: - r = cls._PAGES[browser_id] + browser = Browser(addr_or_opts=addr_or_opts) + if browser.id in cls._PAGES: + r = cls._PAGES[browser.id] while not hasattr(r, '_frame_id'): sleep(.1) return r + r = object.__new__(cls) - r._chromium_options = opt - r._is_exist = is_exist - r._browser_id = browser_id - r.address = opt.address - cls._PAGES[browser_id] = r + r._browser = browser + cls._PAGES[browser.id] = r return r def __init__(self, addr_or_opts=None, tab_id=None, timeout=None): @@ -59,12 +56,9 @@ class ChromiumPage(ChromiumBase): return self._created = True - self._page = self self.tab = self - self._run_browser() - super().__init__(self.address, tab_id) + super().__init__(self.browser.address, tab_id) self._type = 'ChromiumPage' - self._lock = Lock() self.set.timeouts(base=timeout) self._page_init() @@ -86,7 +80,7 @@ class ChromiumPage(ChromiumBase): def _d_set_runtime_settings(self): """设置运行时用到的属性""" - self._timeouts = Timeout(self, page_load=self._chromium_options.timeouts['page_load'], + self._timeouts = Timeout(page_load=self._chromium_options.timeouts['page_load'], script=self._chromium_options.timeouts['script'], base=self._chromium_options.timeouts['base']) if self._chromium_options.timeouts['base'] is not None: @@ -196,7 +190,7 @@ class ChromiumPage(ChromiumBase): return id_or_num with self._lock: - return ChromiumTab(self, id_or_num) + return ChromiumTab(self.browser, id_or_num) def get_tabs(self, title=None, url=None, tab_type='page', as_id=False): """查找符合条件的tab，返回它们组成的列表 @@ -209,7 +203,7 @@ class ChromiumPage(ChromiumBase): if as_id: return [tab['id'] for tab in self._browser.find_tabs(title, url, tab_type)] with self._lock: - return [ChromiumTab(self, tab['id']) for tab in self._browser.find_tabs(title, url, tab_type)] + return [ChromiumTab(self.browser, tab['id']) for tab in self._browser.find_tabs(title, url, tab_type)] def new_tab(self, url=None, new_window=False, background=False, new_context=False): """新建一个标签页 @@ -219,7 +213,7 @@ class ChromiumPage(ChromiumBase): :param new_context: 是否创建新的上下文 :return: 新标签页对象 """ - tab = ChromiumTab(self, tab_id=self.browser.new_tab(new_window, background, new_context)) + tab = ChromiumTab(self.browser, tab_id=self.browser.new_tab(new_window, background, new_context)) if url: tab.get(url) return tab diff --git a/DrissionPage/_pages/chromium_page.pyi b/DrissionPage/_pages/chromium_page.pyi index 7117b9e..2aab4c1 100644 --- a/DrissionPage/_pages/chromium_page.pyi +++ b/DrissionPage/_pages/chromium_page.pyi @@ -20,6 +20,10 @@ from .._units.waiter import PageWaiter class ChromiumPage(ChromiumBase): _PAGES: dict = ... + tab: ChromiumPage = ... + _browser: Browser = ... + _rect: Optional[TabRect] = ... + _is_exist: bool = ... def __new__(cls, addr_or_opts: Union[str, int, ChromiumOptions] = None, @@ -29,15 +33,7 @@ class ChromiumPage(ChromiumBase): def __init__(self, addr_or_opts: Union[str, int, ChromiumOptions] = None, tab_id: str = None, - timeout: float = None): - self.tab: ChromiumPage = ... - self._chromium_options: ChromiumOptions = ... - self._browser: Browser = ... - self._browser_id: str = ... - self._rect: Optional[TabRect] = ... - self._is_exist: bool = ... - self._lock: Lock = ... - self._browser_version: str = ... + timeout: float = None):... def _handle_options(self, addr_or_opts: Union[str, ChromiumOptions]) -> str: ... diff --git a/DrissionPage/_pages/chromium_tab.py b/DrissionPage/_pages/chromium_tab.py index d965a5d..d4084bb 100644 --- a/DrissionPage/_pages/chromium_tab.py +++ b/DrissionPage/_pages/chromium_tab.py @@ -22,10 +22,10 @@ class ChromiumTab(ChromiumBase): """实现浏览器标签页的类""" _TABS = {} - def __new__(cls, page, tab_id): + def __new__(cls, browser, tab_id): """ - :param page: ChromiumPage对象 - :param tab_id: 要控制的标签页id + :param browser: Browser对象 + :param tab_id: 标签页id """ if Settings.singleton_tab_obj and tab_id in cls._TABS: r = cls._TABS[tab_id] @@ -36,38 +36,32 @@ class ChromiumTab(ChromiumBase): cls._TABS[tab_id] = r return r - def __init__(self, page, tab_id): + def __init__(self, browser, tab_id): """ - :param page: ChromiumPage对象 - :param tab_id: 要控制的标签页id + :param browser: Browser对象 + :param tab_id: 标签页id """ if Settings.singleton_tab_obj and hasattr(self, '_created'): return self._created = True - self._page = page self.tab = self - self._browser = page.browser - super().__init__(page.address, tab_id, page.timeout) + self._browser = browser + super().__init__(browser.address, tab_id, browser.timeout) self._rect = None self._type = 'ChromiumTab' def _d_set_runtime_settings(self): """重写设置浏览器运行参数方法""" - self._timeouts = copy(self.page.timeouts) - self.retry_times = self.page.retry_times - self.retry_interval = self.page.retry_interval - self._load_mode = self.page._load_mode - self._download_path = self.page.download_path + self._timeouts = copy(self.browser.timeouts) + self.retry_times = self.browser.retry_times + self.retry_interval = self.browser.retry_interval + self._load_mode = self.browser._load_mode + self._download_path = self.browser.download_path def close(self): """关闭当前标签页""" - self.page.close_tabs(self.tab_id) - - @property - def page(self): - """返回总体page对象""" - return self._page + self.browser.close_tabs(self.tab_id) @property def set(self): @@ -100,11 +94,11 @@ class ChromiumTab(ChromiumBase): ChromiumTab._TABS.pop(self.tab_id, None) -class WebPageTab(SessionPage, ChromiumTab, BasePage): - def __init__(self, page, tab_id): +class MixTab(SessionPage, ChromiumTab, BasePage): + def __init__(self, browser, tab_id): """ - :param page: WebPage对象 - :param tab_id: 要控制的标签页id + :param browser: Browser对象 + :param tab_id: 标签页id """ if Settings.singleton_tab_obj and hasattr(self, '_created'): return @@ -112,10 +106,9 @@ class WebPageTab(SessionPage, ChromiumTab, BasePage): self._mode = 'd' self._has_driver = True self._has_session = True - super().__init__(session_or_options=SessionOptions(read_file=False).from_session(copy(page.session), - page._headers)) - super(SessionPage, self).__init__(page=page, tab_id=tab_id) - self._type = 'WebPageTab' + super().__init__(session_or_options=browser._session_options if browser._session_options else SessionOptions()) + super(SessionPage, self).__init__(browser=browser, tab_id=tab_id) + self._type = 'MixTab' def __call__(self, locator, index=1, timeout=None): """在内部查找元素 @@ -318,7 +311,9 @@ class WebPageTab(SessionPage, ChromiumTab, BasePage): # s模式转d模式 if self._mode == 'd': if self._driver is None: - self._connect_browser(self.page._chromium_options) + tabs = self.browser.tab_ids + tid = self.tab_id if self.tab_id in tabs else tabs[0] + self._connect_browser(tid) self._url = None if not self._has_driver else super(SessionPage, self).url self._has_driver = True @@ -378,7 +373,7 @@ class WebPageTab(SessionPage, ChromiumTab, BasePage): def close(self): """关闭当前标签页""" - self.page.close_tabs(self.tab_id) + self.browser.close_tabs(self.tab_id) self._session.close() if self._response is not None: self._response.close() @@ -398,4 +393,4 @@ class WebPageTab(SessionPage, ChromiumTab, BasePage): return super(SessionPage, self)._find_elements(locator, timeout=timeout, index=index, relative=relative) def __repr__(self): - return f'' + return f'' diff --git a/DrissionPage/_pages/chromium_tab.pyi b/DrissionPage/_pages/chromium_tab.pyi index 2161070..738b712 100644 --- a/DrissionPage/_pages/chromium_tab.pyi +++ b/DrissionPage/_pages/chromium_tab.pyi @@ -12,9 +12,7 @@ from requests import Session, Response from .chromium_base import ChromiumBase from .chromium_frame import ChromiumFrame -from .chromium_page import ChromiumPage from .session_page import SessionPage -from .web_page import WebPage from .._base.browser import Browser from .._elements.chromium_element import ChromiumElement from .._elements.session_element import SessionElement @@ -27,10 +25,9 @@ from .._units.waiter import TabWaiter class ChromiumTab(ChromiumBase): _TABS: dict = ... - def __new__(cls, page: ChromiumPage, tab_id: str): ... + def __new__(cls, browser: Browser, tab_id: str): ... - def __init__(self, page: ChromiumPage, tab_id: str): - self._page: ChromiumPage = ... + def __init__(self, browser: Browser, tab_id: str): self._browser: Browser = ... self._rect: Optional[TabRect] = ... @@ -38,9 +35,6 @@ class ChromiumTab(ChromiumBase): def close(self) -> None: ... - @property - def page(self) -> ChromiumPage: ... - @property def set(self) -> TabSetter: ... @@ -69,22 +63,19 @@ class ChromiumTab(ChromiumBase): generateDocumentOutline: bool = ...) -> Union[bytes, str]: ... -class WebPageTab(SessionPage, ChromiumTab): - def __init__(self, page: WebPage, tab_id: str): - self._page: WebPage = ... - self._browser: Browser = ... - self._mode: str = ... - self._has_driver = ... - self._has_session = ... +class MixTab(SessionPage, ChromiumTab): + _browser: Browser = ... + _mode: str = ... + _has_driver: bool = ... + _has_session: bool = ... + + def __init__(self, browser: Browser, tab_id: str): ... def __call__(self, locator: Union[Tuple[str, str], str, ChromiumElement, SessionElement], index: int = 1, timeout: float = None) -> Union[ChromiumElement, SessionElement]: ... - @property - def page(self) -> WebPage: ... - @property def url(self) -> Union[str, None]: ... @@ -121,9 +112,6 @@ class WebPageTab(SessionPage, ChromiumTab): @property def timeout(self) -> float: ... - @timeout.setter - def timeout(self, second: float) -> None: ... - def get(self, url: str, show_errmsg: bool = False, diff --git a/DrissionPage/_pages/web_page.py b/DrissionPage/_pages/web_page.py index ea9e83d..fbcd2fb 100644 --- a/DrissionPage/_pages/web_page.py +++ b/DrissionPage/_pages/web_page.py @@ -6,7 +6,7 @@ @License : BSD 3-Clause. """ from .chromium_page import ChromiumPage -from .chromium_tab import WebPageTab +from .chromium_tab import MixTab from .session_page import SessionPage from .._base.base import BasePage from .._configs.chromium_options import ChromiumOptions @@ -150,14 +150,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage): """返回通用timeout设置""" return self.timeouts.base - @timeout.setter - def timeout(self, second): - """设置通用超时时间 - :param second: 秒数 - :return: None - """ - self.set.timeouts(base=second) - def get(self, url, show_errmsg=False, retry=None, interval=None, timeout=None, **kwargs): """跳转到一个url :param url: 目标url @@ -321,7 +313,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage): id_or_num = id_or_num elif isinstance(id_or_num, int): id_or_num = self.tab_ids[id_or_num - 1 if id_or_num > 0 else id_or_num] - elif isinstance(id_or_num, WebPageTab): + elif isinstance(id_or_num, MixTab): return id_or_num.tab_id if as_id else id_or_num elif title == url == tab_type is None: @@ -338,7 +330,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage): return id_or_num with self._lock: - return WebPageTab(self, id_or_num) + return MixTab(self, id_or_num) def get_tabs(self, title=None, url=None, tab_type='page', as_id=False): """查找符合条件的tab，返回它们组成的列表 @@ -351,7 +343,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage): if as_id: return [tab['id'] for tab in self._browser.find_tabs(title, url, tab_type)] with self._lock: - return [WebPageTab(self, tab['id']) for tab in self._browser.find_tabs(title, url, tab_type)] + return [MixTab(self, tab['id']) for tab in self._browser.find_tabs(title, url, tab_type)] def new_tab(self, url=None, new_window=False, background=False, new_context=False): """新建一个标签页 @@ -361,7 +353,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage): :param new_context: 是否创建新的上下文 :return: 新标签页对象 """ - tab = WebPageTab(self, tab_id=self.browser.new_tab(new_window, background, new_context)) + tab = MixTab(self, tab_id=self.browser.new_tab(new_window, background, new_context)) if url: tab.get(url) return tab diff --git a/DrissionPage/_pages/web_page.pyi b/DrissionPage/_pages/web_page.pyi index e1046d8..658c976 100644 --- a/DrissionPage/_pages/web_page.pyi +++ b/DrissionPage/_pages/web_page.pyi @@ -11,7 +11,7 @@ from requests import Session, Response from .chromium_frame import ChromiumFrame from .chromium_page import ChromiumPage -from .chromium_tab import WebPageTab +from .chromium_tab import MixTab from .session_page import SessionPage from .._base.base import BasePage from .._base.driver import Driver @@ -129,23 +129,23 @@ class WebPage(SessionPage, ChromiumPage, BasePage): all_info: bool = False) -> Union[dict, list]: ... def get_tab(self, - id_or_num: Union[str, WebPageTab, int] = None, + id_or_num: Union[str, MixTab, int] = None, title: str = None, url: str = None, tab_type: Union[str, list, tuple] = 'page', - as_id: bool = False) -> Union[WebPageTab, str, None]: ... + as_id: bool = False) -> Union[MixTab, str, None]: ... def get_tabs(self, title: str = None, url: str = None, tab_type: Union[str, list, tuple] = 'page', - as_id: bool = False) -> Union[List[WebPageTab], List[str]]: ... + as_id: bool = False) -> Union[List[MixTab], List[str]]: ... def new_tab(self, url: str = None, new_window: bool = False, background: bool = False, - new_context: bool = False) -> WebPageTab: ... + new_context: bool = False) -> MixTab: ... def close_driver(self) -> None: ... @@ -175,7 +175,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage): cert: Any | None = ...) -> Union[bool, Response]: ... @property - def latest_tab(self) -> Union[WebPageTab, WebPage]: ... + def latest_tab(self) -> Union[MixTab, WebPage]: ... @property def set(self) -> WebPageSetter: ... diff --git a/DrissionPage/_units/clicker.py b/DrissionPage/_units/clicker.py index 6006b4c..eb97ba9 100644 --- a/DrissionPage/_units/clicker.py +++ b/DrissionPage/_units/clicker.py @@ -120,12 +120,13 @@ class Clicker(object): """ self._ele.owner.scroll.to_see(self._ele) x, y = self._ele.rect.viewport_click_point + curr_tid = self._ele.tab.browser.tab_ids[0] self._click(x, y, 'middle') if get_tab: - tid = self._ele.page.wait.new_tab() + tid = self._ele.tab.browser.wait.new_tab(curr_tab=curr_tid) if not tid: raise RuntimeError('没有出现新标签页。') - return self._ele.page.get_tab(tid) + return self._ele.tab.browser.get_tab(tid) def at(self, offset_x=None, offset_y=None, button='left', count=1): """带偏移量点击本元素，相对于左上角坐标。不传入x或y值时点击元素中间点 @@ -162,16 +163,15 @@ class Clicker(object): """ if save_path: self._ele.owner.tab.set.download_path(save_path) - elif not self._ele.page._browser._dl_mgr._running: - self._ele.page.set.download_path('.') + elif not self._ele.tab._browser._dl_mgr._running: + self._ele.owner.tab._browser.set.download_path('.') + obj = self._ele.tab._browser if new_tab else self._ele.owner.tab if rename or suffix: - self._ele.owner.tab.set.download_file_name(rename, suffix) - - tab = self._ele.page if new_tab else self._ele.owner + obj.set.download_file_name(rename, suffix) self.left(by_js=by_js) - return tab.wait.download_begin(timeout=timeout) + return obj.wait.download_begin(timeout=timeout) def to_upload(self, file_paths, by_js=False): """触发上传文件选择框并自动填入指定路径 @@ -183,16 +183,18 @@ class Clicker(object): self.left(by_js=by_js) self._ele.owner.wait.upload_paths_inputted() - def for_new_tab(self, by_js=False): + def for_new_tab(self, by_js=False, timeout=3): """点击后等待新tab出现并返回其对象 :param by_js: 是否使用js点击，逻辑与click()一致 + :param timeout: 等待超时时间 :return: 新标签页对象，如果没有等到新标签页出现则抛出异常 """ + curr_tid = self._ele.tab.browser.tab_ids[0] self.left(by_js=by_js) - tid = self._ele.page.wait.new_tab() + tid = self._ele.tab.browser.wait.new_tab(timeout=timeout, curr_tab=curr_tid) if not tid: raise RuntimeError('没有出现新标签页。') - return self._ele.page.get_tab(tid) + return self._ele.tab.browser.get_tab(tid) def _click(self, client_x, client_y, button='left', count=1): """实施点击 diff --git a/DrissionPage/_units/clicker.pyi b/DrissionPage/_units/clicker.pyi index 15fc212..5031e27 100644 --- a/DrissionPage/_units/clicker.pyi +++ b/DrissionPage/_units/clicker.pyi @@ -10,7 +10,7 @@ from typing import Union from .downloader import DownloadMission from .._elements.chromium_element import ChromiumElement -from .._pages.chromium_tab import WebPageTab, ChromiumTab +from .._pages.chromium_tab import MixTab, ChromiumTab class Clicker(object): @@ -23,7 +23,7 @@ class Clicker(object): def right(self) -> None: ... - def middle(self, get_tab: bool = True) -> Union[ChromiumTab, WebPageTab, None]: ... + def middle(self, get_tab: bool = True) -> Union[ChromiumTab, MixTab, None]: ... def at(self, offset_x: float = None, @@ -43,6 +43,6 @@ class Clicker(object): def to_upload(self, file_paths: Union[str, Path, list, tuple], by_js: bool = False) -> None: ... - def for_new_tab(self, by_js: bool = False) -> Union[ChromiumTab, WebPageTab]: ... + def for_new_tab(self, by_js: bool = False, timeout: float = 3) -> Union[ChromiumTab, MixTab]: ... def _click(self, client_x: float, client_y: float, button: str = 'left', count: int = 1) -> None: ... diff --git a/DrissionPage/_units/cookies_setter.pyi b/DrissionPage/_units/cookies_setter.pyi index cbe37de..c929448 100644 --- a/DrissionPage/_units/cookies_setter.pyi +++ b/DrissionPage/_units/cookies_setter.pyi @@ -9,7 +9,7 @@ from http.cookiejar import Cookie, CookieJar from typing import Union from .._pages.chromium_base import ChromiumBase -from .._pages.chromium_tab import WebPageTab +from .._pages.chromium_tab import MixTab from .._pages.session_page import SessionPage from .._pages.web_page import WebPage @@ -39,7 +39,7 @@ class SessionCookiesSetter(object): class WebPageCookiesSetter(CookiesSetter, SessionCookiesSetter): - _owner: Union[WebPage, WebPageTab] + _owner: Union[WebPage, MixTab] def __init__(self, page: SessionPage): ... diff --git a/DrissionPage/_units/downloader.py b/DrissionPage/_units/downloader.py index ec41cc4..0318266 100644 --- a/DrissionPage/_units/downloader.py +++ b/DrissionPage/_units/downloader.py @@ -20,21 +20,24 @@ class DownloadManager(object): :param browser: Browser对象 """ self._browser = browser - self._page = browser.page - self._when_download_file_exists = 'rename' - self._save_path = None + # self._page = browser.page + # self._when_download_file_exists = 'rename' + # self._save_path = None + + t = TabDownloadSettings('browser') + t.path = self._browser.download_path + t.rename = None + t.suffix = None + t.when_file_exists = 'rename' - t = TabDownloadSettings(self._page.tab_id) - t.path = self._page.download_path self._missions = {} # {guid: DownloadMission} self._tab_missions = {} # {tab_id: DownloadMission} self._flags = {} # {tab_id: [bool, DownloadMission]} - if self._page.download_path: - self.set_path(self._page, self._page.download_path) + # if self._page.download_path: + # self.set_path(self._page, self._page.download_path) - else: - self._running = False + self._running = False @property def missions(self): @@ -47,13 +50,13 @@ class DownloadManager(object): :param path: 下载路径（绝对路径str） :return: None """ - TabDownloadSettings(tab.tab_id).path = path - if tab is self._page or not self._running: - self._browser.driver.set_callback('Browser.downloadProgress', self._onDownloadProgress) - self._browser.driver.set_callback('Browser.downloadWillBegin', self._onDownloadWillBegin) - r = self._browser.run_cdp('Browser.setDownloadBehavior', downloadPath=path, + tid = tab if isinstance(tab, str) else tab.tab_id + TabDownloadSettings(tid).path = path + if not self._running or tid == 'browser': + self._browser._driver.set_callback('Browser.downloadProgress', self._onDownloadProgress) + self._browser._driver.set_callback('Browser.downloadWillBegin', self._onDownloadWillBegin) + r = self._browser.run_cdp('Browser.setDownloadBehavior', downloadPath=self._browser._download_path, behavior='allowAndName', eventsEnabled=True) - self._save_path = path if 'error' in r: print('浏览器版本太低无法使用下载管理功能。') self._running = True @@ -149,10 +152,11 @@ class DownloadManager(object): def _onDownloadWillBegin(self, **kwargs): """用于获取弹出新标签页触发的下载任务""" + # print(kwargs) guid = kwargs['guid'] - tab_id = self._browser._frames.get(kwargs['frameId'], self._page.tab_id) + tab_id = self._browser._frames.get(kwargs['frameId'], 'browser') - settings = TabDownloadSettings(tab_id if tab_id in TabDownloadSettings.TABS else self._page.tab_id) + settings = TabDownloadSettings(tab_id if tab_id in TabDownloadSettings.TABS else 'browser') if settings.rename: if settings.suffix is not None: name = f'{settings.rename}.{settings.suffix}' if settings.suffix else settings.rename @@ -184,7 +188,7 @@ class DownloadManager(object): elif settings.when_file_exists == 'overwrite': goal_path.unlink() - m = DownloadMission(self, tab_id, guid, settings.path, name, kwargs['url'], self._save_path) + m = DownloadMission(self, tab_id, guid, settings.path, name, kwargs['url'], self._browser.download_path) self._missions[guid] = m if self.get_flag(tab_id) is False: # 取消该任务 @@ -199,6 +203,7 @@ class DownloadManager(object): def _onDownloadProgress(self, **kwargs): """下载状态变化时执行""" + # print(kwargs) if kwargs['guid'] in self._missions: mission = self._missions[kwargs['guid']] if kwargs['state'] == 'inProgress': diff --git a/DrissionPage/_units/downloader.pyi b/DrissionPage/_units/downloader.pyi index eadcc44..10fd5e3 100644 --- a/DrissionPage/_units/downloader.pyi +++ b/DrissionPage/_units/downloader.pyi @@ -9,28 +9,28 @@ from typing import Dict, Optional, Union, Literal from .._base.browser import Browser from .._pages.chromium_base import ChromiumBase -from .._pages.chromium_page import ChromiumPage - +# from .._pages.chromium_page import ChromiumPage +FILE_EXISTS = Literal['skip', 'rename', 'overwrite', 's', 'r', 'o'] class DownloadManager(object): _browser: Browser = ... - _page: ChromiumPage = ... + # _page: ChromiumPage = ... _missions: Dict[str, DownloadMission] = ... _tab_missions: dict = ... _flags: dict = ... _running: bool = ... - _save_path: Optional[str] = ... + # _save_path: Optional[str] = ... def __init__(self, browser: Browser): ... @property def missions(self) -> Dict[str, DownloadMission]: ... - def set_path(self, tab: ChromiumBase, path: str) -> None: ... + def set_path(self, tab: Union[str,ChromiumBase], path: str) -> None: ... def set_rename(self, tab_id: str, rename: str = None, suffix: str = None) -> None: ... - def set_file_exists(self, tab_id: str, mode: Literal['skip', 'rename', 'overwrite', 's', 'r', 'o']) -> None: ... + def set_file_exists(self, tab_id: str, mode: FILE_EXISTS) -> None: ... def set_flag(self, tab_id: str, flag: Union[bool, DownloadMission, None]) -> None: ... diff --git a/DrissionPage/_units/rect.py b/DrissionPage/_units/rect.py index b751d4b..b8bbbb8 100644 --- a/DrissionPage/_units/rect.py +++ b/DrissionPage/_units/rect.py @@ -184,7 +184,7 @@ class TabRect(object): def _get_window_rect(self): """获取窗口范围信息""" - return self._owner.browser.get_window_bounds(self._owner.tab_id) + return self._owner.browser._driver.run('Browser.getWindowForTarget', targetId=self._owner.tab_id)['bounds'] class FrameRect(object): diff --git a/DrissionPage/_units/rect.pyi b/DrissionPage/_units/rect.pyi index 4fa4e73..c1df2d5 100644 --- a/DrissionPage/_units/rect.pyi +++ b/DrissionPage/_units/rect.pyi @@ -12,7 +12,7 @@ from .._elements.chromium_element import ChromiumElement from .._pages.chromium_base import ChromiumBase from .._pages.chromium_frame import ChromiumFrame from .._pages.chromium_page import ChromiumPage -from .._pages.chromium_tab import ChromiumTab, WebPageTab +from .._pages.chromium_tab import ChromiumTab, MixTab from .._pages.web_page import WebPage @@ -63,7 +63,7 @@ class ElementRect(object): class TabRect(object): def __init__(self, owner: ChromiumBase): - self._owner: Union[ChromiumPage, ChromiumTab, WebPage, WebPageTab] = ... + self._owner: Union[ChromiumPage, ChromiumTab, WebPage, MixTab] = ... @property def window_state(self) -> str: ... diff --git a/DrissionPage/_units/screencast.py b/DrissionPage/_units/screencast.py index fca2b68..bd12711 100644 --- a/DrissionPage/_units/screencast.py +++ b/DrissionPage/_units/screencast.py @@ -39,9 +39,9 @@ class Screencast(object): raise ValueError('save_path必须设置。') if self._mode in ('frugal_video', 'video'): - if self._owner.browser.page._chromium_options.tmp_path: + if self._owner.browser._chromium_options.tmp_path: self._tmp_path = Path( - self._owner.browser.page._chromium_options.tmp_path) / f'screencast_tmp_{time()}_{randint(0, 100)}' + self._owner.browser._chromium_options.tmp_path) / f'screencast_tmp_{time()}_{randint(0, 100)}' else: self._tmp_path = Path(gettempdir()) / 'DrissionPage' / f'screencast_tmp_{time()}_{randint(0, 100)}' self._tmp_path.mkdir(parents=True, exist_ok=True) diff --git a/DrissionPage/_units/setter.py b/DrissionPage/_units/setter.py index 9f0e443..e6e6695 100644 --- a/DrissionPage/_units/setter.py +++ b/DrissionPage/_units/setter.py @@ -33,8 +33,101 @@ class BasePageSetter(object): self._owner._none_ele_return_value = on_off self._owner._none_ele_value = value + def retry_times(self, times): + """设置连接失败重连次数""" + self._owner.retry_times = times -class ChromiumBaseSetter(BasePageSetter): + def retry_interval(self, interval): + """设置连接失败重连间隔""" + self._owner.retry_interval = interval + + def download_path(self, path): + """设置下载路径 + :param path: 下载路径 + :return: None + """ + if path is None: + path = '.' + self._owner._download_path = str(Path(path).absolute()) + + +class BrowserBaseSetter(BasePageSetter): + + @property + def load_mode(self): + """返回用于设置页面加载策略的对象""" + return LoadMode(self._owner) + + def timeouts(self, base=None, page_load=None, script=None): + """设置超时时间，单位为秒 + :param base: 基本等待时间，除页面加载和脚本超时，其它等待默认使用 + :param page_load: 页面加载超时时间 + :param script: 脚本运行超时时间 + :return: None + """ + if base is not None: + self._owner.timeouts.base = base + self._owner._timeout = base + + if page_load is not None: + self._owner.timeouts.page_load = page_load + + if script is not None: + self._owner.timeouts.script = script + + +class BrowserSetter(BrowserBaseSetter): + + def cookies(self, cookies): + pass # todo: 研究Storage.setCookies和Network.setCookies差别 + + def tab_to_front(self, tab_or_id): + """激活标签页使其处于最前面 + :param tab_or_id: 标签页对象或id + :return: None + """ + if not isinstance(tab_or_id, str): # 传入Tab对象 + tab_or_id = tab_or_id.tab_id + self._owner.activate_tab(tab_or_id) + + def auto_handle_alert(self, on_off=True, accept=True): + """设置是否启用自动处理弹窗 + :param on_off: bool表示开或关 + :param accept: bool表示确定还是取消 + :return: None + """ + Settings.auto_handle_alert = accept if on_off else None + + def download_path(self, path): + """设置下载路径 + :param path: 下载路径 + :return: None + """ + super().download_path(path) + self._owner._dl_mgr.set_path('browser', self._owner._download_path) + + def download_file_name(self, name=None, suffix=None): + """设置下一个被下载文件的名称 + :param name: 文件名，可不含后缀，会自动使用远程文件后缀 + :param suffix: 后缀名，显式设置后缀名，不使用远程文件后缀 + :return: None + """ + self._owner._dl_mgr.set_rename('browser', name, suffix) + + def when_download_file_exists(self, mode): + """设置当存在同名文件时的处理方式 + :param mode: 可在 'rename', 'overwrite', 'skip', 'r', 'o', 's'中选择 + :return: None + """ + types = {'rename': 'rename', 'overwrite': 'overwrite', 'skip': 'skip', 'r': 'rename', 'o': 'overwrite', + 's': 'skip'} + mode = types.get(mode, mode) + if mode not in types: + raise ValueError(f'''mode参数只能是 '{"', '".join(types.keys())}' 之一，现在是：{mode}''') + self._owner._dl_mgr.set_file_exists('browser', mode) + + +class ChromiumBaseSetter(BrowserBaseSetter): def __init__(self, owner): """ :param owner: ChromiumBase对象 @@ -42,10 +135,10 @@ class ChromiumBaseSetter(BasePageSetter): super().__init__(owner) self._cookies_setter = None - @property - def load_mode(self): - """返回用于设置页面加载策略的对象""" - return LoadMode(self._owner) + # @property + # def load_mode(self): + # """返回用于设置页面加载策略的对象""" + # return LoadMode(self._owner) @property def scroll(self): @@ -59,31 +152,30 @@ class ChromiumBaseSetter(BasePageSetter): self._cookies_setter = CookiesSetter(self._owner) return self._cookies_setter - def retry_times(self, times): - """设置连接失败重连次数""" - self._owner.retry_times = times - - def retry_interval(self, interval): - """设置连接失败重连间隔""" - self._owner.retry_interval = interval - - def timeouts(self, base=None, page_load=None, script=None, implicit=None): - """设置超时时间，单位为秒 - :param base: 基本等待时间，除页面加载和脚本超时，其它等待默认使用 - :param page_load: 页面加载超时时间 - :param script: 脚本运行超时时间 - :return: None - """ - base = base if base is not None else implicit - if base is not None: - self._owner.timeouts.base = base - self._owner._timeout = base - - if page_load is not None: - self._owner.timeouts.page_load = page_load - - if script is not None: - self._owner.timeouts.script = script + # def retry_times(self, times): + # """设置连接失败重连次数""" + # self._owner.retry_times = times + # + # def retry_interval(self, interval): + # """设置连接失败重连间隔""" + # self._owner.retry_interval = interval + # + # def timeouts(self, base=None, page_load=None, script=None): + # """设置超时时间，单位为秒 + # :param base: 基本等待时间，除页面加载和脚本超时，其它等待默认使用 + # :param page_load: 页面加载超时时间 + # :param script: 脚本运行超时时间 + # :return: None + # """ + # if base is not None: + # self._owner.timeouts.base = base + # self._owner._timeout = base + # + # if page_load is not None: + # self._owner.timeouts.page_load = page_load + # + # if script is not None: + # self._owner.timeouts.script = script def user_agent(self, ua, platform=None): """为当前tab设置user agent，只在当前tab有效 @@ -191,11 +283,10 @@ class TabSetter(ChromiumBaseSetter): :param path: 下载路径 :return: None """ - path = str(Path(path).absolute()) - self._owner._download_path = path - self._owner.browser._dl_mgr.set_path(self._owner, path) + super().download_path(path) + self._owner.browser._dl_mgr.set_path(self._owner, self._owner._download_path) if self._owner._DownloadKit: - self._owner._DownloadKit.set.goal_path(path) + self._owner._DownloadKit.set.goal_path(self._owner._download_path) def download_file_name(self, name=None, suffix=None): """设置下一个被下载文件的名称 @@ -215,7 +306,6 @@ class TabSetter(ChromiumBaseSetter): mode = types.get(mode, mode) if mode not in types: raise ValueError(f'''mode参数只能是 '{"', '".join(types.keys())}' 之一，现在是：{mode}''') - self._owner.browser._dl_mgr.set_file_exists(self._owner.tab_id, mode) def activate(self): @@ -264,23 +354,22 @@ class SessionPageSetter(BasePageSetter): self._cookies_setter = SessionCookiesSetter(self._owner) return self._cookies_setter - def retry_times(self, times): - """设置连接失败时重连次数""" - self._owner.retry_times = times - - def retry_interval(self, interval): - """设置连接失败时重连间隔""" - self._owner.retry_interval = interval + # def retry_times(self, times): + # """设置连接失败时重连次数""" + # self._owner.retry_times = times + # + # def retry_interval(self, interval): + # """设置连接失败时重连间隔""" + # self._owner.retry_interval = interval def download_path(self, path): """设置下载路径 :param path: 下载路径 :return: None """ - path = str(Path(path).absolute()) - self._owner._download_path = path + super().download_path(path) if self._owner._DownloadKit: - self._owner._DownloadKit.set.goal_path(path) + self._owner._DownloadKit.set.goal_path(self._owner._download_path) def timeout(self, second): """设置连接超时时间 @@ -464,7 +553,7 @@ class ChromiumElementSetter(object): """ self._ele = ele - def attr(self, name, value): + def attr(self, name, value=''): """设置元素attribute属性 :param name: 属性名 :param value: 属性值 diff --git a/DrissionPage/_units/setter.pyi b/DrissionPage/_units/setter.pyi index f371bf6..f01dbaf 100644 --- a/DrissionPage/_units/setter.pyi +++ b/DrissionPage/_units/setter.pyi @@ -14,11 +14,12 @@ from requests.auth import HTTPBasicAuth from .cookies_setter import SessionCookiesSetter, CookiesSetter, WebPageCookiesSetter from .scroller import PageScroller from .._base.base import BasePage +from .._base.browser import Browser from .._elements.chromium_element import ChromiumElement from .._pages.chromium_base import ChromiumBase from .._pages.chromium_frame import ChromiumFrame from .._pages.chromium_page import ChromiumPage -from .._pages.chromium_tab import ChromiumTab, WebPageTab +from .._pages.chromium_tab import ChromiumTab, MixTab from .._pages.session_page import SessionPage from .._pages.web_page import WebPage @@ -26,11 +27,41 @@ FILE_EXISTS = Literal['skip', 'rename', 'overwrite', 's', 'r', 'o'] class BasePageSetter(object): - def __init__(self, owner: BasePage): - self._owner: BasePage = ... + def __init__(self, owner: Union[Browser, BasePage]): + self._owner: Union[Browser, BasePage] = ... def NoneElement_value(self, value: Any = None, on_off: bool = True) -> None: ... + def retry_times(self, times: int) -> None: ... + + def retry_interval(self, interval: float) -> None: ... + + def download_path(self, path: Union[str, Path, None]) -> None: ... + + +class BrowserBaseSetter(BasePageSetter): + + @property + def load_mode(self) -> LoadMode: ... + + def timeouts(self, base=None, page_load=None, script=None) -> None: ... + + +class BrowserSetter(BasePageSetter): + _owner: Browser = ... + + def tab_to_front(self, tab_or_id: Union[str, ChromiumTab]) -> None: ... + + def cookies(self, cookies): ... + + def auto_handle_alert(self, on_off: bool = True, accept: bool = True): ... + + def download_path(self, path: Union[Path, str, None]): ... + + def download_file_name(self, name: str = None, suffix: str = None): ... + + def when_download_file_exists(self, mode: FILE_EXISTS): ... + class ChromiumBaseSetter(BasePageSetter): def __init__(self, owner): @@ -70,12 +101,12 @@ class ChromiumBaseSetter(BasePageSetter): class TabSetter(ChromiumBaseSetter): _owner: ChromiumTab = ... - def __init__(self, owner: Union[ChromiumTab, WebPageTab, WebPage, ChromiumPage]): ... + def __init__(self, owner: Union[ChromiumTab, MixTab, WebPage, ChromiumPage]): ... @property def window(self) -> WindowSetter: ... - def download_path(self, path: Union[str, Path]) -> None: ... + def download_path(self, path: Union[str, Path, None]) -> None: ... def download_file_name(self, name: str = None, suffix: str = None) -> None: ... @@ -105,7 +136,7 @@ class SessionPageSetter(BasePageSetter): def retry_interval(self, interval: float) -> None: ... - def download_path(self, path: Union[str, Path]) -> None: ... + def download_path(self, path: Union[str, Path, None]) -> None: ... def timeout(self, second: float) -> None: ... @@ -152,7 +183,7 @@ class WebPageSetter(ChromiumPageSetter): class WebPageTabSetter(TabSetter): - _owner: WebPageTab = ... + _owner: MixTab = ... _session_setter: SessionPageSetter = ... _chromium_setter: ChromiumBaseSetter = ... @@ -168,7 +199,7 @@ class ChromiumElementSetter(object): def __init__(self, ele: ChromiumElement): self._ele: ChromiumElement = ... - def attr(self, name: str, value: str) -> None: ... + def attr(self, name: str, value: str = '') -> None: ... def property(self, name: str, value: str) -> None: ... @@ -186,8 +217,9 @@ class ChromiumFrameSetter(ChromiumBaseSetter): class LoadMode(object): - def __init__(self, owner: ChromiumBase): - self._owner: ChromiumBase = ... + _owner: Union[Browser, ChromiumBase] = ... + + def __init__(self, owner: Union[Browser, ChromiumBase]): ... def __call__(self, value: str) -> None: ... diff --git a/DrissionPage/_units/waiter.py b/DrissionPage/_units/waiter.py index 35bf3b4..9a26ecc 100644 --- a/DrissionPage/_units/waiter.py +++ b/DrissionPage/_units/waiter.py @@ -26,6 +26,84 @@ class OriginWaiter(object): sleep(uniform(second, scope)) +class BrowserWaiter(OriginWaiter): + def __init__(self, owner): + self._owner = owner + + def download_begin(self, timeout=None, cancel_it=False): + """等待浏览器下载开始，可将其拦截 + :param timeout: 超时时间（秒），None使用页面对象超时时间 + :param cancel_it: 是否取消该任务 + :return: 成功返回任务对象，失败返回False + """ + if not self._owner._dl_mgr._running: + raise RuntimeError('此功能需显式设置下载路径（使用set.download_path()方法、配置对象或ini文件均可）。') + self._owner._dl_mgr.set_flag('browser', False if cancel_it else True) + if timeout is None: + timeout = self._owner.timeout + + r = False + end_time = perf_counter() + timeout + while perf_counter() < end_time: + v = self._owner._dl_mgr.get_flag('browser') + if not isinstance(v, bool): + r = v + break + sleep(.005) + + self._owner._dl_mgr.set_flag('browser', None) + return r + + def new_tab(self, timeout=None, curr_tab=None, raise_err=None): + """等待新标签页出现 + :param timeout: 超时时间（秒），为None则使用页面对象timeout属性 + :param curr_tab: 指定当前最新的tab id，为None自动获取 + :param raise_err: 等待失败时是否报错，为None时根据Settings设置 + :return: 等到新标签页返回其id，否则返回False + """ + curr_tid = curr_tab if curr_tab else self._owner.tab_ids[0] + timeout = timeout if timeout is not None else self._owner.timeout + end_time = perf_counter() + timeout + while perf_counter() < end_time: + latest_tid = self._owner.tab_ids[0] + if curr_tid != latest_tid: + return latest_tid + sleep(.01) + + if raise_err is True or Settings.raise_when_wait_failed is True: + raise WaitTimeoutError(f'等待新标签页失败（等待{timeout}秒）。') + else: + return False + + def all_downloads_done(self, timeout=None, cancel_if_timeout=True): + """等待所有浏览器下载任务结束 + :param timeout: 超时时间（秒），为None时无限等待 + :param cancel_if_timeout: 超时时是否取消剩余任务 + :return: 是否等待成功 + """ + if not self._owner._dl_mgr._running: + raise RuntimeError('此功能需显式设置下载路径（使用set.download_path()方法、配置对象或ini文件均可）。') + if not timeout: + while self._owner._dl_mgr._missions: + sleep(.5) + return True + + else: + end_time = perf_counter() + timeout + while perf_counter() < end_time: + if not self._owner._dl_mgr._missions: + return True + sleep(.5) + + if self._owner._dl_mgr._missions: + if cancel_if_timeout: + for m in list(self._owner._dl_mgr._missions.values()): + m.cancel() + return False + else: + return True + + class BaseWaiter(OriginWaiter): def __init__(self, page_or_ele): """ diff --git a/DrissionPage/_units/waiter.pyi b/DrissionPage/_units/waiter.pyi index f05ca94..f4c3966 100644 --- a/DrissionPage/_units/waiter.pyi +++ b/DrissionPage/_units/waiter.pyi @@ -8,6 +8,7 @@ from typing import Union, Tuple, Literal, List from .downloader import DownloadMission +from .._base.browser import Browser from .._elements.chromium_element import ChromiumElement from .._pages.chromium_base import ChromiumBase from .._pages.chromium_frame import ChromiumFrame @@ -18,6 +19,17 @@ class OriginWaiter(object): def __call__(self, second: float, scope: float = None) -> None: ... +class BrowserWaiter(OriginWaiter): + def __init__(self, owner: Browser): + self._owner = owner + + def download_begin(self, timeout: float = None, cancel_it: bool = False) -> DownloadMission: ... + + def new_tab(self, timeout: float = None, curr_tab: str = None, raise_err: bool = None) -> Union[str, bool]: ... + + def all_downloads_done(self, timeout: float = None, cancel_if_timeout: bool = True): ... + + class BaseWaiter(OriginWaiter): def __init__(self, page: ChromiumBase): self._driver: ChromiumBase = ... diff --git a/DrissionPage/items.py b/DrissionPage/items.py index 0715e91..5b680e8 100644 --- a/DrissionPage/items.py +++ b/DrissionPage/items.py @@ -9,7 +9,7 @@ from ._elements.chromium_element import ChromiumElement, ShadowRoot from ._elements.none_element import NoneElement from ._elements.session_element import SessionElement from ._pages.chromium_frame import ChromiumFrame -from ._pages.chromium_tab import ChromiumTab, WebPageTab +from ._pages.chromium_tab import ChromiumTab, MixTab __all__ = ['ChromiumElement', 'ShadowRoot', 'NoneElement', 'SessionElement', 'ChromiumFrame', 'ChromiumTab', - 'WebPageTab'] + 'MixTab']