对Page对象进行解耦,未完成

This commit is contained in:
g1879 2024-06-28 15:39:09 +08:00
parent 989c558e05
commit 9f49f874ca
31 changed files with 853 additions and 355 deletions

View File

@ -5,13 +5,13 @@
@Copyright: (c) 2024 by g1879, Inc. All Rights Reserved. @Copyright: (c) 2024 by g1879, Inc. All Rights Reserved.
@License : BSD 3-Clause. @License : BSD 3-Clause.
""" """
from ._pages.chromium_page import ChromiumPage from ._base.browser import Browser
from ._pages.session_page import SessionPage
from ._pages.web_page import WebPage
# 启动配置类
from ._configs.chromium_options import ChromiumOptions from ._configs.chromium_options import ChromiumOptions
from ._configs.session_options import SessionOptions from ._configs.session_options import SessionOptions
from ._pages.session_page import SessionPage
__all__ = ['ChromiumPage', 'ChromiumOptions', 'SessionOptions', 'SessionPage', 'WebPage', '__version__'] # from ._pages.chromium_page import ChromiumPage
__version__ = '4.0.5.4' # from ._pages.web_page import WebPage
__all__ = ['Browser', 'ChromiumOptions', 'SessionOptions', 'SessionPage', '__version__']
__version__ = '4.0.5.3'

View File

@ -54,7 +54,7 @@ class BaseElement(BaseParser):
def __init__(self, owner=None): def __init__(self, owner=None):
self.owner = owner self.owner = owner
self.page = owner._page if owner else None # self.page = owner._page if owner else None
self._type = 'BaseElement' self._type = 'BaseElement'
# ----------------以下属性或方法由后代实现---------------- # ----------------以下属性或方法由后代实现----------------
@ -366,11 +366,6 @@ class BasePage(BaseParser):
"""返回查找元素时等待的秒数""" """返回查找元素时等待的秒数"""
return self._timeout return self._timeout
@timeout.setter
def timeout(self, second):
"""设置查找元素时等待的秒数"""
self._timeout = second
@property @property
def url_available(self): def url_available(self):
"""返回当前访问的url有效性""" """返回当前访问的url有效性"""

View File

@ -59,7 +59,7 @@ class BaseElement(BaseParser):
def __init__(self, owner: BasePage = None): def __init__(self, owner: BasePage = None):
self.owner: BasePage = ... self.owner: BasePage = ...
self.page: Union[ChromiumPage, SessionPage, WebPage] = ... # self.page: Union[ChromiumPage, SessionPage, WebPage] = ...
# ----------------以下属性或方法由后代实现---------------- # ----------------以下属性或方法由后代实现----------------
@property @property

View File

@ -7,50 +7,80 @@
""" """
from pathlib import Path from pathlib import Path
from shutil import rmtree 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 websocket import WebSocketBadStatusException
from .driver import BrowserDriver, Driver 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 .._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.downloader import DownloadManager
from .._units.setter import BrowserSetter
from .._units.waiter import BrowserWaiter
from ..errors import BrowserConnectError
from ..errors import PageDisconnectedError from ..errors import PageDisconnectedError
__ERROR__ = 'error' __ERROR__ = 'error'
class Browser(object): 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 addr_or_opts: 浏览器地址:端口ChromiumOptions对象或端口数字int
:param browser_id: 浏览器id :param session_options: 使用双模Tab时使用的默认Session配置为True使用ini文件配置
:param page: ChromiumPage对象
""" """
if browser_id in cls.BROWSERS: opt = handle_options(addr_or_opts)
return cls.BROWSERS[browser_id] is_exist, browser_id = run_browser(opt)
return object.__new__(cls) 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 addr_or_opts: 浏览器地址:端口ChromiumOptions对象或端口数字int
:param browser_id: 浏览器id :param session_options: 使用双模Tab时使用的默认Session配置为True使用ini文件配置
:param page: ChromiumPage对象
""" """
if hasattr(self, '_created'): if hasattr(self, '_created'):
return return
self._created = True self._created = True
Browser.BROWSERS[browser_id] = self
self.page = page self._type = 'Browser'
self.address = address self._driver = BrowserDriver(self.id, 'browser', self.address, self)
self._driver = BrowserDriver(browser_id, 'browser', address, self) self.version = self.run_cdp('Browser.getVersion')['product']
self.id = browser_id
self._frames = {} self._frames = {}
self._drivers = {} self._drivers = {}
self._all_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 self._process_id = None
try: try:
@ -65,6 +95,9 @@ class Browser(object):
self.run_cdp('Target.setDiscoverTargets', discover=True) self.run_cdp('Target.setDiscoverTargets', discover=True)
self._driver.set_callback('Target.targetDestroyed', self._onTargetDestroyed) self._driver.set_callback('Target.targetDestroyed', self._onTargetDestroyed)
self._driver.set_callback('Target.targetCreated', self._onTargetCreated) 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): def _get_driver(self, tab_id, owner=None):
"""新建并返回指定tab id的Driver """新建并返回指定tab id的Driver
@ -104,12 +137,6 @@ class Browser(object):
self._drivers.pop(tab_id, None) self._drivers.pop(tab_id, None)
self._all_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): def run_cdp(self, cmd, **cmd_args):
"""执行Chrome DevTools Protocol语句 """执行Chrome DevTools Protocol语句
:param cmd: 协议项目 :param cmd: 协议项目
@ -121,8 +148,37 @@ class Browser(object):
return r if __ERROR__ not in r else raise_error(r, ignore) return r if __ERROR__ not in r else raise_error(r, ignore)
@property @property
def driver(self): def process_id(self):
return self._driver """返回浏览器进程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 @property
def tabs_count(self): def tabs_count(self):
@ -138,16 +194,154 @@ class Browser(object):
and not i['url'].startswith('devtools://')] and not i['url'].startswith('devtools://')]
@property @property
def process_id(self): def latest_tab(self):
"""返回浏览器进程id""" """返回最新的标签页,最新标签页指最后创建或最后被激活的
return self._process_id 当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是与关系 """查找符合条件的tab返回它们组成的列表title和url是与关系
:param title: 要匹配title的文本 :param title: 要匹配title的文本
:param url: 要匹配url的文本 :param url: 要匹配url的文本
:param tab_type: tab类型可用列表输入多个 :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 tabs = self._driver.get(f'http://{self.address}/json').json() # 不要改用cdp
@ -158,24 +352,49 @@ class Browser(object):
elif tab_type is not None: elif tab_type is not None:
raise TypeError('tab_type只能是set、list、tuple、str、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))] 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): def close_tabs(self, tabs_or_ids=None, others=False):
"""关闭标签页 """关闭传入的标签页,默认关闭当前页。可传入多个
:param tab_id: 标签页id :param tabs_or_ids: 要关闭的标签页对象或id可传入列表或元组为None时关闭最后操作的
:param others: 是否关闭指定标签页之外的
:return: None :return: None
""" """
self._onTargetDestroyed(targetId=tab_id) all_tabs = set(self.tab_ids)
self.driver.run('Target.closeTarget', targetId=tab_id) 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): if others:
"""停止一个Driver tabs = all_tabs - tabs
:param driver: Driver对象
:return: None end_len = len(set(all_tabs) - set(tabs))
""" if end_len <= 0:
driver.stop() self.quit()
self._all_drivers.get(driver.id, set()).discard(driver) 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): def activate_tab(self, tab_id):
"""使标签页变为活动状态 """使标签页变为活动状态
@ -184,37 +403,6 @@ class Browser(object):
""" """
self.run_cdp('Target.activateTarget', targetId=tab_id) 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): def reconnect(self):
"""断开重连""" """断开重连"""
self._driver.stop() self._driver.stop()
@ -234,7 +422,7 @@ class Browser(object):
self.run_cdp('Browser.close') self.run_cdp('Browser.close')
except PageDisconnectedError: except PageDisconnectedError:
pass pass
self.driver.stop() self._driver.stop()
drivers = list(self._all_drivers.values()) drivers = list(self._all_drivers.values())
for tab in drivers: for tab in drivers:
@ -276,10 +464,9 @@ class Browser(object):
break break
def _on_disconnect(self): def _on_disconnect(self):
self.page._on_disconnect() Browser._BROWSERS.pop(self.id, None)
Browser.BROWSERS.pop(self.id, None) if self._chromium_options.is_auto_port and self._chromium_options.user_data_path:
if self.page._chromium_options.is_auto_port and self.page._chromium_options.user_data_path: path = Path(self._chromium_options.user_data_path)
path = Path(self.page._chromium_options.user_data_path)
end_time = perf_counter() + 7 end_time = perf_counter() + 7
while perf_counter() < end_time: while perf_counter() < end_time:
if not path.exists(): if not path.exists():
@ -290,3 +477,57 @@ class Browser(object):
except (PermissionError, FileNotFoundError, OSError): except (PermissionError, FileNotFoundError, OSError):
pass pass
sleep(.03) sleep(.03)
def handle_options(addr_or_opts):
"""设置浏览器启动属性
:param addr_or_opts: 'ip:port'ChromiumOptionsDriver
: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

