mirror of
https://gitee.com/g1879/DrissionPage.git
synced 2024-12-10 04:00:23 +08:00
download和download_path属性移到BasePage；ini文件删除chromedriver_path
This commit is contained in:
g1879
parent
a0c5072b12
commit
53aa546231
@ -4,9 +4,12 @@
|
||||
@Contact : g1879@qq.com
|
||||
"""
|
||||
from abc import abstractmethod
|
||||
from pathlib import Path
|
||||
from re import sub
|
||||
from urllib.parse import quote
|
||||
|
||||
from DownloadKit import DownloadKit
|
||||
|
||||
from .commons.constants import Settings, NoneElement
|
||||
from .commons.locator import get_loc
|
||||
from .commons.web import format_html
|
||||
@ -353,6 +356,8 @@ class BasePage(BaseParser):
|
||||
self.retry_times = 3
|
||||
self.retry_interval = 2
|
||||
self._url_available = None
|
||||
self._download_path = ''
|
||||
self._DownloadKit = None
|
||||
|
||||
@property
|
||||
def title(self):
|
||||
@ -380,6 +385,18 @@ class BasePage(BaseParser):
|
||||
"""返回当前访问的url有效性"""
|
||||
return self._url_available
|
||||
|
||||
@property
|
||||
def download_path(self):
|
||||
"""返回默认下载路径"""
|
||||
return str(Path(self._download_path).absolute())
|
||||
|
||||
@property
|
||||
def download(self):
|
||||
"""返回下载器对象"""
|
||||
if self._DownloadKit is None:
|
||||
self._DownloadKit = DownloadKit(session=self, goal_path=self.download_path)
|
||||
return self._DownloadKit
|
||||
|
||||
def _before_connect(self, url, retry, interval):
|
||||
"""连接前的准备
|
||||
:param url: 要访问的url
|
||||
|
||||
@ -6,6 +6,8 @@
|
||||
from abc import abstractmethod
|
||||
from typing import Union, Tuple, List
|
||||
|
||||
from DownloadKit import DownloadKit
|
||||
|
||||
from .commons.constants import NoneElement
|
||||
|
||||
|
||||
@ -154,7 +156,9 @@ class BasePage(BaseParser):
|
||||
self._url_available: bool = ...
|
||||
self.retry_times: int = ...
|
||||
self.retry_interval: float = ...
|
||||
self._timeout = float = ...
|
||||
self._timeout: float = ...
|
||||
self._download_path: str = ...
|
||||
self._DownloadKit: DownloadKit = ...
|
||||
|
||||
@property
|
||||
def title(self) -> Union[str, None]: ...
|
||||
@ -171,6 +175,12 @@ class BasePage(BaseParser):
|
||||
@property
|
||||
def url_available(self) -> bool: ...
|
||||
|
||||
@property
|
||||
def download_path(self) -> str: ...
|
||||
|
||||
@property
|
||||
def download(self) -> DownloadKit: ...
|
||||
|
||||
def _before_connect(self, url: str, retry: int, interval: float) -> tuple: ...
|
||||
|
||||
# ----------------以下属性或方法由后代实现----------------
|
||||
|
||||
@ -10,7 +10,6 @@ from pathlib import Path
|
||||
from threading import Thread
|
||||
from time import perf_counter, sleep, time
|
||||
|
||||
from DownloadKit import DownloadKit
|
||||
from requests import Session
|
||||
|
||||
from .base import BasePage
|
||||
@ -43,8 +42,6 @@ class ChromiumBase(BasePage):
|
||||
self._set = None
|
||||
self._screencast = None
|
||||
self._listener = None
|
||||
self._DownloadKit = None
|
||||
self._download_path = None
|
||||
|
||||
if isinstance(address, int) or (isinstance(address, str) and address.isdigit()):
|
||||
address = f'127.0.0.1:{address}'
|
||||
@ -386,19 +383,6 @@ class ChromiumBase(BasePage):
|
||||
self._listener = NetworkListener(self)
|
||||
return self._listener
|
||||
|
||||
@property
|
||||
def download_path(self):
|
||||
"""返回默认下载路径"""
|
||||
p = self._download_path or ''
|
||||
return str(Path(p).absolute())
|
||||
|
||||
@property
|
||||
def download(self):
|
||||
"""返回下载器对象"""
|
||||
if self._DownloadKit is None:
|
||||
self._DownloadKit = DownloadKit(session=self, goal_path=self.download_path)
|
||||
return self._DownloadKit
|
||||
|
||||
def run_cdp(self, cmd, **cmd_args):
|
||||
"""执行Chrome DevTools Protocol语句
|
||||
:param cmd: 协议项目
|
||||
|
||||
@ -7,7 +7,6 @@ from pathlib import Path
|
||||
from typing import Union, Tuple, List, Any
|
||||
|
||||
from DataRecorder import Recorder
|
||||
from DownloadKit import DownloadKit
|
||||
from requests import Session
|
||||
from requests.cookies import RequestsCookieJar
|
||||
|
||||
@ -21,7 +20,6 @@ from .session_element import SessionElement
|
||||
|
||||
|
||||
class ChromiumBase(BasePage):
|
||||
|
||||
def __init__(self,
|
||||
address: Union[str, int],
|
||||
tab_id: str = None,
|
||||
@ -44,8 +42,6 @@ class ChromiumBase(BasePage):
|
||||
self._set: ChromiumBaseSetter = ...
|
||||
self._screencast: Screencast = ...
|
||||
self._listener: NetworkListener = ...
|
||||
self._download_path: str = ...
|
||||
self._DownloadKit: DownloadKit = ...
|
||||
|
||||
def _connect_browser(self, tab_id: str = None) -> None: ...
|
||||
|
||||
@ -139,12 +135,6 @@ class ChromiumBase(BasePage):
|
||||
@property
|
||||
def listener(self) -> NetworkListener: ...
|
||||
|
||||
@property
|
||||
def download_path(self) -> str: ...
|
||||
|
||||
@property
|
||||
def download(self) -> DownloadKit: ...
|
||||
|
||||
def run_js(self, script: str, *args: Any, as_expr: bool = False) -> Any: ...
|
||||
|
||||
def run_js_loaded(self, script: str, *args: Any, as_expr: bool = False) -> Any: ...
|
||||
|
||||
@ -24,8 +24,6 @@ class ChromiumPage(ChromiumBase):
|
||||
:param tab_id: 要控制的标签页id，不指定默认为激活的
|
||||
:param timeout: 超时时间
|
||||
"""
|
||||
# self._download_set = None
|
||||
# self._download_path = None
|
||||
super().__init__(addr_driver_opts, tab_id, timeout)
|
||||
|
||||
def _set_start_options(self, addr_driver_opts, none):
|
||||
|
||||
@ -52,11 +52,12 @@ class WebPageTab(SessionPage, ChromiumTab):
|
||||
self._has_driver = True
|
||||
self._has_session = True
|
||||
self._session = copy(page.session)
|
||||
|
||||
self._response = None
|
||||
self._download_set = None
|
||||
self._download_path = None
|
||||
self._set = None
|
||||
|
||||
self._download_set = None
|
||||
self._download_path = page.download_path
|
||||
self._DownloadKit = None
|
||||
super(SessionPage, self)._set_runtime_settings()
|
||||
self._connect_browser(tab_id)
|
||||
|
||||
@ -164,11 +165,6 @@ class WebPageTab(SessionPage, ChromiumTab):
|
||||
self._set = WebPageTabSetter(self)
|
||||
return self._set
|
||||
|
||||
@property
|
||||
def download(self):
|
||||
"""返回下载器对象"""
|
||||
return self.download_set._switched_DownloadKit
|
||||
|
||||
def get(self, url, show_errmsg=False, retry=None, interval=None, timeout=None, **kwargs):
|
||||
"""跳转到一个url
|
||||
:param url: 目标url
|
||||
@ -297,17 +293,12 @@ class WebPageTab(SessionPage, ChromiumTab):
|
||||
selenium_user_agent = self.run_cdp('Runtime.evaluate', expression='navigator.userAgent;')['result']['value']
|
||||
self.session.headers.update({"User-Agent": selenium_user_agent})
|
||||
|
||||
# set_session_cookies(self.session, self._get_driver_cookies(as_dict=True))
|
||||
# set_session_cookies(self.session, self._get_driver_cookies(all_domains=True))
|
||||
set_session_cookies(self.session, super(SessionPage, self).get_cookies())
|
||||
|
||||
def cookies_to_browser(self):
|
||||
"""把session对象的cookies复制到浏览器"""
|
||||
if not self._has_driver:
|
||||
return
|
||||
|
||||
# set_browser_cookies(self, super().get_cookies(as_dict=True))
|
||||
# set_browser_cookies(self, super().get_cookies(all_domains=True))
|
||||
set_browser_cookies(self, super().get_cookies())
|
||||
|
||||
def get_cookies(self, as_dict=False, all_domains=False, all_info=False):
|
||||
@ -320,24 +311,8 @@ class WebPageTab(SessionPage, ChromiumTab):
|
||||
if self._mode == 's':
|
||||
return super().get_cookies(as_dict, all_domains, all_info)
|
||||
elif self._mode == 'd':
|
||||
# return self._get_driver_cookies(as_dict, all_info)
|
||||
return super(SessionPage, self).get_cookies(as_dict, all_domains, all_info)
|
||||
|
||||
# def _get_driver_cookies(self, as_dict=False, all_info=False):
|
||||
# """获取浏览器cookies
|
||||
# :param as_dict: 是否以dict形式返回，为True时all_info无效
|
||||
# :param all_info: 是否返回所有信息，为False时只返回name、value、domain
|
||||
# :return: cookies信息
|
||||
# """
|
||||
# cookies = self.run_cdp('Network.getCookies')['cookies']
|
||||
# if as_dict:
|
||||
# return {cookie['name']: cookie['value'] for cookie in cookies}
|
||||
# elif all_info:
|
||||
# return cookies
|
||||
# else:
|
||||
# return [{'name': cookie['name'], 'value': cookie['value'], 'domain': cookie['domain']}
|
||||
# for cookie in cookies]
|
||||
|
||||
def _find_elements(self, loc_or_ele, timeout=None, single=True, relative=False, raise_err=None):
|
||||
"""返回页面中符合条件的元素、属性或节点文本，默认返回第一个
|
||||
:param loc_or_ele: 元素的定位信息，可以是元素对象，loc元组，或查询字符串
|
||||
|
||||
@ -5,7 +5,6 @@
|
||||
"""
|
||||
from typing import Union, Tuple, Any, List
|
||||
|
||||
from DownloadKit import DownloadKit
|
||||
from requests import Session, Response
|
||||
|
||||
from .chromium_base import ChromiumBase, ChromiumBaseSetter
|
||||
@ -13,7 +12,7 @@ from .chromium_element import ChromiumElement
|
||||
from .chromium_frame import ChromiumFrame
|
||||
from .chromium_page import ChromiumPage, ChromiumTabRect
|
||||
from .session_element import SessionElement
|
||||
from .session_page import SessionPage, SessionPageSetter, DownloadSetter
|
||||
from .session_page import SessionPage, SessionPageSetter
|
||||
from .web_page import WebPage
|
||||
|
||||
|
||||
@ -36,8 +35,6 @@ class WebPageTab(SessionPage, ChromiumTab):
|
||||
self._mode: str = ...
|
||||
self._has_driver = ...
|
||||
self._has_session = ...
|
||||
self._download_set = ...
|
||||
self._download_path = ...
|
||||
|
||||
def __call__(self,
|
||||
loc_or_str: Union[Tuple[str, str], str, ChromiumElement, SessionElement],
|
||||
@ -124,8 +121,6 @@ class WebPageTab(SessionPage, ChromiumTab):
|
||||
def get_cookies(self, as_dict: bool = False, all_domains: bool = False,
|
||||
all_info: bool = False) -> Union[dict, list]: ...
|
||||
|
||||
# def _get_driver_cookies(self, as_dict: bool = False, all_info: bool = False) -> dict: ...
|
||||
|
||||
# ----------------重写SessionPage的函数-----------------------
|
||||
def post(self,
|
||||
url: str,
|
||||
@ -150,12 +145,6 @@ class WebPageTab(SessionPage, ChromiumTab):
|
||||
@property
|
||||
def set(self) -> WebPageTabSetter: ...
|
||||
|
||||
@property
|
||||
def download(self) -> DownloadKit: ...
|
||||
|
||||
@property
|
||||
def download_set(self) -> WebPageTabDownloadSetter: ...
|
||||
|
||||
def _find_elements(self, loc_or_ele: Union[Tuple[str, str], str, ChromiumElement, SessionElement, ChromiumFrame],
|
||||
timeout: float = None, single: bool = True, relative: bool = False, raise_err: bool = None) \
|
||||
-> Union[ChromiumElement, SessionElement, ChromiumFrame, str, None, List[Union[SessionElement, str]], List[
|
||||
@ -172,13 +161,3 @@ class WebPageTabSetter(ChromiumBaseSetter):
|
||||
def headers(self, headers: dict) -> None: ...
|
||||
|
||||
def cookies(self, cookies) -> None: ...
|
||||
|
||||
|
||||
class WebPageTabDownloadSetter(DownloadSetter):
|
||||
"""用于设置下载参数的类"""
|
||||
|
||||
def __init__(self, page: WebPageTab):
|
||||
self._page: WebPageTab = ...
|
||||
|
||||
@property
|
||||
def _switched_DownloadKit(self) -> DownloadKit: ...
|
||||
|
||||
@ -26,7 +26,7 @@ class ChromiumOptions(object):
|
||||
self.ini_path = om.ini_path
|
||||
options = om.chrome_options
|
||||
|
||||
self._download_path = om.paths.get('download_path', None)
|
||||
self._download_path = om.paths.get('download_path', '')
|
||||
self._arguments = options.get('arguments', [])
|
||||
self._binary_location = options.get('binary_location', '')
|
||||
self._extensions = options.get('extensions', [])
|
||||
@ -62,7 +62,7 @@ class ChromiumOptions(object):
|
||||
self.ini_path = None
|
||||
self._binary_location = "chrome"
|
||||
self._arguments = []
|
||||
self._download_path = None
|
||||
self._download_path = ''
|
||||
self._extensions = []
|
||||
self._prefs = {}
|
||||
self._timeouts = {'implicit': 10, 'pageLoad': 30, 'script': 30}
|
||||
|
||||
@ -1,5 +1,4 @@
|
||||
[paths]
|
||||
chromedriver_path =
|
||||
download_path =
|
||||
|
||||
[chrome_options]
|
||||
|
||||
@ -21,7 +21,7 @@ class SessionOptions(object):
|
||||
:param ini_path: ini文件路径
|
||||
"""
|
||||
self.ini_path = None
|
||||
self._download_path = None
|
||||
self._download_path = ''
|
||||
self._headers = None
|
||||
self._cookies = None
|
||||
self._auth = None
|
||||
@ -73,7 +73,7 @@ class SessionOptions(object):
|
||||
|
||||
self.set_proxies(om.proxies.get('http', None), om.proxies.get('https', None))
|
||||
self._timeout = om.timeouts.get('implicit', 10)
|
||||
self._download_path = om.paths.get('download_path', None)
|
||||
self._download_path = om.paths.get('download_path', '')
|
||||
|
||||
# ===========须独立处理的项开始============
|
||||
@property
|
||||
|
||||
@ -7,7 +7,6 @@ from re import search
|
||||
from time import sleep
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from DownloadKit import DownloadKit
|
||||
from requests import Session, Response
|
||||
from requests.structures import CaseInsensitiveDict
|
||||
from tldextract import extract
|
||||
@ -27,7 +26,6 @@ class SessionPage(BasePage):
|
||||
:param timeout: 连接超时时间，为None时从ini文件读取
|
||||
"""
|
||||
self._response = None
|
||||
self._DownloadKit = None
|
||||
self._session = None
|
||||
self._set = None
|
||||
self._set_start_options(session_or_options, None)
|
||||
@ -103,18 +101,6 @@ class SessionPage(BasePage):
|
||||
"""返回user agent"""
|
||||
return self.session.headers.get('user-agent', '')
|
||||
|
||||
@property
|
||||
def download_path(self):
|
||||
"""返回下载路径"""
|
||||
return self._download_path
|
||||
|
||||
@property
|
||||
def download(self):
|
||||
"""返回下载器对象"""
|
||||
if self._DownloadKit is None:
|
||||
self._DownloadKit = DownloadKit(session=self, goal_path=self.download_path)
|
||||
return self._DownloadKit
|
||||
|
||||
@property
|
||||
def session(self):
|
||||
"""返回session对象"""
|
||||
|
||||
@ -6,7 +6,7 @@
|
||||
from http.cookiejar import Cookie
|
||||
from typing import Any, Union, Tuple, List
|
||||
|
||||
from DownloadKit import DownloadKit
|
||||
# from DownloadKit import DownloadKit
|
||||
from requests import Session, Response
|
||||
from requests.adapters import HTTPAdapter
|
||||
from requests.auth import HTTPBasicAuth
|
||||
@ -27,8 +27,8 @@ class SessionPage(BasePage):
|
||||
self._session_options: SessionOptions = ...
|
||||
self._url: str = ...
|
||||
self._response: Response = ...
|
||||
self._download_path: str = ...
|
||||
self._DownloadKit: DownloadKit = ...
|
||||
# self._download_path: str = ...
|
||||
# self._DownloadKit: DownloadKit = ...
|
||||
self._url_available: bool = ...
|
||||
self.timeout: float = ...
|
||||
self.retry_times: int = ...
|
||||
@ -118,8 +118,8 @@ class SessionPage(BasePage):
|
||||
@property
|
||||
def set(self) -> SessionPageSetter: ...
|
||||
|
||||
@property
|
||||
def download(self) -> DownloadKit: ...
|
||||
# @property
|
||||
# def download(self) -> DownloadKit: ...
|
||||
|
||||
def post(self,
|
||||
url: str,
|
||||
|
||||
@ -43,7 +43,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
self._response = None
|
||||
self._set = None
|
||||
self._screencast = None
|
||||
self._DownloadKit = None
|
||||
|
||||
self._set_start_options(driver_or_options, session_or_options)
|
||||
self._set_runtime_settings()
|
||||
@ -103,7 +102,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
|
||||
self._timeouts = Timeout(self)
|
||||
self._page_load_strategy = self._driver_options.page_load_strategy
|
||||
self._download_path = None
|
||||
|
||||
if se_opt is not False:
|
||||
self.set.timeouts(implicit=self._session_options.timeout)
|
||||
@ -350,16 +348,12 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
user_agent = self.run_cdp('Runtime.evaluate', expression='navigator.userAgent;')['result']['value']
|
||||
self.session.headers.update({"User-Agent": user_agent})
|
||||
|
||||
# set_session_cookies(self.session, self._get_driver_cookies())
|
||||
set_session_cookies(self.session, super(SessionPage, self).get_cookies())
|
||||
|
||||
def cookies_to_browser(self):
|
||||
"""把session对象的cookies复制到浏览器"""
|
||||
if not self._has_driver:
|
||||
return
|
||||
|
||||
# set_browser_cookies(self, super().get_cookies(as_dict=True))
|
||||
# set_browser_cookies(self, super().get_cookies(all_domains=True))
|
||||
set_browser_cookies(self, super().get_cookies())
|
||||
|
||||
def get_cookies(self, as_dict=False, all_domains=False, all_info=False):
|
||||
@ -373,7 +367,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
return super().get_cookies(as_dict, all_domains, all_info)
|
||||
elif self._mode == 'd':
|
||||
return super(SessionPage, self).get_cookies(as_dict, all_domains, all_info)
|
||||
# return self._get_driver_cookies(as_dict, all_info)
|
||||
|
||||
def get_tab(self, tab_id=None):
|
||||
"""获取一个标签页对象
|
||||
@ -383,21 +376,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
tab_id = tab_id or self.tab_id
|
||||
return WebPageTab(self, tab_id)
|
||||
|
||||
# def _get_driver_cookies(self, as_dict=False, all_info=False):
|
||||
# """获取浏览器cookies
|
||||
# :param as_dict: 是否以dict形式返回，为True时all_info无效
|
||||
# :param all_info: 是否返回所有信息
|
||||
# :return: cookies信息
|
||||
# """
|
||||
# cookies = self.run_cdp('Network.getCookies')['cookies']
|
||||
# if as_dict:
|
||||
# return {cookie['name']: cookie['value'] for cookie in cookies}
|
||||
# elif all_info:
|
||||
# return cookies
|
||||
# else:
|
||||
# return [{'name': cookie['name'], 'value': cookie['value'], 'domain': cookie['domain']}
|
||||
# for cookie in cookies]
|
||||
|
||||
def close_driver(self):
|
||||
"""关闭driver及浏览器"""
|
||||
if self._has_driver:
|
||||
|
||||
@ -5,14 +5,13 @@
|
||||
"""
|
||||
from typing import Union, Tuple, List, Any
|
||||
|
||||
from DownloadKit import DownloadKit
|
||||
from requests import Session, Response
|
||||
|
||||
from .base import BasePage
|
||||
from .chromium_driver import ChromiumDriver
|
||||
from .chromium_element import ChromiumElement
|
||||
from .chromium_frame import ChromiumFrame
|
||||
from .chromium_page import ChromiumPage, ChromiumDownloadSetter, ChromiumPageSetter
|
||||
from .chromium_page import ChromiumPage, ChromiumPageSetter
|
||||
from .chromium_tab import WebPageTab
|
||||
from .configs.chromium_options import ChromiumOptions
|
||||
from .configs.session_options import SessionOptions
|
||||
@ -33,8 +32,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
self.address: str = ...
|
||||
self._session_options: Union[SessionOptions, None] = ...
|
||||
self._driver_options: Union[ChromiumOptions, None] = ...
|
||||
self._download_set: WebPageDownloadSetter = ...
|
||||
self._download_path: str = ...
|
||||
self._tab_obj: ChromiumDriver = ...
|
||||
|
||||
def __call__(self,
|
||||
@ -81,12 +78,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
@timeout.setter
|
||||
def timeout(self, second: float) -> None: ...
|
||||
|
||||
@property
|
||||
def download_path(self) -> str: ...
|
||||
|
||||
@property
|
||||
def download_set(self) -> WebPageDownloadSetter: ...
|
||||
|
||||
def get(self,
|
||||
url: str,
|
||||
show_errmsg: bool = False,
|
||||
@ -131,8 +122,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
|
||||
def get_tab(self, tab_id: str = None) -> WebPageTab: ...
|
||||
|
||||
# def _get_driver_cookies(self, as_dict: bool = False, all_info: bool = False) -> dict: ...
|
||||
|
||||
def close_driver(self) -> None: ...
|
||||
|
||||
def close_session(self) -> None: ...
|
||||
@ -158,9 +147,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
verify: Any | None = ...,
|
||||
cert: Any | None = ...) -> bool: ...
|
||||
|
||||
@property
|
||||
def download(self) -> DownloadKit: ...
|
||||
|
||||
@property
|
||||
def set(self) -> WebPageSetter: ...
|
||||
|
||||
@ -187,21 +173,3 @@ class WebPageSetter(ChromiumPageSetter):
|
||||
def headers(self, headers: dict) -> None: ...
|
||||
|
||||
def cookies(self, cookies) -> None: ...
|
||||
|
||||
|
||||
class WebPageDownloadSetter(ChromiumDownloadSetter):
|
||||
def __init__(self, page: WebPage):
|
||||
self._page: WebPage = ...
|
||||
self._behavior: str = ...
|
||||
self._session: Session = None
|
||||
|
||||
@property
|
||||
def _switched_DownloadKit(self) -> DownloadKit: ...
|
||||
|
||||
def save_path(self, path) -> None: ...
|
||||
|
||||
def by_browser(self) -> None: ...
|
||||
|
||||
def by_DownloadKit(self) -> None: ...
|
||||
|
||||
def _download_by_DownloadKit(self, **kwargs) -> None: ...
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user