diff --git a/DrissionPage/__init__.py b/DrissionPage/__init__.py index 66b3008..9660a96 100644 --- a/DrissionPage/__init__.py +++ b/DrissionPage/__init__.py @@ -10,8 +10,8 @@ from ._configs.chromium_options import ChromiumOptions from ._configs.session_options import SessionOptions from ._pages.session_page import SessionPage -# from ._pages.chromium_page import ChromiumPage -# from ._pages.web_page import WebPage +# 即将废弃 +from ._pages.chromium_page import ChromiumPage +from ._pages.web_page import WebPage -__all__ = ['Browser', 'ChromiumOptions', 'SessionOptions', 'SessionPage', '__version__'] -__version__ = '4.0.5.3' +__version__ = '4.1.0.0b0' diff --git a/DrissionPage/__init__.pyi b/DrissionPage/__init__.pyi new file mode 100644 index 0000000..437adc7 --- /dev/null +++ b/DrissionPage/__init__.pyi @@ -0,0 +1,17 @@ +# -*- coding:utf-8 -*- +""" +@Author : g1879 +@Contact : g1879@qq.com +@Copyright: (c) 2024 by g1879, Inc. All Rights Reserved. +@License : BSD 3-Clause. +""" +from ._base.browser import Browser +from ._configs.chromium_options import ChromiumOptions +from ._configs.session_options import SessionOptions +from ._pages.session_page import SessionPage + +from ._pages.chromium_page import ChromiumPage +from ._pages.web_page import WebPage + +__all__ = ['WebPage', 'ChromiumPage', 'Browser', 'ChromiumOptions', 'SessionOptions', 'SessionPage', '__version__'] +__version__: str = ... diff --git a/DrissionPage/_base/browser.py b/DrissionPage/_base/browser.py index 93b85ae..9d232f0 100644 --- a/DrissionPage/_base/browser.py +++ b/DrissionPage/_base/browser.py @@ -126,6 +126,11 @@ class Browser(object): """返回timeouts设置""" return self._timeouts + @property + def load_mode(self): + """返回加载模式""" + return self._load_mode + @property def download_path(self): """返回默认下载路径""" diff --git a/DrissionPage/_base/browser.pyi b/DrissionPage/_base/browser.pyi index fbe2d65..82b5611 100644 --- a/DrissionPage/_base/browser.pyi +++ b/DrissionPage/_base/browser.pyi @@ -65,6 +65,9 @@ class Browser(object): @property def timeouts(self) -> Timeout: ... + @property + def load_mode(self) -> str: ... + @property def download_path(self) -> str: ... diff --git a/DrissionPage/_pages/chromium_base.py b/DrissionPage/_pages/chromium_base.py index 285bbef..27b2e3c 100644 --- a/DrissionPage/_pages/chromium_base.py +++ b/DrissionPage/_pages/chromium_base.py @@ -1158,6 +1158,10 @@ class Timeout(object): def __repr__(self): return str({'base': self.base, 'page_load': self.page_load, 'script': self.script}) + @property + def as_dict(self): + return {'base': self.base, 'page_load': self.page_load, 'script': self.script} + class Alert(object): """用于保存alert信息的类""" diff --git a/DrissionPage/_pages/chromium_base.pyi b/DrissionPage/_pages/chromium_base.pyi index efde024..6653a38 100644 --- a/DrissionPage/_pages/chromium_base.pyi +++ b/DrissionPage/_pages/chromium_base.pyi @@ -34,7 +34,7 @@ class ChromiumBase(BasePage): def __init__(self, browser: Browser, tab_id: str = None): - self._tab:Union[ChromiumTab, MixTab, ChromiumFrame] = ... + self._tab: Union[ChromiumTab, MixTab, ChromiumFrame] = ... self._browser: Browser = ... self._driver: Driver = ... self._frame_id: str = ... @@ -277,6 +277,9 @@ class Timeout(object): self.page_load: float = ... self.script: float = ... + @property + def as_dict(self) -> dict: ... + class Alert(object): diff --git a/DrissionPage/_pages/chromium_page.py b/DrissionPage/_pages/chromium_page.py index 76eec26..a8bfda8 100644 --- a/DrissionPage/_pages/chromium_page.py +++ b/DrissionPage/_pages/chromium_page.py @@ -5,22 +5,13 @@ @Copyright: (c) 2024 by g1879, Inc. All Rights Reserved. @License : BSD 3-Clause. """ -from pathlib import Path -from time import sleep, perf_counter - -from requests import Session +from time import sleep from .._base.browser import Browser -from .._configs.chromium_options import ChromiumOptions -from .._functions.browser import connect_browser -from .._functions.settings import Settings -from .._functions.tools import PortFinder from .._functions.web import save_page -from .._pages.chromium_base import ChromiumBase, Timeout -from .._pages.chromium_tab import ChromiumTab +from .._pages.chromium_base import ChromiumBase from .._units.setter import ChromiumPageSetter from .._units.waiter import PageWaiter -from ..errors import BrowserConnectError class ChromiumPage(ChromiumBase): @@ -56,45 +47,18 @@ class ChromiumPage(ChromiumBase): self._created = True self.tab = self - super().__init__(self.browser.address, tab_id) + super().__init__(self.browser, tab_id) self._type = 'ChromiumPage' self.set.timeouts(base=timeout) - self._page_init() - - def _run_browser(self): - """连接浏览器""" - self._browser = Browser(self._chromium_options.address, self._browser_id, self) - r = self._browser._run_cdp('Browser.getVersion') - self._browser_version = r['product'] - if self._is_exist and self._chromium_options._headless is False and 'headless' in r['userAgent'].lower(): - self._browser.quit(3) - connect_browser(self._chromium_options) - s = Session() - s.trust_env = False - ws = s.get(f'http://{self._chromium_options.address}/json/version', headers={'Connection': 'close'}) - bid = ws.json()['webSocketDebuggerUrl'].split('/')[-1] - self._browser = Browser(self._chromium_options.address, bid, self) - ws.close() - s.close() + self._tab = self def _d_set_runtime_settings(self): """设置运行时用到的属性""" - 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 = None if self._chromium_options.download_path is None \ - else str(Path(self._chromium_options.download_path).absolute()) - self.retry_times = self._chromium_options.retry_times - self.retry_interval = self._chromium_options.retry_interval - - def _page_init(self): - """浏览器相关设置""" - self._browser.connect_to_page() - - # ----------挂件---------- + self._timeouts = self.browser.timeouts + self._load_mode = self.browser._load_mode + self._download_path = self.browser.download_path + self.retry_times = self.browser.retry_times + self.retry_interval = self.browser.retry_interval @property def set(self): @@ -131,7 +95,7 @@ class ChromiumPage(ChromiumBase): 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) + return self.browser.latest_tab @property def process_id(self): @@ -141,7 +105,7 @@ class ChromiumPage(ChromiumBase): @property def browser_version(self): """返回所控制的浏览器版本号""" - return self._browser_version + return self._browser.version def save(self, path=None, name=None, as_pdf=False, **kwargs): """把当前页面保存为文件,如果path和name参数都为None,只返回文本 @@ -162,34 +126,7 @@ class ChromiumPage(ChromiumBase): :param as_id: 是否返回标签页id而不是标签页对象 :return: ChromiumTab对象 """ - 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) - - elif title == url == tab_type is None: - id_or_num = self.tab_id - - else: - id_or_num = self._browser.find_tabs(title, url, tab_type) - if id_or_num: - id_or_num = id_or_num[0]['id'] - else: - return None - - if as_id: - return id_or_num - - with self._lock: - return ChromiumTab(self.browser, id_or_num) + return self.browser.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,返回它们组成的列表 @@ -199,10 +136,7 @@ class ChromiumPage(ChromiumBase): :param as_id: 是否返回标签页id而不是标签页对象 :return: ChromiumTab对象组成的列表 """ - if as_id: - return [tab['id'] for tab in self._browser.find_tabs(title, url, tab_type)] - with self._lock: - return [ChromiumTab(self.browser, tab['id']) for tab in self._browser.find_tabs(title, url, tab_type)] + return self.browser.get_tabs(title=title, url=url, tab_type=tab_type, as_id=as_id) def new_tab(self, url=None, new_window=False, background=False, new_context=False): """新建一个标签页 @@ -212,10 +146,7 @@ class ChromiumPage(ChromiumBase): :param new_context: 是否创建新的上下文 :return: 新标签页对象 """ - tab = ChromiumTab(self.browser, tab_id=self.browser.new_tab(new_window, background, new_context)) - if url: - tab.get(url) - return tab + return self.browser.new_tab(url=url, new_window=new_window, background=background, new_context=new_context) def close(self): """关闭Page管理的标签页""" @@ -227,32 +158,7 @@ class ChromiumPage(ChromiumBase): :param others: 是否关闭指定标签页之外的 :return: None """ - 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_id} - 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。') - - 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.browser.close_tab(tab) - sleep(.2) - end_time = perf_counter() + 3 - while self.tabs_count != end_len and perf_counter() < end_time: - sleep(.1) + self.browser.close_tabs(tabs_or_ids=tabs_or_ids, others=others) def quit(self, timeout=5, force=True): """关闭浏览器 @@ -264,69 +170,8 @@ class ChromiumPage(ChromiumBase): def _on_disconnect(self): """浏览器退出时执行""" - ChromiumPage._PAGES.pop(self._browser_id, None) + print('kkk') + ChromiumPage._PAGES.pop(self._browser.id, None) def __repr__(self): return f'' - - -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 - - -def get_rename(original, rename): - if '.' in rename: - return rename - else: - suffix = original[original.rfind('.'):] if '.' in original else '' - return f'{rename}{suffix}' diff --git a/DrissionPage/_pages/chromium_page.pyi b/DrissionPage/_pages/chromium_page.pyi index 2aab4c1..335a604 100644 --- a/DrissionPage/_pages/chromium_page.pyi +++ b/DrissionPage/_pages/chromium_page.pyi @@ -110,12 +110,3 @@ class ChromiumPage(ChromiumBase): def quit(self, timeout: float = 5, force: bool = True) -> None: ... def _on_disconnect(self) -> None: ... - - -def handle_options(addr_or_opts): ... - - -def run_browser(chromium_options): ... - - -def get_rename(original: str, rename: str) -> str: ... diff --git a/DrissionPage/_pages/web_page.py b/DrissionPage/_pages/web_page.py index e993f35..90f3790 100644 --- a/DrissionPage/_pages/web_page.py +++ b/DrissionPage/_pages/web_page.py @@ -6,7 +6,6 @@ @License : BSD 3-Clause. """ from .chromium_page import ChromiumPage -from .chromium_tab import MixTab from .session_page import SessionPage from .._base.base import BasePage from .._configs.chromium_options import ChromiumOptions @@ -287,17 +286,16 @@ class WebPage(SessionPage, ChromiumPage, BasePage): return set_tab_cookies(self, super().cookies()) - def cookies(self, as_dict=False, all_domains=False, all_info=False): + def cookies(self, all_domains=False, all_info=False): """返回cookies - :param as_dict: 为True时以dict格式返回,为False时返回list且all_info无效 :param all_domains: 是否返回所有域的cookies :param all_info: 是否返回所有信息,False则只返回name、value、domain :return: cookies信息 """ if self._mode == 's': - return super().cookies(as_dict, all_domains, all_info) + return super().cookies(all_domains, all_info) elif self._mode == 'd': - return super(SessionPage, self).cookies(as_dict, all_domains, all_info) + return super(SessionPage, self).cookies(all_domains, all_info) def get_tab(self, id_or_num=None, title=None, url=None, tab_type='page', as_id=False): """获取一个标签页对象,id_or_num不为None时,后面几个参数无效 @@ -308,29 +306,8 @@ class WebPage(SessionPage, ChromiumPage, BasePage): :param as_id: 是否返回标签页id而不是标签页对象 :return: WebPageTab对象 """ - 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, MixTab): - return id_or_num.tab_id if as_id else id_or_num - - elif title == url == tab_type is None: - id_or_num = self.tab_id - - else: - id_or_num = self._browser.find_tabs(title, url, tab_type) - if id_or_num: - id_or_num = id_or_num[0]['id'] - else: - return None - - if as_id: - return id_or_num - - with self._lock: - return MixTab(self, id_or_num) + return self.browser._get_tab(id_or_num=id_or_num, title=title, url=url, + tab_type=tab_type, mix=True, as_id=as_id) def get_tabs(self, title=None, url=None, tab_type='page', as_id=False): """查找符合条件的tab,返回它们组成的列表 @@ -340,10 +317,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage): :param as_id: 是否返回标签页id而不是标签页对象 :return: ChromiumTab对象组成的列表 """ - if as_id: - return [tab['id'] for tab in self._browser.find_tabs(title, url, tab_type)] - with self._lock: - return [MixTab(self, tab['id']) for tab in self._browser.find_tabs(title, url, tab_type)] + return self.browser._get_tabs(title=title, url=url, tab_type=tab_type, mix=True, as_id=as_id) def new_tab(self, url=None, new_window=False, background=False, new_context=False): """新建一个标签页 @@ -353,10 +327,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage): :param new_context: 是否创建新的上下文 :return: 新标签页对象 """ - tab = MixTab(self, tab_id=self.browser.new_tab(new_window, background, new_context)) - if url: - tab.get(url) - return tab + return self.browser.new_mix_tab(url=url, new_window=new_window, background=background, new_context=new_context) def close_driver(self): """关闭driver及浏览器""" diff --git a/DrissionPage/_pages/web_page.pyi b/DrissionPage/_pages/web_page.pyi index 4adce01..1b0e28c 100644 --- a/DrissionPage/_pages/web_page.pyi +++ b/DrissionPage/_pages/web_page.pyi @@ -121,7 +121,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage): def cookies_to_browser(self) -> None: ... def cookies(self, - as_dict: bool = False, all_domains: bool = False, all_info: bool = False) -> Union[dict, list]: ...