View File

@ -5,36 +5,71 @@
@Copyright: (c) 2024 by g1879, Inc. All Rights Reserved. @Copyright: (c) 2024 by g1879, Inc. All Rights Reserved.
@License : BSD 3-Clause. @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 .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.downloader import DownloadManager
from .._units.setter import BrowserSetter
from .._units.waiter import BrowserWaiter
class Browser(object): class Browser(object):
BROWSERS: dict = ...
page: ChromiumPage = ...
_driver: BrowserDriver = ...
id: str = ... id: str = ...
address: str = ... address: str = ...
version: str = ...
retry_times: int = ...
retry_interval: float = ...
_BROWSERS: dict = ...
_chromium_options: ChromiumOptions = ...
_session_options: SessionOptions = ...
_driver: BrowserDriver = ...
_frames: dict = ... _frames: dict = ...
_drivers: Dict[str, Driver] = ... _drivers: Dict[str, Driver] = ...
_all_drivers: Dict[str, Set[Driver]] = ... _all_drivers: Dict[str, Set[Driver]] = ...
_process_id: Optional[int] = ... _process_id: Optional[int] = ...
_dl_mgr: DownloadManager = ... _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 _get_driver(self, tab_id: str, owner=None) -> Driver: ...
def run_cdp(self, cmd, **cmd_args) -> dict: ... def run_cdp(self, cmd, **cmd_args) -> dict: ...
@property @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 @property
def tabs_count(self) -> int: ... def tabs_count(self) -> int: ...
@ -43,25 +78,75 @@ class Browser(object):
def tab_ids(self) -> List[str]: ... def tab_ids(self) -> List[str]: ...
@property @property
def process_id(self) -> Optional[int]: ... def latest_tab(self) -> Union[ChromiumTab, str]: ...
def find_tabs(self, title: str = None, url: str = None, def close_tabs(self,
tab_type: Union[str, list, tuple] = None) -> List[dict]: ... 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 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 reconnect(self) -> None: ...
def connect_to_page(self) -> None: ...
def _onTargetCreated(self, **kwargs) -> None: ... def _onTargetCreated(self, **kwargs) -> None: ...
def _onTargetDestroyed(self, **kwargs) -> None: ... def _onTargetDestroyed(self, **kwargs) -> None: ...

View File

@ -36,7 +36,7 @@ class ChromiumOptions(object):
om = OptionsManager(ini_path) om = OptionsManager(ini_path)
options = om.chromium_options 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._tmp_path = om.paths.get('tmp_path', None) or None
self._arguments = options.get('arguments', []) self._arguments = options.get('arguments', [])
self._browser_path = options.get('browser_path', '') self._browser_path = options.get('browser_path', '')
@ -282,14 +282,13 @@ class ChromiumOptions(object):
self._prefs = {} self._prefs = {}
return self 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 base: 默认超时时间
:param page_load: 页面加载超时时间 :param page_load: 页面加载超时时间
:param script: 脚本运行超时时间 :param script: 脚本运行超时时间
:return: 当前对象 :return: 当前对象
""" """
base = base if base is not None else implicit
if base is not None: if base is not None:
self._timeouts['base'] = base self._timeouts['base'] = base
if page_load is not None: if page_load is not None:
@ -450,7 +449,7 @@ class ChromiumOptions(object):
:param path: 下载路径 :param path: 下载路径
:return: 当前对象 :return: 当前对象
""" """
self._download_path = str(path) self._download_path = '.' if path is None else str(path)
return self return self
def set_tmp_path(self, path): def set_tmp_path(self, path):

View File

@ -24,7 +24,7 @@ class SessionOptions(object):
:param ini_path: ini文件路径 :param ini_path: ini文件路径
""" """
self.ini_path = None self.ini_path = None
self._download_path = None self._download_path = '.'
self._timeout = 10 self._timeout = 10
self._del_set = set() # 记录要从ini文件删除的参数 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.set_proxies(om.proxies.get('http', None), om.proxies.get('https', None))
self._timeout = om.timeouts.get('base', 10) 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 others = om.others
self._retry_times = others.get('retry_times', 3) self._retry_times = others.get('retry_times', 3)
@ -100,7 +100,7 @@ class SessionOptions(object):
:param path: 下载路径 :param path: 下载路径
:return: 返回当前对象 :return: 返回当前对象
""" """
self._download_path = str(path) self._download_path = '.' if path is None else str(path)
return self return self
@property @property
@ -419,7 +419,7 @@ class SessionOptions(object):
return session_options_to_dict(self) return session_options_to_dict(self)
def make_session(self): def make_session(self):
"""根据内在的配置生成Session对象ua从对象中分离""" """根据内在的配置生成Session对象headers从对象中分离"""
s = Session() s = Session()
h = CaseInsensitiveDict(self.headers) if self.headers else CaseInsensitiveDict() h = CaseInsensitiveDict(self.headers) if self.headers else CaseInsensitiveDict()

View File

@ -183,7 +183,7 @@ class ChromiumElement(DrissionElement):
def select(self) -> SelectElement: ... def select(self) -> SelectElement: ...
@property @property
def value(self) -> None: ... def value(self) -> str: ...
def check(self, uncheck: bool = False, by_js: bool = False) -> None: ... def check(self, uncheck: bool = False, by_js: bool = False) -> None: ...

View File

@ -38,11 +38,10 @@ __ERROR__ = 'error'
class ChromiumBase(BasePage): class ChromiumBase(BasePage):
"""标签页、frame、页面基类""" """标签页、frame、页面基类"""
def __init__(self, address, tab_id=None, timeout=None): def __init__(self, address, tab_id=None):
""" """
:param address: 浏览器 ip:port :param address: 浏览器 ip:port
:param tab_id: 要控制的标签页id不指定默认为激活的 :param tab_id: 要控制的标签页id不指定默认为激活的
:param timeout: 超时时间
""" """
super().__init__() super().__init__()
self._is_loading = None self._is_loading = None
@ -71,8 +70,6 @@ class ChromiumBase(BasePage):
self._d_set_start_options(address) self._d_set_start_options(address)
self._d_set_runtime_settings() self._d_set_runtime_settings()
self._connect_browser(tab_id) self._connect_browser(tab_id)
if timeout is not None:
self.timeout = timeout
def _d_set_start_options(self, address): def _d_set_start_options(self, address):
"""设置浏览器启动属性 """设置浏览器启动属性
@ -93,7 +90,7 @@ class ChromiumBase(BasePage):
self._is_reading = False self._is_reading = False
if not tab_id: 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 tabs = [(i['id'], i['url']) for i in tabs
if i['type'] in ('page', 'webview') and not i['url'].startswith('devtools://')] if i['type'] in ('page', 'webview') and not i['url'].startswith('devtools://')]
dialog = None dialog = None
@ -861,7 +858,8 @@ class ChromiumBase(BasePage):
def disconnect(self): def disconnect(self):
"""断开与页面的连接,不关闭页面""" """断开与页面的连接,不关闭页面"""
if self._driver: 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): def reconnect(self, wait=0):
"""断开与页面原来的页面,重新建立连接 """断开与页面原来的页面,重新建立连接
@ -1119,15 +1117,12 @@ class ChromiumBase(BasePage):
class Timeout(object): class Timeout(object):
"""用于保存d模式timeout信息的类""" """用于保存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 base: 默认超时时间
:param page_load: 页面加载超时时间 :param page_load: 页面加载超时时间
:param script: js超时时间 :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.base = 10 if base is None else base
self.page_load = 30 if page_load is None else page_load self.page_load = 30 if page_load is None else page_load
self.script = 30 if script is None else script self.script = 30 if script is None else script

View File

@ -32,8 +32,7 @@ PIC_TYPE = Literal['jpg', 'jpeg', 'png', 'webp', True]
class ChromiumBase(BasePage): class ChromiumBase(BasePage):
def __init__(self, def __init__(self,
address: Union[str, int], address: Union[str, int],
tab_id: str = None, tab_id: str = None):
timeout: float = None):
self._browser: Browser = ... self._browser: Browser = ...
self._page: ChromiumPage = ... self._page: ChromiumPage = ...
self.tab: Union[ChromiumPage, ChromiumTab] = ... self.tab: Union[ChromiumPage, ChromiumTab] = ...
@ -267,8 +266,7 @@ class ChromiumBase(BasePage):
class Timeout(object): class Timeout(object):
def __init__(self, page: ChromiumBase, base=None, page_load=None, script=None): def __init__(self, base=None, page_load=None, script=None):
self._page: ChromiumBase = ...
self.base: float = ... self.base: float = ...
self.page_load: float = ... self.page_load: float = ...
self.script: float = ... self.script: float = ...

View File

@ -27,14 +27,17 @@ class ChromiumFrame(ChromiumBase):
:param ele: frame所在元素 :param ele: frame所在元素
:param info: frame所在元素信息 :param info: frame所在元素信息
""" """
if owner._type in ('ChromiumPage', 'WebPage'): self.tab = owner.tab
self._page = self._target_page = self.tab = owner
self._browser = owner.browser self._browser = owner.browser
else: # Tab、Frame
self._page = owner.page
self._browser = self._page.browser
self._target_page = owner self._target_page = owner
self.tab = owner.tab if owner._type == 'ChromiumFrame' else 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.address = owner.address
self._tab_id = owner.tab_id self._tab_id = owner.tab_id
@ -250,10 +253,10 @@ class ChromiumFrame(ChromiumBase):
"""返回cdp中的node id""" """返回cdp中的node id"""
return self.frame_ele._node_id return self.frame_ele._node_id
@property # @property
def page(self): # def page(self):
"""返回所属Page对象""" # """返回所属Page对象"""
return self._page # return self._page
@property @property
def owner(self): def owner(self):

View File

@ -9,7 +9,7 @@ from pathlib import Path
from typing import Union, Tuple, List, Any, Optional from typing import Union, Tuple, List, Any, Optional
from .chromium_base import ChromiumBase from .chromium_base import ChromiumBase
from .chromium_page import ChromiumPage # from .chromium_page import ChromiumPage
from .chromium_tab import ChromiumTab from .chromium_tab import ChromiumTab
from .web_page import WebPage from .web_page import WebPage
from .._elements.chromium_element import ChromiumElement from .._elements.chromium_element import ChromiumElement
@ -25,13 +25,12 @@ from .._units.waiter import FrameWaiter
class ChromiumFrame(ChromiumBase): class ChromiumFrame(ChromiumBase):
def __init__(self, def __init__(self,
owner: Union[ChromiumPage, WebPage, ChromiumTab, ChromiumFrame], owner: Union[ChromiumTab, ChromiumFrame],
ele: ChromiumElement, ele: ChromiumElement,
info: dict = None): info: dict = None):
self._target_page: ChromiumBase = ... self._target_page: ChromiumBase = ...
self._page: ChromiumPage = ... self.tab: ChromiumTab = ...
self.tab: Union[ChromiumPage, ChromiumTab] = ... # self._tab_id: str = ...
self._tab_id: str = ...
self._set: ChromiumFrameSetter = ... self._set: ChromiumFrameSetter = ...
self._frame_ele: ChromiumElement = ... self._frame_ele: ChromiumElement = ...
self._backend_id: int = ... self._backend_id: int = ...
@ -66,8 +65,8 @@ class ChromiumFrame(ChromiumBase):
def _onInspectorDetached(self, **kwargs): ... def _onInspectorDetached(self, **kwargs): ...
@property # @property
def page(self) -> Union[ChromiumPage, WebPage]: ... # def page(self) -> Union[ChromiumPage, WebPage]: ...
@property @property
def owner(self) -> ChromiumBase: ... def owner(self) -> ChromiumBase: ...

View File

@ -34,19 +34,16 @@ class ChromiumPage(ChromiumBase):
:param tab_id: 要控制的标签页id不指定默认为激活的 :param tab_id: 要控制的标签页id不指定默认为激活的
:param timeout: 超时时间 :param timeout: 超时时间
""" """
opt = handle_options(addr_or_opts) browser = Browser(addr_or_opts=addr_or_opts)
is_exist, browser_id = run_browser(opt) if browser.id in cls._PAGES:
if browser_id in cls._PAGES: r = cls._PAGES[browser.id]
r = cls._PAGES[browser_id]
while not hasattr(r, '_frame_id'): while not hasattr(r, '_frame_id'):
sleep(.1) sleep(.1)
return r return r
r = object.__new__(cls) r = object.__new__(cls)
r._chromium_options = opt r._browser = browser
r._is_exist = is_exist cls._PAGES[browser.id] = r
r._browser_id = browser_id
r.address = opt.address
cls._PAGES[browser_id] = r
return r return r
def __init__(self, addr_or_opts=None, tab_id=None, timeout=None): def __init__(self, addr_or_opts=None, tab_id=None, timeout=None):
@ -59,12 +56,9 @@ class ChromiumPage(ChromiumBase):
return return
self._created = True self._created = True
self._page = self
self.tab = self self.tab = self
self._run_browser() super().__init__(self.browser.address, tab_id)
super().__init__(self.address, tab_id)
self._type = 'ChromiumPage' self._type = 'ChromiumPage'
self._lock = Lock()
self.set.timeouts(base=timeout) self.set.timeouts(base=timeout)
self._page_init() self._page_init()
@ -86,7 +80,7 @@ class ChromiumPage(ChromiumBase):
def _d_set_runtime_settings(self): 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'], script=self._chromium_options.timeouts['script'],
base=self._chromium_options.timeouts['base']) base=self._chromium_options.timeouts['base'])
if self._chromium_options.timeouts['base'] is not None: if self._chromium_options.timeouts['base'] is not None:
@ -196,7 +190,7 @@ class ChromiumPage(ChromiumBase):
return id_or_num return id_or_num
with self._lock: 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): def get_tabs(self, title=None, url=None, tab_type='page', as_id=False):
"""查找符合条件的tab返回它们组成的列表 """查找符合条件的tab返回它们组成的列表
@ -209,7 +203,7 @@ class ChromiumPage(ChromiumBase):
if as_id: if as_id:
return [tab['id'] for tab in self._browser.find_tabs(title, url, tab_type)] return [tab['id'] for tab in self._browser.find_tabs(title, url, tab_type)]
with self._lock: 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): def new_tab(self, url=None, new_window=False, background=False, new_context=False):
"""新建一个标签页 """新建一个标签页
@ -219,7 +213,7 @@ class ChromiumPage(ChromiumBase):
:param new_context: 是否创建新的上下文 :param new_context: 是否创建新的上下文
:return: 新标签页对象 :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: if url:
tab.get(url) tab.get(url)
return tab return tab

View File

@ -20,6 +20,10 @@ from .._units.waiter import PageWaiter
class ChromiumPage(ChromiumBase): class ChromiumPage(ChromiumBase):
_PAGES: dict = ... _PAGES: dict = ...
tab: ChromiumPage = ...
_browser: Browser = ...
_rect: Optional[TabRect] = ...
_is_exist: bool = ...
def __new__(cls, def __new__(cls,
addr_or_opts: Union[str, int, ChromiumOptions] = None, addr_or_opts: Union[str, int, ChromiumOptions] = None,
@ -29,15 +33,7 @@ class ChromiumPage(ChromiumBase):
def __init__(self, def __init__(self,
addr_or_opts: Union[str, int, ChromiumOptions] = None, addr_or_opts: Union[str, int, ChromiumOptions] = None,
tab_id: str = None, tab_id: str = None,
timeout: float = 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 = ...
def _handle_options(self, addr_or_opts: Union[str, ChromiumOptions]) -> str: ... def _handle_options(self, addr_or_opts: Union[str, ChromiumOptions]) -> str: ...

View File

@ -22,10 +22,10 @@ class ChromiumTab(ChromiumBase):
"""实现浏览器标签页的类""" """实现浏览器标签页的类"""
_TABS = {} _TABS = {}
def __new__(cls, page, tab_id): def __new__(cls, browser, tab_id):
""" """
:param page: ChromiumPage对象 :param browser: Browser对象
:param tab_id: 要控制的标签页id :param tab_id: 标签页id
""" """
if Settings.singleton_tab_obj and tab_id in cls._TABS: if Settings.singleton_tab_obj and tab_id in cls._TABS:
r = cls._TABS[tab_id] r = cls._TABS[tab_id]
@ -36,38 +36,32 @@ class ChromiumTab(ChromiumBase):
cls._TABS[tab_id] = r cls._TABS[tab_id] = r
return r return r
def __init__(self, page, tab_id): def __init__(self, browser, tab_id):
""" """
:param page: ChromiumPage对象 :param browser: Browser对象
:param tab_id: 要控制的标签页id :param tab_id: 标签页id
""" """
if Settings.singleton_tab_obj and hasattr(self, '_created'): if Settings.singleton_tab_obj and hasattr(self, '_created'):
return return
self._created = True self._created = True
self._page = page
self.tab = self self.tab = self
self._browser = page.browser self._browser = browser
super().__init__(page.address, tab_id, page.timeout) super().__init__(browser.address, tab_id, browser.timeout)
self._rect = None self._rect = None
self._type = 'ChromiumTab' self._type = 'ChromiumTab'
def _d_set_runtime_settings(self): def _d_set_runtime_settings(self):
"""重写设置浏览器运行参数方法""" """重写设置浏览器运行参数方法"""
self._timeouts = copy(self.page.timeouts) self._timeouts = copy(self.browser.timeouts)
self.retry_times = self.page.retry_times self.retry_times = self.browser.retry_times
self.retry_interval = self.page.retry_interval self.retry_interval = self.browser.retry_interval
self._load_mode = self.page._load_mode self._load_mode = self.browser._load_mode
self._download_path = self.page.download_path self._download_path = self.browser.download_path
def close(self): def close(self):
"""关闭当前标签页""" """关闭当前标签页"""
self.page.close_tabs(self.tab_id) self.browser.close_tabs(self.tab_id)
@property
def page(self):
"""返回总体page对象"""
return self._page
@property @property
def set(self): def set(self):
@ -100,11 +94,11 @@ class ChromiumTab(ChromiumBase):
ChromiumTab._TABS.pop(self.tab_id, None) ChromiumTab._TABS.pop(self.tab_id, None)
class WebPageTab(SessionPage, ChromiumTab, BasePage): class MixTab(SessionPage, ChromiumTab, BasePage):
def __init__(self, page, tab_id): def __init__(self, browser, tab_id):
""" """
:param page: WebPage对象 :param browser: Browser对象
:param tab_id: 要控制的标签页id :param tab_id: 标签页id
""" """
if Settings.singleton_tab_obj and hasattr(self, '_created'): if Settings.singleton_tab_obj and hasattr(self, '_created'):
return return
@ -112,10 +106,9 @@ class WebPageTab(SessionPage, ChromiumTab, BasePage):
self._mode = 'd' self._mode = 'd'
self._has_driver = True self._has_driver = True
self._has_session = True self._has_session = True
super().__init__(session_or_options=SessionOptions(read_file=False).from_session(copy(page.session), super().__init__(session_or_options=browser._session_options if browser._session_options else SessionOptions())
page._headers)) super(SessionPage, self).__init__(browser=browser, tab_id=tab_id)
super(SessionPage, self).__init__(page=page, tab_id=tab_id) self._type = 'MixTab'
self._type = 'WebPageTab'
def __call__(self, locator, index=1, timeout=None): def __call__(self, locator, index=1, timeout=None):
"""在内部查找元素 """在内部查找元素
@ -318,7 +311,9 @@ class WebPageTab(SessionPage, ChromiumTab, BasePage):
# s模式转d模式 # s模式转d模式
if self._mode == 'd': if self._mode == 'd':
if self._driver is None: 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._url = None if not self._has_driver else super(SessionPage, self).url
self._has_driver = True self._has_driver = True
@ -378,7 +373,7 @@ class WebPageTab(SessionPage, ChromiumTab, BasePage):
def close(self): def close(self):
"""关闭当前标签页""" """关闭当前标签页"""
self.page.close_tabs(self.tab_id) self.browser.close_tabs(self.tab_id)
self._session.close() self._session.close()
if self._response is not None: if self._response is not None:
self._response.close() 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) return super(SessionPage, self)._find_elements(locator, timeout=timeout, index=index, relative=relative)
def __repr__(self): def __repr__(self):
return f'<WebPageTab browser_id={self.browser.id} tab_id={self.tab_id}>' return f'<MixTab browser_id={self.browser.id} tab_id={self.tab_id}>'

View File

@ -12,9 +12,7 @@ from requests import Session, Response
from .chromium_base import ChromiumBase from .chromium_base import ChromiumBase
from .chromium_frame import ChromiumFrame from .chromium_frame import ChromiumFrame
from .chromium_page import ChromiumPage
from .session_page import SessionPage from .session_page import SessionPage
from .web_page import WebPage
from .._base.browser import Browser from .._base.browser import Browser
from .._elements.chromium_element import ChromiumElement from .._elements.chromium_element import ChromiumElement
from .._elements.session_element import SessionElement from .._elements.session_element import SessionElement
@ -27,10 +25,9 @@ from .._units.waiter import TabWaiter
class ChromiumTab(ChromiumBase): class ChromiumTab(ChromiumBase):
_TABS: dict = ... _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): def __init__(self, browser: Browser, tab_id: str):
self._page: ChromiumPage = ...
self._browser: Browser = ... self._browser: Browser = ...
self._rect: Optional[TabRect] = ... self._rect: Optional[TabRect] = ...
@ -38,9 +35,6 @@ class ChromiumTab(ChromiumBase):
def close(self) -> None: ... def close(self) -> None: ...
@property
def page(self) -> ChromiumPage: ...
@property @property
def set(self) -> TabSetter: ... def set(self) -> TabSetter: ...
@ -69,22 +63,19 @@ class ChromiumTab(ChromiumBase):
generateDocumentOutline: bool = ...) -> Union[bytes, str]: ... generateDocumentOutline: bool = ...) -> Union[bytes, str]: ...
class WebPageTab(SessionPage, ChromiumTab): class MixTab(SessionPage, ChromiumTab):
def __init__(self, page: WebPage, tab_id: str): _browser: Browser = ...
self._page: WebPage = ... _mode: str = ...
self._browser: Browser = ... _has_driver: bool = ...
self._mode: str = ... _has_session: bool = ...
self._has_driver = ...
self._has_session = ... def __init__(self, browser: Browser, tab_id: str): ...
def __call__(self, def __call__(self,
locator: Union[Tuple[str, str], str, ChromiumElement, SessionElement], locator: Union[Tuple[str, str], str, ChromiumElement, SessionElement],
index: int = 1, index: int = 1,
timeout: float = None) -> Union[ChromiumElement, SessionElement]: ... timeout: float = None) -> Union[ChromiumElement, SessionElement]: ...
@property
def page(self) -> WebPage: ...
@property @property
def url(self) -> Union[str, None]: ... def url(self) -> Union[str, None]: ...
@ -121,9 +112,6 @@ class WebPageTab(SessionPage, ChromiumTab):
@property @property
def timeout(self) -> float: ... def timeout(self) -> float: ...
@timeout.setter
def timeout(self, second: float) -> None: ...
def get(self, def get(self,
url: str, url: str,
show_errmsg: bool = False, show_errmsg: bool = False,

View File

@ -6,7 +6,7 @@
@License : BSD 3-Clause. @License : BSD 3-Clause.
""" """
from .chromium_page import ChromiumPage from .chromium_page import ChromiumPage
from .chromium_tab import WebPageTab from .chromium_tab import MixTab
from .session_page import SessionPage from .session_page import SessionPage
from .._base.base import BasePage from .._base.base import BasePage
from .._configs.chromium_options import ChromiumOptions from .._configs.chromium_options import ChromiumOptions
@ -150,14 +150,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
"""返回通用timeout设置""" """返回通用timeout设置"""
return self.timeouts.base 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): def get(self, url, show_errmsg=False, retry=None, interval=None, timeout=None, **kwargs):
"""跳转到一个url """跳转到一个url
:param url: 目标url :param url: 目标url
@ -321,7 +313,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
id_or_num = id_or_num id_or_num = id_or_num
elif isinstance(id_or_num, int): 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] 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 return id_or_num.tab_id if as_id else id_or_num
elif title == url == tab_type is None: elif title == url == tab_type is None:
@ -338,7 +330,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
return id_or_num return id_or_num
with self._lock: 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): def get_tabs(self, title=None, url=None, tab_type='page', as_id=False):
"""查找符合条件的tab返回它们组成的列表 """查找符合条件的tab返回它们组成的列表
@ -351,7 +343,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
if as_id: if as_id:
return [tab['id'] for tab in self._browser.find_tabs(title, url, tab_type)] return [tab['id'] for tab in self._browser.find_tabs(title, url, tab_type)]
with self._lock: 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): 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: 是否创建新的上下文 :param new_context: 是否创建新的上下文
:return: 新标签页对象 :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: if url:
tab.get(url) tab.get(url)
return tab return tab

View File

@ -11,7 +11,7 @@ from requests import Session, Response
from .chromium_frame import ChromiumFrame from .chromium_frame import ChromiumFrame
from .chromium_page import ChromiumPage from .chromium_page import ChromiumPage
from .chromium_tab import WebPageTab from .chromium_tab import MixTab
from .session_page import SessionPage from .session_page import SessionPage
from .._base.base import BasePage from .._base.base import BasePage
from .._base.driver import Driver from .._base.driver import Driver
@ -129,23 +129,23 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
all_info: bool = False) -> Union[dict, list]: ... all_info: bool = False) -> Union[dict, list]: ...
def get_tab(self, def get_tab(self,
id_or_num: Union[str, WebPageTab, int] = None, id_or_num: Union[str, MixTab, int] = None,
title: str = None, title: str = None,
url: str = None, url: str = None,
tab_type: Union[str, list, tuple] = 'page', 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, def get_tabs(self,
title: str = None, title: str = None,
url: str = None, url: str = None,
tab_type: Union[str, list, tuple] = 'page', 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, def new_tab(self,
url: str = None, url: str = None,
new_window: bool = False, new_window: bool = False,
background: bool = False, background: bool = False,
new_context: bool = False) -> WebPageTab: ... new_context: bool = False) -> MixTab: ...
def close_driver(self) -> None: ... def close_driver(self) -> None: ...
@ -175,7 +175,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
cert: Any | None = ...) -> Union[bool, Response]: ... cert: Any | None = ...) -> Union[bool, Response]: ...
@property @property
def latest_tab(self) -> Union[WebPageTab, WebPage]: ... def latest_tab(self) -> Union[MixTab, WebPage]: ...
@property @property
def set(self) -> WebPageSetter: ... def set(self) -> WebPageSetter: ...

View File

@ -120,12 +120,13 @@ class Clicker(object):
""" """
self._ele.owner.scroll.to_see(self._ele) self._ele.owner.scroll.to_see(self._ele)
x, y = self._ele.rect.viewport_click_point x, y = self._ele.rect.viewport_click_point
curr_tid = self._ele.tab.browser.tab_ids[0]
self._click(x, y, 'middle') self._click(x, y, 'middle')
if get_tab: 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: if not tid:
raise RuntimeError('没有出现新标签页。') 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): def at(self, offset_x=None, offset_y=None, button='left', count=1):
"""带偏移量点击本元素相对于左上角坐标。不传入x或y值时点击元素中间点 """带偏移量点击本元素相对于左上角坐标。不传入x或y值时点击元素中间点
@ -162,16 +163,15 @@ class Clicker(object):
""" """
if save_path: if save_path:
self._ele.owner.tab.set.download_path(save_path) self._ele.owner.tab.set.download_path(save_path)
elif not self._ele.page._browser._dl_mgr._running: elif not self._ele.tab._browser._dl_mgr._running:
self._ele.page.set.download_path('.') 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: if rename or suffix:
self._ele.owner.tab.set.download_file_name(rename, suffix) obj.set.download_file_name(rename, suffix)
tab = self._ele.page if new_tab else self._ele.owner
self.left(by_js=by_js) 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): def to_upload(self, file_paths, by_js=False):
"""触发上传文件选择框并自动填入指定路径 """触发上传文件选择框并自动填入指定路径
@ -183,16 +183,18 @@ class Clicker(object):
self.left(by_js=by_js) self.left(by_js=by_js)
self._ele.owner.wait.upload_paths_inputted() 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出现并返回其对象 """点击后等待新tab出现并返回其对象
:param by_js: 是否使用js点击逻辑与click()一致 :param by_js: 是否使用js点击逻辑与click()一致
:param timeout: 等待超时时间
:return: 新标签页对象如果没有等到新标签页出现则抛出异常 :return: 新标签页对象如果没有等到新标签页出现则抛出异常
""" """
curr_tid = self._ele.tab.browser.tab_ids[0]
self.left(by_js=by_js) 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: if not tid:
raise RuntimeError('没有出现新标签页。') 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): def _click(self, client_x, client_y, button='left', count=1):
"""实施点击 """实施点击

