mirror of
https://gitee.com/g1879/DrissionPage.git
synced 2024-12-10 04:00:23 +08:00
优化页面对象启动逻辑,未完成
This commit is contained in:
parent
716d8df5aa
commit
6cfb7dc322
@ -13,7 +13,7 @@ class ActionChains:
|
||||
"""用于实现动作链的类"""
|
||||
|
||||
def __init__(self, page):
|
||||
"""初始化 \n
|
||||
"""
|
||||
:param page: ChromiumPage对象
|
||||
"""
|
||||
self.page = page
|
||||
|
@ -9,18 +9,18 @@ from time import perf_counter, sleep
|
||||
from requests import Session
|
||||
|
||||
from .base import BasePage
|
||||
from .chromium_driver import ChromiumDriver
|
||||
from .chromium_element import ChromiumElementWaiter, ChromiumScroll, ChromiumElement, run_js, make_chromium_ele
|
||||
from .functions.locator import get_loc
|
||||
from .functions.web import offset_scroll, cookies_to_tuple
|
||||
from .session_element import make_session_ele
|
||||
from .chromium_driver import ChromiumDriver
|
||||
|
||||
|
||||
class ChromiumBase(BasePage):
|
||||
"""标签页、frame、页面基类"""
|
||||
|
||||
def __init__(self, address, tab_id=None, timeout=None):
|
||||
"""初始化 \n
|
||||
"""
|
||||
:param address: 浏览器 ip:port
|
||||
:param tab_id: 要控制的标签页id,不指定默认为激活的
|
||||
:param timeout: 超时时间
|
||||
@ -29,7 +29,14 @@ class ChromiumBase(BasePage):
|
||||
self._root_id = None
|
||||
self._debug = False
|
||||
self._debug_recorder = None
|
||||
self.timeouts = Timeout(self)
|
||||
self._control_session = Session()
|
||||
self._control_session.keep_alive = False
|
||||
self._first_run = True
|
||||
self._is_reading = False # 用于避免不同线程重复读取document
|
||||
|
||||
self._timeouts = None
|
||||
self._page_load_strategy = None
|
||||
|
||||
self._connect_browser(address, tab_id)
|
||||
timeout = timeout if timeout is not None else self.timeouts.implicit
|
||||
super().__init__(timeout)
|
||||
@ -40,21 +47,20 @@ class ChromiumBase(BasePage):
|
||||
:param tab_id: 要控制的标签页id,不指定默认为激活的
|
||||
:return: None
|
||||
"""
|
||||
self._root_id = None
|
||||
self._control_session = Session()
|
||||
self._control_session.keep_alive = False
|
||||
self._first_run = True
|
||||
self._is_reading = False # 用于避免不同线程重复读取document
|
||||
|
||||
self.address = addr_driver_opts
|
||||
if not tab_id:
|
||||
json = self._control_session.get(f'http://{self.address}/json').json()
|
||||
tab_id = [i['id'] for i in json if i['type'] == 'page'][0]
|
||||
self._set_options()
|
||||
self._init_page(tab_id)
|
||||
self._set_options()
|
||||
self._get_document()
|
||||
self._first_run = False
|
||||
|
||||
def _set_options(self):
|
||||
"""用于设置浏览器运行参数"""
|
||||
self._timeouts = Timeout(self)
|
||||
self._page_load_strategy = 'normal'
|
||||
|
||||
def _init_page(self, tab_id=None):
|
||||
"""新建页面、页面刷新、切换标签页后要进行的cdp参数初始化
|
||||
:param tab_id: 要跳转到的标签页id
|
||||
@ -62,8 +68,7 @@ class ChromiumBase(BasePage):
|
||||
"""
|
||||
self._is_loading = True
|
||||
if tab_id:
|
||||
self._tab_obj = ChromiumDriver(tab_id=tab_id, tab_type='page',
|
||||
ws_url=f'ws://{self.address}/devtools/page/{tab_id}')
|
||||
self._tab_obj = ChromiumDriver(tab_id=tab_id, tab_type='page', address=self.address)
|
||||
|
||||
self._tab_obj.start()
|
||||
self._tab_obj.DOM.enable()
|
||||
@ -174,12 +179,6 @@ class ChromiumBase(BasePage):
|
||||
if self._debug_recorder:
|
||||
self._debug_recorder.add_data((perf_counter(), '加载流程', 'navigated'))
|
||||
|
||||
def _set_options(self):
|
||||
self.set_timeouts(page_load=10,
|
||||
script=10,
|
||||
implicit=10)
|
||||
self._page_load_strategy = 'normal'
|
||||
|
||||
def __call__(self, loc_or_str, timeout=None):
|
||||
"""在内部查找元素 \n
|
||||
例:ele = page('@id=ele_id') \n
|
||||
@ -268,6 +267,11 @@ class ChromiumBase(BasePage):
|
||||
self._scroll = ChromiumScroll(self)
|
||||
return self._scroll
|
||||
|
||||
@property
|
||||
def timeouts(self):
|
||||
"""返回timeouts设置"""
|
||||
return self._timeouts
|
||||
|
||||
@property
|
||||
def set_page_load_strategy(self):
|
||||
"""返回用于设置页面加载策略的对象"""
|
||||
@ -281,13 +285,13 @@ class ChromiumBase(BasePage):
|
||||
:return: None
|
||||
"""
|
||||
if implicit is not None:
|
||||
self.timeouts.implicit = implicit
|
||||
self._timeouts.implicit = implicit
|
||||
|
||||
if page_load is not None:
|
||||
self.timeouts.page_load = page_load
|
||||
self._timeouts.page_load = page_load
|
||||
|
||||
if script is not None:
|
||||
self.timeouts.script = script
|
||||
self._timeouts.script = script
|
||||
|
||||
def run_js(self, script, as_expr=False, *args):
|
||||
"""运行javascript代码 \n
|
||||
@ -661,11 +665,17 @@ class ChromiumBase(BasePage):
|
||||
class Timeout(object):
|
||||
"""用于保存d模式timeout信息的类"""
|
||||
|
||||
def __init__(self, page):
|
||||
def __init__(self, page, implicit=None, page_load=None, script=None):
|
||||
"""
|
||||
:param page: ChromiumBase页面
|
||||
:param implicit: 默认超时时间
|
||||
:param page_load: 页面加载超时时间
|
||||
:param script: js超时时间
|
||||
"""
|
||||
self._page = page
|
||||
self.implicit = 10
|
||||
self.page_load = 30
|
||||
self.script = 30
|
||||
self.implicit = 10 if implicit is None else implicit
|
||||
self.page_load = 30 if page_load is None else page_load
|
||||
self.script = 30 if script is None else script
|
||||
|
||||
def __repr__(self):
|
||||
return str({'implicit': self.implicit, 'page_load': self.page_load, 'script': self.script})
|
||||
|
@ -10,11 +10,11 @@ from requests import Session
|
||||
from requests.cookies import RequestsCookieJar
|
||||
|
||||
from .base import BasePage
|
||||
from .chromium_driver import ChromiumDriver
|
||||
from .chromium_element import ChromiumElement, ChromiumElementWaiter, ChromiumScroll
|
||||
from .chromium_frame import ChromiumFrame
|
||||
from .config import DriverOptions
|
||||
from .configs.driver_options import DriverOptions
|
||||
from .session_element import SessionElement
|
||||
from .chromium_driver import ChromiumDriver
|
||||
|
||||
|
||||
class ChromiumBase(BasePage):
|
||||
@ -27,7 +27,7 @@ class ChromiumBase(BasePage):
|
||||
self.address: str = ...
|
||||
self._tab_obj: ChromiumDriver = ...
|
||||
self._is_reading: bool = ...
|
||||
self.timeouts: Timeout = ...
|
||||
self._timeouts: Timeout = ...
|
||||
self._first_run: bool = ...
|
||||
self._is_loading: bool = ...
|
||||
self._page_load_strategy: str = ...
|
||||
@ -104,6 +104,9 @@ class ChromiumBase(BasePage):
|
||||
@property
|
||||
def scroll(self) -> ChromiumScroll: ...
|
||||
|
||||
@property
|
||||
def timeouts(self) -> Timeout: ...
|
||||
|
||||
@property
|
||||
def set_page_load_strategy(self) -> PageLoadStrategy: ...
|
||||
|
||||
@ -192,7 +195,7 @@ class ChromiumBase(BasePage):
|
||||
|
||||
class Timeout(object):
|
||||
|
||||
def __init__(self, page: ChromiumBase):
|
||||
def __init__(self, page: ChromiumBase, implicit=None, page_load=None, script=None):
|
||||
self._page: ChromiumBase = ...
|
||||
self.implicit: float = ...
|
||||
self.page_load: float = ...
|
||||
|
@ -31,18 +31,19 @@ class ChromiumDriver(object):
|
||||
_STARTED_ = 'started'
|
||||
_STOPPED_ = 'stopped'
|
||||
|
||||
def __init__(self, tab_id, tab_type, ws_url):
|
||||
def __init__(self, tab_id, tab_type, address):
|
||||
"""
|
||||
:param tab_id: 标签页id
|
||||
:param tab_type: 标签页类型
|
||||
:param ws_url: web socket url
|
||||
:param address: 浏览器连接地址
|
||||
"""
|
||||
self.id = tab_id
|
||||
self.type = tab_type
|
||||
self.debug = False
|
||||
self.has_alert = False
|
||||
|
||||
self._websocket_url = ws_url
|
||||
self._websocket_url = f'ws://{address}/devtools/{tab_type}/{tab_id}'
|
||||
self._address = address
|
||||
self._cur_id = 0
|
||||
self._ws = None
|
||||
|
||||
@ -64,6 +65,11 @@ class ChromiumDriver(object):
|
||||
"""返回websocket连接地址"""
|
||||
return self._websocket_url
|
||||
|
||||
@property
|
||||
def address(self):
|
||||
"""返回连接地址"""
|
||||
return self._address
|
||||
|
||||
def _send(self, message, timeout=None):
|
||||
"""发送信息到浏览器,并返回浏览器返回的信息
|
||||
:param message: 发送给浏览器的数据
|
||||
|
@ -32,11 +32,14 @@ class ChromiumDriver(object):
|
||||
method_results: dict
|
||||
event_queue: Queue
|
||||
|
||||
def __init__(self, tab_id: str, tab_type: str, ws_url: str): ...
|
||||
def __init__(self, tab_id: str, tab_type: str, address: str): ...
|
||||
|
||||
@property
|
||||
def websocket_url(self) -> str: ...
|
||||
|
||||
@property
|
||||
def address(self) -> str: ...
|
||||
|
||||
def _send(self, message: dict, timeout: Union[int, float] = None) -> dict: ...
|
||||
|
||||
def _recv_loop(self) -> None: ...
|
||||
|
@ -1477,7 +1477,7 @@ class ChromiumSelect(object):
|
||||
"""ChromiumSelect 类专门用于处理 d 模式下 select 标签"""
|
||||
|
||||
def __init__(self, ele):
|
||||
"""初始化 \n
|
||||
"""
|
||||
:param ele: select 元素对象
|
||||
"""
|
||||
if ele.tag != 'select':
|
||||
|
@ -43,6 +43,11 @@ class ChromiumFrame(ChromiumBase):
|
||||
attrs = [f"{attr}='{attrs[attr]}'" for attr in attrs]
|
||||
return f'<ChromiumFrame {self.frame_ele.tag} {" ".join(attrs)}>'
|
||||
|
||||
def _set_options(self):
|
||||
"""重写设置浏览器运行参数方法"""
|
||||
self._timeouts = self.page.timeouts
|
||||
self._page_load_strategy = self.page.page_load_strategy
|
||||
|
||||
def _reload(self):
|
||||
self._frame_ele = ChromiumElement(self.page, backend_id=self._backend_id)
|
||||
node = self.page.run_cdp('DOM.describeNode', nodeId=self._frame_ele.node_id, not_change=True)['node']
|
||||
|
@ -28,6 +28,8 @@ class ChromiumFrame(ChromiumBase):
|
||||
|
||||
def __repr__(self) -> str: ...
|
||||
|
||||
def _set_options(self) -> None: ...
|
||||
|
||||
def _reload(self) -> None: ...
|
||||
|
||||
def _check_ok(self) -> None: ...
|
||||
|
@ -5,16 +5,14 @@
|
||||
"""
|
||||
from pathlib import Path
|
||||
from platform import system
|
||||
from re import search
|
||||
from time import perf_counter, sleep
|
||||
|
||||
from requests import Session
|
||||
|
||||
from .chromium_base import Timeout, ChromiumBase
|
||||
from .chromium_base import ChromiumBase, Timeout
|
||||
from .chromium_driver import ChromiumDriver
|
||||
from .chromium_tab import ChromiumTab
|
||||
from .functions.browser import connect_browser
|
||||
from .configs.chromium_options import ChromiumOptions
|
||||
from .configs.driver_options import DriverOptions
|
||||
from .functions.browser import connect_browser
|
||||
from .session_page import DownloadSetter
|
||||
|
||||
|
||||
@ -23,7 +21,7 @@ class ChromiumPage(ChromiumBase):
|
||||
|
||||
def __init__(self, addr_driver_opts=None, tab_id=None, timeout=None):
|
||||
"""
|
||||
:param addr_driver_opts: 浏览器地址:端口、ChromiumDriver对象或DriverOptions对象
|
||||
:param addr_driver_opts: 浏览器地址:端口、ChromiumDriver对象或ChromiumOptions对象
|
||||
:param tab_id: 要控制的标签页id,不指定默认为激活的
|
||||
:param timeout: 超时时间
|
||||
"""
|
||||
@ -35,29 +33,20 @@ class ChromiumPage(ChromiumBase):
|
||||
:param tab_id: 要控制的标签页id,不指定默认为激活的
|
||||
:return: None
|
||||
"""
|
||||
self._is_loading = False
|
||||
self._is_reading = False
|
||||
self._root_id = None
|
||||
self.timeouts = Timeout(self)
|
||||
self._control_session = Session()
|
||||
self._control_session.keep_alive = False
|
||||
self._alert = Alert()
|
||||
self._first_run = True
|
||||
|
||||
# 接管或启动浏览器
|
||||
if addr_driver_opts is None or isinstance(addr_driver_opts, DriverOptions):
|
||||
self.options = addr_driver_opts or DriverOptions() # 从ini文件读取
|
||||
self.address = self.options.debugger_address
|
||||
self.process = connect_browser(self.options)[1]
|
||||
if addr_driver_opts is None or isinstance(addr_driver_opts, (ChromiumOptions, DriverOptions)):
|
||||
self._driver_options = addr_driver_opts or ChromiumOptions() # 从ini文件读取
|
||||
self.address = self._driver_options.debugger_address
|
||||
self.process = connect_browser(self._driver_options)[1]
|
||||
json = self._control_session.get(f'http://{self.address}/json').json()
|
||||
tab_id = [i['id'] for i in json if i['type'] == 'page'][0]
|
||||
|
||||
# 接收浏览器地址和端口
|
||||
elif isinstance(addr_driver_opts, str):
|
||||
self.address = addr_driver_opts
|
||||
self.options = DriverOptions(read_file=False)
|
||||
self.options.debugger_address = addr_driver_opts
|
||||
self.process = connect_browser(self.options)[1]
|
||||
self._driver_options = ChromiumOptions(read_file=False)
|
||||
self._driver_options.debugger_address = addr_driver_opts
|
||||
self.process = connect_browser(self._driver_options)[1]
|
||||
if not tab_id:
|
||||
json = self._control_session.get(f'http://{self.address}/json').json()
|
||||
tab_id = [i['id'] for i in json if i['type'] == 'page'][0]
|
||||
@ -65,18 +54,30 @@ class ChromiumPage(ChromiumBase):
|
||||
# 接收传递过来的ChromiumDriver,浏览器
|
||||
elif isinstance(addr_driver_opts, ChromiumDriver):
|
||||
self._tab_obj = addr_driver_opts
|
||||
self.address = search(r'ws://(.*?)/dev', addr_driver_opts.websocket_url).group(1)
|
||||
self.address = addr_driver_opts.address
|
||||
self.process = None
|
||||
self.options = DriverOptions(read_file=False)
|
||||
self._driver_options = ChromiumOptions(read_file=False)
|
||||
self._driver_options.debugger_address = addr_driver_opts.address
|
||||
|
||||
else:
|
||||
raise TypeError('只能接收ChromiumDriver或DriverOptions类型参数。')
|
||||
raise TypeError('只能接收ChromiumDriver或ChromiumOptions类型参数。')
|
||||
|
||||
self._set_options()
|
||||
self._init_page(tab_id)
|
||||
self._set_options()
|
||||
self._get_document()
|
||||
self._first_run = False
|
||||
|
||||
def _set_options(self):
|
||||
"""从配置中读取设置"""
|
||||
self._timeouts = Timeout(self,
|
||||
page_load=self._driver_options.timeouts['pageLoad'],
|
||||
script=self._driver_options.timeouts['script'],
|
||||
implicit=self._driver_options.timeouts['implicit'])
|
||||
self._page_load_strategy = self._driver_options.page_load_strategy
|
||||
self._main_tab = self.tab_id
|
||||
self._alert = Alert()
|
||||
self._window_setter = None
|
||||
self._download_path = self._driver_options.download_path
|
||||
|
||||
def _init_page(self, tab_id=None):
|
||||
"""新建页面、页面刷新、切换标签页后要进行的cdp参数初始化
|
||||
@ -84,17 +85,8 @@ class ChromiumPage(ChromiumBase):
|
||||
:return: None
|
||||
"""
|
||||
super()._init_page(tab_id)
|
||||
|
||||
self._tab_obj.Page.javascriptDialogOpening = self._on_alert_open
|
||||
self._tab_obj.Page.javascriptDialogClosed = self._on_alert_close
|
||||
self.download_set.save_path(self.options.download_path)
|
||||
|
||||
def _set_options(self):
|
||||
"""从配置中读取设置"""
|
||||
self.set_timeouts(page_load=self.options.timeouts['pageLoad'],
|
||||
script=self.options.timeouts['script'],
|
||||
implicit=self.options.timeouts['implicit'])
|
||||
self._page_load_strategy = self.options.page_load_strategy
|
||||
|
||||
@property
|
||||
def tabs_count(self):
|
||||
@ -122,7 +114,7 @@ class ChromiumPage(ChromiumBase):
|
||||
@property
|
||||
def set_window(self):
|
||||
"""返回用于设置窗口大小的对象"""
|
||||
if not hasattr(self, '_window_setter'):
|
||||
if self._window_setter is None:
|
||||
self._window_setter = WindowSetter(self)
|
||||
return self._window_setter
|
||||
|
||||
@ -403,7 +395,7 @@ class WindowSetter(object):
|
||||
"""用于设置窗口大小的类"""
|
||||
|
||||
def __init__(self, page):
|
||||
self.driver = page._driver
|
||||
self.driver = page.driver
|
||||
self.window_id = self._get_info()['windowId']
|
||||
|
||||
def maximized(self):
|
||||
|
@ -10,7 +10,7 @@ from typing import Union, Tuple, List
|
||||
from .chromium_base import ChromiumBase
|
||||
from .chromium_driver import ChromiumDriver
|
||||
from .chromium_tab import ChromiumTab
|
||||
from .config import DriverOptions
|
||||
from .configs.driver_options import DriverOptions
|
||||
|
||||
|
||||
class ChromiumPage(ChromiumBase):
|
||||
@ -19,7 +19,7 @@ class ChromiumPage(ChromiumBase):
|
||||
addr_driver_opts: Union[str, ChromiumDriver, DriverOptions] = None,
|
||||
tab_id: str = None,
|
||||
timeout: float = None):
|
||||
self.options: DriverOptions = ...
|
||||
self._driver_options: [ChromiumDriver, DriverOptions] = ...
|
||||
self.process: popen = ...
|
||||
self._window_setter: WindowSetter = ...
|
||||
self._main_tab: str = ...
|
||||
|
@ -10,7 +10,7 @@ class ChromiumTab(ChromiumBase):
|
||||
"""实现浏览器标签页的类"""
|
||||
|
||||
def __init__(self, page, tab_id=None):
|
||||
"""初始化 \n
|
||||
"""
|
||||
:param page: ChromiumPage对象
|
||||
:param tab_id: 要控制的标签页id,不指定默认为激活的
|
||||
"""
|
||||
@ -18,7 +18,6 @@ class ChromiumTab(ChromiumBase):
|
||||
super().__init__(page.address, tab_id, page.timeout)
|
||||
|
||||
def _set_options(self):
|
||||
self.set_timeouts(page_load=self.page.timeouts.page_load,
|
||||
script=self.page.timeouts.script,
|
||||
implicit=self.page.timeouts.implicit)
|
||||
"""重写设置浏览器运行参数方法"""
|
||||
self._timeouts = self.page.timeouts
|
||||
self._page_load_strategy = self.page.page_load_strategy
|
||||
|
@ -14,6 +14,7 @@ class ChromiumOptions(object):
|
||||
"""
|
||||
self._user_data_path = None
|
||||
self._user = 'Default'
|
||||
self._prefs_to_del = []
|
||||
|
||||
if read_file:
|
||||
self.ini_path = ini_path or str(Path(__file__).parent / 'configs.ini')
|
||||
@ -52,7 +53,6 @@ class ChromiumOptions(object):
|
||||
self._download_path = None
|
||||
self._extensions = []
|
||||
self._prefs = {}
|
||||
self._prefs_to_del = []
|
||||
self._timeouts = {'implicit': 10, 'pageLoad': 30, 'script': 30}
|
||||
self._debugger_address = '127.0.0.1:9222'
|
||||
self._page_load_strategy = 'normal'
|
||||
@ -98,6 +98,15 @@ class ChromiumOptions(object):
|
||||
"""返回浏览器地址,ip:port"""
|
||||
return self._debugger_address
|
||||
|
||||
@debugger_address.setter
|
||||
def debugger_address(self, address):
|
||||
self._debugger_address = address
|
||||
|
||||
@property
|
||||
def arguments(self):
|
||||
"""返回浏览器命令行设置列表"""
|
||||
return self._arguments
|
||||
|
||||
@property
|
||||
def extensions(self):
|
||||
"""以list形式返回要加载的插件路径"""
|
||||
|
@ -44,6 +44,12 @@ class ChromiumOptions(object):
|
||||
@property
|
||||
def debugger_address(self) -> str: ...
|
||||
|
||||
@property
|
||||
def arguments(self) -> list: ...
|
||||
|
||||
@debugger_address.setter
|
||||
def debugger_address(self, address: str): ...
|
||||
|
||||
@property
|
||||
def extensions(self) -> list: ...
|
||||
|
||||
|
@ -12,7 +12,8 @@ from selenium.webdriver.chrome.options import Options
|
||||
from selenium.webdriver.chrome.webdriver import WebDriver
|
||||
from selenium.webdriver.remote.webdriver import WebDriver as RemoteWebDriver
|
||||
|
||||
from .config import SessionOptions, DriverOptions
|
||||
from .configs.driver_options import DriverOptions
|
||||
from .configs.session_options import SessionOptions
|
||||
|
||||
|
||||
class Drission(object):
|
||||
|
@ -928,7 +928,7 @@ class Select(object):
|
||||
"""Select 类专门用于处理 d 模式下 select 标签"""
|
||||
|
||||
def __init__(self, ele: DriverElement):
|
||||
"""初始化 \n
|
||||
"""
|
||||
:param ele: select 元素对象
|
||||
"""
|
||||
if ele.tag != 'select':
|
||||
|
@ -11,7 +11,7 @@ from typing import Union
|
||||
|
||||
from selenium import webdriver
|
||||
|
||||
from .common import unzip
|
||||
from .functions.tools import unzip
|
||||
from .configs.options_manage import OptionsManager
|
||||
from .configs.driver_options import DriverOptions
|
||||
from .drission import Drission
|
||||
|
@ -1,4 +1,5 @@
|
||||
# -*- coding:utf-8 -*-
|
||||
from json import load, dump
|
||||
from pathlib import Path
|
||||
from platform import system
|
||||
from subprocess import Popen
|
||||
@ -74,9 +75,16 @@ def set_prefs(opt):
|
||||
:param opt: DriverOptions或ChromiumOptions
|
||||
:return: None
|
||||
"""
|
||||
# todo: 支持删除pref项
|
||||
prefs = opt.experimental_options.get('prefs', None) if isinstance(opt, DriverOptions) else opt.preferences
|
||||
if prefs and opt.user_data_path:
|
||||
if isinstance(opt, DriverOptions):
|
||||
prefs = opt.experimental_options.get('prefs', None)
|
||||
del_list = []
|
||||
else:
|
||||
prefs = opt.preferences
|
||||
del_list = opt._prefs_to_del
|
||||
|
||||
if not opt.user_data_path:
|
||||
return
|
||||
|
||||
args = opt.arguments
|
||||
user = 'Default'
|
||||
for arg in args:
|
||||
@ -85,23 +93,26 @@ def set_prefs(opt):
|
||||
break
|
||||
|
||||
prefs_file = Path(opt.user_data_path) / user / 'Preferences'
|
||||
|
||||
if not prefs_file.exists():
|
||||
prefs_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(prefs_file, 'w') as f:
|
||||
f.write('{}')
|
||||
|
||||
from json import load, dump
|
||||
with open(prefs_file, "r", encoding='utf-8') as f:
|
||||
j = load(f)
|
||||
prefs_dict = load(f)
|
||||
|
||||
for pref in prefs:
|
||||
value = prefs[pref]
|
||||
pref = pref.split('.')
|
||||
_make_leave_in_dict(j, pref, 0, len(pref))
|
||||
_set_value_to_dict(j, pref, value)
|
||||
_make_leave_in_dict(prefs_dict, pref, 0, len(pref))
|
||||
_set_value_to_dict(prefs_dict, pref, value)
|
||||
|
||||
for pref in del_list:
|
||||
_remove_arg_from_dict(prefs_dict, pref)
|
||||
|
||||
with open(prefs_file, 'w', encoding='utf-8') as f:
|
||||
dump(j, f)
|
||||
dump(prefs_dict, f)
|
||||
|
||||
|
||||
def _run_browser(port, path: str, args) -> Popen:
|
||||
@ -130,7 +141,7 @@ def _run_browser(port, path: str, args) -> Popen:
|
||||
|
||||
def _make_leave_in_dict(target_dict: dict, src: list, num: int, end: int) -> None:
|
||||
"""把prefs中a.b.c形式的属性转为a['b']['c']形式
|
||||
:param target_dict: 要处理的dict
|
||||
:param target_dict: 要处理的字典
|
||||
:param src: 属性层级列表[a, b, c]
|
||||
:param num: 当前处理第几个
|
||||
:param end: src长度
|
||||
@ -146,7 +157,7 @@ def _make_leave_in_dict(target_dict: dict, src: list, num: int, end: int) -> Non
|
||||
|
||||
def _set_value_to_dict(target_dict: dict, src: list, value) -> None:
|
||||
"""把a.b.c形式的属性的值赋值到a['b']['c']形式的字典中
|
||||
:param target_dict: 要处理的dict
|
||||
:param target_dict: 要处理的字典
|
||||
:param src: 属性层级列表[a, b, c]
|
||||
:param value: 属性值
|
||||
:return: None
|
||||
@ -154,3 +165,22 @@ def _set_value_to_dict(target_dict: dict, src: list, value) -> None:
|
||||
src = "']['".join(src)
|
||||
src = f"target_dict['{src}']=value"
|
||||
exec(src)
|
||||
|
||||
|
||||
def _remove_arg_from_dict(target_dict: dict, arg: str) -> None:
|
||||
"""把a.b.c形式的属性从字典中删除
|
||||
:param target_dict: 要处理的字典
|
||||
:param arg: 层级属性,形式'a.b.c'
|
||||
:return: None
|
||||
"""
|
||||
args = arg.split('.')
|
||||
args = [f"['{i}']" for i in args]
|
||||
src = ''.join(args)
|
||||
src = f"target_dict{src}"
|
||||
try:
|
||||
exec(src)
|
||||
src = ''.join(args[:-1])
|
||||
src = f"target_dict{src}.pop({args[-1][1:-1]})"
|
||||
exec(src)
|
||||
except:
|
||||
pass
|
||||
|
@ -13,7 +13,8 @@ from selenium.webdriver.chrome.webdriver import WebDriver
|
||||
from selenium.webdriver.remote.webelement import WebElement
|
||||
|
||||
from .base import BasePage
|
||||
from .config import DriverOptions, SessionOptions
|
||||
from .configs.session_options import SessionOptions
|
||||
from .configs.driver_options import DriverOptions
|
||||
from .drission import Drission
|
||||
from .driver_element import DriverElement
|
||||
from .driver_page import DriverPage
|
||||
|
@ -22,11 +22,12 @@ class SessionPage(BasePage):
|
||||
"""SessionPage封装了页面操作的常用功能,使用requests来获取、解析网页"""
|
||||
|
||||
def __init__(self, session_or_options=None, timeout=None):
|
||||
"""初始化 \n
|
||||
"""
|
||||
:param session_or_options: Session对象或SessionOptions对象
|
||||
:param timeout: 连接超时时间,为None时从ini文件读取
|
||||
"""
|
||||
self._response = None
|
||||
self._download_kit = None
|
||||
self._create_session(session_or_options)
|
||||
timeout = timeout if timeout is not None else self.timeout
|
||||
super().__init__(timeout)
|
||||
@ -39,12 +40,13 @@ class SessionPage(BasePage):
|
||||
if Session_or_Options is None or isinstance(Session_or_Options, SessionOptions):
|
||||
options = Session_or_Options or SessionOptions()
|
||||
self._set_session(options)
|
||||
self.timeout = options.timeout
|
||||
self._timeout = options.timeout
|
||||
self._download_path = options.download_path
|
||||
|
||||
elif isinstance(Session_or_Options, Session):
|
||||
self._session = Session_or_Options
|
||||
self._timeout = 10
|
||||
self._download_path = None
|
||||
self._download_kit = None
|
||||
|
||||
def _set_session(self, opt):
|
||||
"""根据传入字典对session进行设置 \n
|
||||
|
@ -12,7 +12,7 @@ from requests.cookies import RequestsCookieJar
|
||||
from requests.structures import CaseInsensitiveDict
|
||||
|
||||
from .base import BasePage
|
||||
from .config import SessionOptions
|
||||
from .configs.session_options import SessionOptions
|
||||
from .session_element import SessionElement
|
||||
|
||||
|
||||
|
@ -13,8 +13,9 @@ from .base import BasePage
|
||||
from .chromium_base import ChromiumBase, Timeout
|
||||
from .chromium_driver import ChromiumDriver
|
||||
from .chromium_page import ChromiumPage, ChromiumDownloadSetter
|
||||
from .configs.session_options import SessionOptions
|
||||
from .configs.chromium_options import ChromiumOptions
|
||||
from .configs.driver_options import DriverOptions
|
||||
from .configs.session_options import SessionOptions
|
||||
from .functions.web import cookies_to_tuple
|
||||
from .session_page import SessionPage
|
||||
|
||||
@ -33,14 +34,18 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
self._mode = mode.lower()
|
||||
if self._mode not in ('s', 'd'):
|
||||
raise ValueError('mode参数只能是s或d。')
|
||||
self._has_driver, self._has_session = (None, True) if self._mode == 's' else (True, None)
|
||||
|
||||
self._debug = False
|
||||
self._debug_recorder = None
|
||||
|
||||
self._session = None
|
||||
self._tab_obj = None
|
||||
self._is_loading = False
|
||||
self.timeouts = Timeout(self)
|
||||
self._has_driver, self._has_session = (None, True) if self._mode == 's' else (True, None)
|
||||
|
||||
self._driver_options = None
|
||||
self._session_options = None
|
||||
|
||||
self._set_both_options(driver_or_options, session_or_options)
|
||||
self._setting_tab_id = tab_id
|
||||
self._response = None
|
||||
@ -53,6 +58,58 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
t = timeout if timeout is not None else self.timeouts.implicit
|
||||
super(ChromiumBase, self).__init__(t) # 调用Base的__init__()
|
||||
|
||||
def _set_both_options(self, dr_opt, se_opt):
|
||||
"""处理两种模式的设置
|
||||
:param dr_opt: ChromiumDriver或DriverOptions对象,为None则从ini读取,为False用默认信息创建
|
||||
:param se_opt: Session、SessionOptions对象或配置信息,为None则从ini读取,为False用默认信息创建
|
||||
:return: None
|
||||
"""
|
||||
if isinstance(dr_opt, ChromiumDriver):
|
||||
self._connect_browser(dr_opt)
|
||||
self._has_driver = True
|
||||
self._driver_options = None
|
||||
dr_opt = False
|
||||
|
||||
else:
|
||||
if dr_opt is None:
|
||||
self._driver_options = ChromiumOptions()
|
||||
|
||||
elif dr_opt is False:
|
||||
self._driver_options = ChromiumOptions(read_file=False)
|
||||
|
||||
elif isinstance(dr_opt, (ChromiumOptions, DriverOptions)):
|
||||
self._driver_options = dr_opt
|
||||
|
||||
else:
|
||||
raise TypeError('driver_or_options参数只能接收ChromiumDriver, ChromiumOptionsOptions、None或False。')
|
||||
|
||||
if isinstance(se_opt, Session):
|
||||
self._session = se_opt
|
||||
self._has_session = True
|
||||
self._session_options = None
|
||||
se_opt = False
|
||||
|
||||
else:
|
||||
if se_opt is None:
|
||||
self._session_options = SessionOptions()
|
||||
|
||||
elif se_opt is False:
|
||||
self._session_options = SessionOptions(read_file=False)
|
||||
|
||||
elif isinstance(se_opt, SessionOptions):
|
||||
self._session_options = se_opt
|
||||
|
||||
else:
|
||||
raise TypeError('session_or_options参数只能接收Session, SessionOptions、None或False。')
|
||||
|
||||
self._timeouts = Timeout(self)
|
||||
if se_opt is not False:
|
||||
self.set_timeouts(implicit=self._session_options.timeout)
|
||||
|
||||
if dr_opt is not False:
|
||||
t = self._driver_options.timeouts
|
||||
self.set_timeouts(t['implicit'], t['pageLoad'], t['script'])
|
||||
|
||||
def __call__(self, loc_or_str, timeout=None):
|
||||
"""在内部查找元素 \n
|
||||
例:ele = page('@id=ele_id') \n
|
||||
@ -399,59 +456,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
elif self._mode == 'd':
|
||||
return super(SessionPage, self)._ele(loc_or_ele, timeout=timeout, single=single, relative=relative)
|
||||
|
||||
def _set_both_options(self, dr_opt, se_opt):
|
||||
"""处理两种模式的设置
|
||||
:param dr_opt: ChromiumDriver或DriverOptions对象,为None则从ini读取,为False用默认信息创建
|
||||
:param se_opt: Session、SessionOptions对象或配置信息,为None则从ini读取,为False用默认信息创建
|
||||
:return: None
|
||||
"""
|
||||
if isinstance(dr_opt, ChromiumDriver):
|
||||
self._connect_browser(dr_opt)
|
||||
self._has_driver = True
|
||||
self._driver_options = None
|
||||
dr_opt = False
|
||||
|
||||
else:
|
||||
if dr_opt is None:
|
||||
self._driver_options = DriverOptions()
|
||||
|
||||
elif dr_opt is False:
|
||||
self._driver_options = DriverOptions(read_file=False)
|
||||
|
||||
elif isinstance(dr_opt, DriverOptions):
|
||||
self._driver_options = dr_opt
|
||||
|
||||
else:
|
||||
raise TypeError('driver_or_options参数只能接收ChromiumDriver, DriverOptions、None或False。')
|
||||
|
||||
if isinstance(se_opt, Session):
|
||||
self._session = se_opt
|
||||
self._has_session = True
|
||||
self._session_options = None
|
||||
se_opt = False
|
||||
|
||||
else:
|
||||
if se_opt is None:
|
||||
so = SessionOptions()
|
||||
|
||||
elif se_opt is False:
|
||||
so = SessionOptions(read_file=False)
|
||||
|
||||
elif isinstance(se_opt, SessionOptions):
|
||||
so = se_opt
|
||||
|
||||
else:
|
||||
raise TypeError('session_or_options参数只能接收Session, dict, SessionOptions、None或False。')
|
||||
|
||||
self._session_options = so
|
||||
|
||||
if se_opt is not False:
|
||||
self.set_timeouts(implicit=self._session_options.timeout)
|
||||
|
||||
if dr_opt is not False:
|
||||
t = self._driver_options.timeouts
|
||||
self.set_timeouts(t['implicit'], t['pageLoad'], t['script'])
|
||||
|
||||
def quit(self):
|
||||
"""关闭浏览器,关闭session"""
|
||||
if self._has_session:
|
||||
|
@ -13,7 +13,9 @@ from .chromium_driver import ChromiumDriver
|
||||
from .chromium_element import ChromiumElement
|
||||
from .chromium_frame import ChromiumFrame
|
||||
from .chromium_page import ChromiumPage, ChromiumDownloadSetter
|
||||
from .config import DriverOptions, SessionOptions
|
||||
from .configs.chromium_options import ChromiumOptions
|
||||
from .configs.driver_options import DriverOptions
|
||||
from .configs.session_options import SessionOptions
|
||||
from .session_element import SessionElement
|
||||
from .session_page import SessionPage
|
||||
|
||||
@ -24,7 +26,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
mode: str = 'd',
|
||||
timeout: float = None,
|
||||
tab_id: str = None,
|
||||
driver_or_options: Union[ChromiumDriver, DriverOptions, bool] = None,
|
||||
driver_or_options: Union[ChromiumDriver, ChromiumOptions, DriverOptions, bool] = None,
|
||||
session_or_options: Union[Session, SessionOptions, bool] = None) -> None:
|
||||
self._mode: str = ...
|
||||
self._has_driver: bool = ...
|
||||
|
Loading…
x
Reference in New Issue
Block a user