View File

@ -10,7 +10,7 @@ from typing import Union
from .downloader import DownloadMission from .downloader import DownloadMission
from .._elements.chromium_element import ChromiumElement from .._elements.chromium_element import ChromiumElement
from .._pages.chromium_tab import WebPageTab, ChromiumTab from .._pages.chromium_tab import MixTab, ChromiumTab
class Clicker(object): class Clicker(object):
@ -23,7 +23,7 @@ class Clicker(object):
def right(self) -> None: ... 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, def at(self,
offset_x: float = None, 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 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: ... def _click(self, client_x: float, client_y: float, button: str = 'left', count: int = 1) -> None: ...

View File

@ -9,7 +9,7 @@ from http.cookiejar import Cookie, CookieJar
from typing import Union from typing import Union
from .._pages.chromium_base import ChromiumBase 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.session_page import SessionPage
from .._pages.web_page import WebPage from .._pages.web_page import WebPage
@ -39,7 +39,7 @@ class SessionCookiesSetter(object):
class WebPageCookiesSetter(CookiesSetter, SessionCookiesSetter): class WebPageCookiesSetter(CookiesSetter, SessionCookiesSetter):
_owner: Union[WebPage, WebPageTab] _owner: Union[WebPage, MixTab]
def __init__(self, page: SessionPage): ... def __init__(self, page: SessionPage): ...

View File

@ -20,20 +20,23 @@ class DownloadManager(object):
:param browser: Browser对象 :param browser: Browser对象
""" """
self._browser = browser self._browser = browser
self._page = browser.page # self._page = browser.page
self._when_download_file_exists = 'rename' # self._when_download_file_exists = 'rename'
self._save_path = None # 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._missions = {} # {guid: DownloadMission}
self._tab_missions = {} # {tab_id: DownloadMission} self._tab_missions = {} # {tab_id: DownloadMission}
self._flags = {} # {tab_id: [bool, DownloadMission]} self._flags = {} # {tab_id: [bool, DownloadMission]}
if self._page.download_path: # if self._page.download_path:
self.set_path(self._page, self._page.download_path) # self.set_path(self._page, self._page.download_path)
else:
self._running = False self._running = False
@property @property
@ -47,13 +50,13 @@ class DownloadManager(object):
:param path: 下载路径绝对路径str :param path: 下载路径绝对路径str
:return: None :return: None
""" """
TabDownloadSettings(tab.tab_id).path = path tid = tab if isinstance(tab, str) else tab.tab_id
if tab is self._page or not self._running: TabDownloadSettings(tid).path = path
self._browser.driver.set_callback('Browser.downloadProgress', self._onDownloadProgress) if not self._running or tid == 'browser':
self._browser.driver.set_callback('Browser.downloadWillBegin', self._onDownloadWillBegin) self._browser._driver.set_callback('Browser.downloadProgress', self._onDownloadProgress)
r = self._browser.run_cdp('Browser.setDownloadBehavior', downloadPath=path, 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) behavior='allowAndName', eventsEnabled=True)
self._save_path = path
if 'error' in r: if 'error' in r:
print('浏览器版本太低无法使用下载管理功能。') print('浏览器版本太低无法使用下载管理功能。')
self._running = True self._running = True
@ -149,10 +152,11 @@ class DownloadManager(object):
def _onDownloadWillBegin(self, **kwargs): def _onDownloadWillBegin(self, **kwargs):
"""用于获取弹出新标签页触发的下载任务""" """用于获取弹出新标签页触发的下载任务"""
# print(kwargs)
guid = kwargs['guid'] 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.rename:
if settings.suffix is not None: if settings.suffix is not None:
name = f'{settings.rename}.{settings.suffix}' if settings.suffix else settings.rename 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': elif settings.when_file_exists == 'overwrite':
goal_path.unlink() 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 self._missions[guid] = m
if self.get_flag(tab_id) is False: # 取消该任务 if self.get_flag(tab_id) is False: # 取消该任务
@ -199,6 +203,7 @@ class DownloadManager(object):
def _onDownloadProgress(self, **kwargs): def _onDownloadProgress(self, **kwargs):
"""下载状态变化时执行""" """下载状态变化时执行"""
# print(kwargs)
if kwargs['guid'] in self._missions: if kwargs['guid'] in self._missions:
mission = self._missions[kwargs['guid']] mission = self._missions[kwargs['guid']]
if kwargs['state'] == 'inProgress': if kwargs['state'] == 'inProgress':

View File

@ -9,28 +9,28 @@ from typing import Dict, Optional, Union, Literal
from .._base.browser import Browser from .._base.browser import Browser
from .._pages.chromium_base import ChromiumBase 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): class DownloadManager(object):
_browser: Browser = ... _browser: Browser = ...
_page: ChromiumPage = ... # _page: ChromiumPage = ...
_missions: Dict[str, DownloadMission] = ... _missions: Dict[str, DownloadMission] = ...
_tab_missions: dict = ... _tab_missions: dict = ...
_flags: dict = ... _flags: dict = ...
_running: bool = ... _running: bool = ...
_save_path: Optional[str] = ... # _save_path: Optional[str] = ...
def __init__(self, browser: Browser): ... def __init__(self, browser: Browser): ...
@property @property
def missions(self) -> Dict[str, DownloadMission]: ... 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_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: ... def set_flag(self, tab_id: str, flag: Union[bool, DownloadMission, None]) -> None: ...

View File

@ -184,7 +184,7 @@ class TabRect(object):
def _get_window_rect(self): 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): class FrameRect(object):

View File

@ -12,7 +12,7 @@ from .._elements.chromium_element import ChromiumElement
from .._pages.chromium_base import ChromiumBase from .._pages.chromium_base import ChromiumBase
from .._pages.chromium_frame import ChromiumFrame from .._pages.chromium_frame import ChromiumFrame
from .._pages.chromium_page import ChromiumPage 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 from .._pages.web_page import WebPage
@ -63,7 +63,7 @@ class ElementRect(object):
class TabRect(object): class TabRect(object):
def __init__(self, owner: ChromiumBase): def __init__(self, owner: ChromiumBase):
self._owner: Union[ChromiumPage, ChromiumTab, WebPage, WebPageTab] = ... self._owner: Union[ChromiumPage, ChromiumTab, WebPage, MixTab] = ...
@property @property
def window_state(self) -> str: ... def window_state(self) -> str: ...

View File

@ -39,9 +39,9 @@ class Screencast(object):
raise ValueError('save_path必须设置。') raise ValueError('save_path必须设置。')
if self._mode in ('frugal_video', 'video'): 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._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: else:
self._tmp_path = Path(gettempdir()) / 'DrissionPage' / f'screencast_tmp_{time()}_{randint(0, 100)}' self._tmp_path = Path(gettempdir()) / 'DrissionPage' / f'screencast_tmp_{time()}_{randint(0, 100)}'
self._tmp_path.mkdir(parents=True, exist_ok=True) self._tmp_path.mkdir(parents=True, exist_ok=True)

View File

@ -33,8 +33,101 @@ class BasePageSetter(object):
self._owner._none_ele_return_value = on_off self._owner._none_ele_return_value = on_off
self._owner._none_ele_value = value 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): def __init__(self, owner):
""" """
:param owner: ChromiumBase对象 :param owner: ChromiumBase对象
@ -42,10 +135,10 @@ class ChromiumBaseSetter(BasePageSetter):
super().__init__(owner) super().__init__(owner)
self._cookies_setter = None self._cookies_setter = None
@property # @property
def load_mode(self): # def load_mode(self):
"""返回用于设置页面加载策略的对象""" # """返回用于设置页面加载策略的对象"""
return LoadMode(self._owner) # return LoadMode(self._owner)
@property @property
def scroll(self): def scroll(self):
@ -59,31 +152,30 @@ class ChromiumBaseSetter(BasePageSetter):
self._cookies_setter = CookiesSetter(self._owner) self._cookies_setter = CookiesSetter(self._owner)
return self._cookies_setter return self._cookies_setter
def retry_times(self, times): # def retry_times(self, times):
"""设置连接失败重连次数""" # """设置连接失败重连次数"""
self._owner.retry_times = times # self._owner.retry_times = times
#
def retry_interval(self, interval): # def retry_interval(self, interval):
"""设置连接失败重连间隔""" # """设置连接失败重连间隔"""
self._owner.retry_interval = interval # self._owner.retry_interval = interval
#
def timeouts(self, base=None, page_load=None, script=None, implicit=None): # def timeouts(self, base=None, page_load=None, script=None):
"""设置超时时间,单位为秒 # """设置超时时间,单位为秒
:param base: 基本等待时间除页面加载和脚本超时其它等待默认使用 # :param base: 基本等待时间,除页面加载和脚本超时,其它等待默认使用
:param page_load: 页面加载超时时间 # :param page_load: 页面加载超时时间
:param script: 脚本运行超时时间 # :param script: 脚本运行超时时间
:return: None # :return: None
""" # """
base = base if base is not None else implicit # if base is not None:
if base is not None: # self._owner.timeouts.base = base
self._owner.timeouts.base = base # self._owner._timeout = base
self._owner._timeout = base #
# if page_load is not None:
if page_load is not None: # self._owner.timeouts.page_load = page_load
self._owner.timeouts.page_load = page_load #
# if script is not None:
if script is not None: # self._owner.timeouts.script = script
self._owner.timeouts.script = script
def user_agent(self, ua, platform=None): def user_agent(self, ua, platform=None):
"""为当前tab设置user agent只在当前tab有效 """为当前tab设置user agent只在当前tab有效
@ -191,11 +283,10 @@ class TabSetter(ChromiumBaseSetter):
:param path: 下载路径 :param path: 下载路径
:return: None :return: None
""" """
path = str(Path(path).absolute()) super().download_path(path)
self._owner._download_path = path self._owner.browser._dl_mgr.set_path(self._owner, self._owner._download_path)
self._owner.browser._dl_mgr.set_path(self._owner, path)
if self._owner._DownloadKit: 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): def download_file_name(self, name=None, suffix=None):
"""设置下一个被下载文件的名称 """设置下一个被下载文件的名称
@ -215,7 +306,6 @@ class TabSetter(ChromiumBaseSetter):
mode = types.get(mode, mode) mode = types.get(mode, mode)
if mode not in types: if mode not in types:
raise ValueError(f'''mode参数只能是 '{"', '".join(types.keys())}' 之一,现在是:{mode}''') raise ValueError(f'''mode参数只能是 '{"', '".join(types.keys())}' 之一,现在是:{mode}''')
self._owner.browser._dl_mgr.set_file_exists(self._owner.tab_id, mode) self._owner.browser._dl_mgr.set_file_exists(self._owner.tab_id, mode)
def activate(self): def activate(self):
@ -264,23 +354,22 @@ class SessionPageSetter(BasePageSetter):
self._cookies_setter = SessionCookiesSetter(self._owner) self._cookies_setter = SessionCookiesSetter(self._owner)
return self._cookies_setter return self._cookies_setter
def retry_times(self, times): # def retry_times(self, times):
"""设置连接失败时重连次数""" # """设置连接失败时重连次数"""
self._owner.retry_times = times # self._owner.retry_times = times
#
def retry_interval(self, interval): # def retry_interval(self, interval):
"""设置连接失败时重连间隔""" # """设置连接失败时重连间隔"""
self._owner.retry_interval = interval # self._owner.retry_interval = interval
def download_path(self, path): def download_path(self, path):
"""设置下载路径 """设置下载路径
:param path: 下载路径 :param path: 下载路径
:return: None :return: None
""" """
path = str(Path(path).absolute()) super().download_path(path)
self._owner._download_path = path
if self._owner._DownloadKit: 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): def timeout(self, second):
"""设置连接超时时间 """设置连接超时时间
@ -464,7 +553,7 @@ class ChromiumElementSetter(object):
""" """
self._ele = ele self._ele = ele
def attr(self, name, value): def attr(self, name, value=''):
"""设置元素attribute属性 """设置元素attribute属性
:param name: 属性名 :param name: 属性名
:param value: 属性值 :param value: 属性值

View File

@ -14,11 +14,12 @@ from requests.auth import HTTPBasicAuth
from .cookies_setter import SessionCookiesSetter, CookiesSetter, WebPageCookiesSetter from .cookies_setter import SessionCookiesSetter, CookiesSetter, WebPageCookiesSetter
from .scroller import PageScroller from .scroller import PageScroller
from .._base.base import BasePage from .._base.base import BasePage
from .._base.browser import Browser
from .._elements.chromium_element import ChromiumElement from .._elements.chromium_element import ChromiumElement
from .._pages.chromium_base import ChromiumBase from .._pages.chromium_base import ChromiumBase
from .._pages.chromium_frame import ChromiumFrame from .._pages.chromium_frame import ChromiumFrame
from .._pages.chromium_page import ChromiumPage 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.session_page import SessionPage
from .._pages.web_page import WebPage from .._pages.web_page import WebPage
@ -26,11 +27,41 @@ FILE_EXISTS = Literal['skip', 'rename', 'overwrite', 's', 'r', 'o']
class BasePageSetter(object): class BasePageSetter(object):
def __init__(self, owner: BasePage): def __init__(self, owner: Union[Browser, BasePage]):
self._owner: BasePage = ... self._owner: Union[Browser, BasePage] = ...
def NoneElement_value(self, value: Any = None, on_off: bool = True) -> None: ... 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): class ChromiumBaseSetter(BasePageSetter):
def __init__(self, owner): def __init__(self, owner):
@ -70,12 +101,12 @@ class ChromiumBaseSetter(BasePageSetter):
class TabSetter(ChromiumBaseSetter): class TabSetter(ChromiumBaseSetter):
_owner: ChromiumTab = ... _owner: ChromiumTab = ...
def __init__(self, owner: Union[ChromiumTab, WebPageTab, WebPage, ChromiumPage]): ... def __init__(self, owner: Union[ChromiumTab, MixTab, WebPage, ChromiumPage]): ...
@property @property
def window(self) -> WindowSetter: ... 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: ... 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 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: ... def timeout(self, second: float) -> None: ...
@ -152,7 +183,7 @@ class WebPageSetter(ChromiumPageSetter):
class WebPageTabSetter(TabSetter): class WebPageTabSetter(TabSetter):
_owner: WebPageTab = ... _owner: MixTab = ...
_session_setter: SessionPageSetter = ... _session_setter: SessionPageSetter = ...
_chromium_setter: ChromiumBaseSetter = ... _chromium_setter: ChromiumBaseSetter = ...
@ -168,7 +199,7 @@ class ChromiumElementSetter(object):
def __init__(self, ele: ChromiumElement): def __init__(self, ele: ChromiumElement):
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: ... def property(self, name: str, value: str) -> None: ...
@ -186,8 +217,9 @@ class ChromiumFrameSetter(ChromiumBaseSetter):
class LoadMode(object): class LoadMode(object):
def __init__(self, owner: ChromiumBase): _owner: Union[Browser, ChromiumBase] = ...
self._owner: ChromiumBase = ...
def __init__(self, owner: Union[Browser, ChromiumBase]): ...
def __call__(self, value: str) -> None: ... def __call__(self, value: str) -> None: ...

View File

@ -26,6 +26,84 @@ class OriginWaiter(object):
sleep(uniform(second, scope)) 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): class BaseWaiter(OriginWaiter):
def __init__(self, page_or_ele): def __init__(self, page_or_ele):
""" """

View File

@ -8,6 +8,7 @@
from typing import Union, Tuple, Literal, List from typing import Union, Tuple, Literal, List
from .downloader import DownloadMission from .downloader import DownloadMission
from .._base.browser import Browser
from .._elements.chromium_element import ChromiumElement from .._elements.chromium_element import ChromiumElement
from .._pages.chromium_base import ChromiumBase from .._pages.chromium_base import ChromiumBase
from .._pages.chromium_frame import ChromiumFrame from .._pages.chromium_frame import ChromiumFrame
@ -18,6 +19,17 @@ class OriginWaiter(object):
def __call__(self, second: float, scope: float = None) -> None: ... 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): class BaseWaiter(OriginWaiter):
def __init__(self, page: ChromiumBase): def __init__(self, page: ChromiumBase):
self._driver: ChromiumBase = ... self._driver: ChromiumBase = ...

View File

@ -9,7 +9,7 @@ from ._elements.chromium_element import ChromiumElement, ShadowRoot
from ._elements.none_element import NoneElement from ._elements.none_element import NoneElement
from ._elements.session_element import SessionElement from ._elements.session_element import SessionElement
from ._pages.chromium_frame import ChromiumFrame 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', __all__ = ['ChromiumElement', 'ShadowRoot', 'NoneElement', 'SessionElement', 'ChromiumFrame', 'ChromiumTab',
'WebPageTab'] 'MixTab']