blackteay/DrissionPage
1
0
Fork 0
mirror of https://gitee.com/g1879/DrissionPage.git synced 2024-12-10 04:00:23 +08:00
Code Issues Projects Releases Wiki Activity

各页面对象删除download_set，待测试

Browse Source
This commit is contained in:
g1879 2023-05-18 14:37:55 +08:00
parent 9774e3d389
commit 453b0fed85
6 changed files with 26 additions and 274 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
DrissionPage/chromium_base.py
View File

@ -24,7 +24,6 @@ from .errors import ContextLossError, ElementLossError, AlertExistsError, CDPErr
NoRectError, BrowserConnectError, GetDocumentError
from .network_listener import NetworkListener
from .session_element import make_session_ele
from .session_page import DownloadSetter
class ChromiumBase(BasePage):
@ -44,8 +43,9 @@ class ChromiumBase(BasePage):
self._set = None
self._screencast = None
self._listener = None
self._download_set = None
self._DownloadKit = None
self._download_path = None
self._session = None
if isinstance(address, int) or (isinstance(address, str) and address.isdigit()):
address = f'127.0.0.1:{address}'
@ -385,17 +385,23 @@ class ChromiumBase(BasePage):
p = self._download_path or ''
return str(Path(p).absolute())
@property
def download_set(self):
"""返回用于设置下载参数的对象"""
if self._download_set is None:
self._download_set = BaseDownloadSetter(self)
return self._download_set
@property
def download(self):
"""返回下载器对象"""
return self.download_set.DownloadKit
if self._DownloadKit is None:
self._DownloadKit = DownloadKit(session=self, goal_path=self.download_path)
return self._DownloadKit
@property
def session(self):
"""返回用于DownloadKit的Session对象"""
if self._session is None:
self._session = Session()
ua = self.run_cdp('Runtime.evaluate', expression='navigator.userAgent;')['result']['value']
self._session.headers.update({"User-Agent": ua})
set_session_cookies(self._session, self.get_cookies(as_dict=False, all_info=False))
return self._session
def run_cdp(self, cmd, **cmd_args):
"""执行Chrome DevTools Protocol语句
@ -894,45 +900,6 @@ class ChromiumBase(BasePage):
return str(path.absolute())
class BaseDownloadSetter(DownloadSetter):
"""用于设置下载参数的类"""
def __init__(self, page):
"""
:param page: ChromiumPage对象
"""
super().__init__(page)
self._session = None
self._show_msg = True
@property
def session(self):
"""返回用于DownloadKit的Session对象"""
if self._session is None:
self._session = Session()
ua = self._page.run_cdp('Runtime.evaluate', expression='navigator.userAgent;')['result']['value']
self._session.headers.update({"User-Agent": ua})
self._cookies_to_session()
return self._session
@property
def DownloadKit(self):
if self._DownloadKit is None:
self._DownloadKit = DownloadKit(session=self._page, goal_path=self._page.download_path)
return self._DownloadKit
def show_msg(self, on_off=True):
"""是否显示下载信息
:param on_off: bool表示开或关
:return: None
"""
self._show_msg = on_off
def _cookies_to_session(self):
"""把driver对象的cookies复制到session对象"""
set_session_cookies(self._session, self._page.get_cookies(as_dict=False, all_info=False))
class ChromiumBaseSetter(object):
def __init__(self, page):
self._page = page

8
DrissionPage/chromium_base.pyi
View File

@ -8,7 +8,6 @@ from typing import Union, Tuple, List, Any
from DataRecorder import Recorder
from DownloadKit import DownloadKit
from DownloadKit.mission import Mission
from requests import Session
from requests.cookies import RequestsCookieJar
@ -47,7 +46,8 @@ class ChromiumBase(BasePage):
self._screencast: Screencast = ...
self._listener: NetworkListener = ...
self._download_path: str = ...
self._download_set: BaseDownloadSetter = ...
self._DownloadKit: DownloadKit = ...
self._session: Session = ...
def _connect_browser(self, tab_id: str = None) -> None: ...
@ -139,13 +139,13 @@ class ChromiumBase(BasePage):
def listener(self) -> NetworkListener: ...
@property
def download_set(self) -> BaseDownloadSetter: ...
def download_path(self) -> str: ...
@property
def download(self) -> DownloadKit: ...
@property
def download_path(self) -> str: ...
def session(self) -> Session: ...
def run_js(self, script: str, *args: Any, as_expr: bool = False) -> Any: ...

24
DrissionPage/chromium_tab.py
View File

@ -7,7 +7,7 @@ from copy import copy
from .chromium_base import ChromiumBase, ChromiumBaseSetter
from .commons.web import set_session_cookies, set_browser_cookies
from .session_page import SessionPage, SessionPageSetter, DownloadSetter
from .session_page import SessionPage, SessionPageSetter
class ChromiumTab(ChromiumBase):
@ -156,13 +156,6 @@ class WebPageTab(SessionPage, ChromiumTab):
self._set = WebPageTabSetter(self)
return self._set
@property
def download_set(self):
"""返回下载设置对象"""
if self._download_set is None:
self._download_set = WebPageTabDownloadSetter(self)
return self._download_set
@property
def download(self):
"""返回下载器对象"""
@ -386,18 +379,3 @@ class WebPageTabSetter(ChromiumBaseSetter):
self._chromium_setter.user_agent(ua, platform)
class WebPageTabDownloadSetter(DownloadSetter):
"""用于设置下载参数的类"""
def __init__(self, page):
super().__init__(page)
self._session = page.session
@property
def _switched_DownloadKit(self):
"""返回从浏览器同步cookies后的Session对象"""
if self._page.mode == 'd':
ua = self._page.run_cdp('Runtime.evaluate', expression='navigator.userAgent;')['result']['value']
self._page.session.headers.update({"User-Agent": ua})
set_session_cookies(self._page.session, self._page.get_cookies(as_dict=False, all_domains=False))
return self.DownloadKit

80
DrissionPage/session_page.py
View File

@ -27,7 +27,7 @@ class SessionPage(BasePage):
:param timeout: 连接超时时间，为None时从ini文件读取
"""
self._response = None
self._download_set = None
self._DownloadKit = None
self._session = None
self._set = None
self._set_start_options(session_or_options, None)
@ -103,17 +103,12 @@ class SessionPage(BasePage):
"""返回下载路径"""
return self._download_path
@property
def download_set(self):
"""返回用于设置下载参数的对象"""
if self._download_set is None:
self._download_set = DownloadSetter(self)
return self._download_set
@property
def download(self):
"""返回下载器对象"""
return self.download_set.DownloadKit
if self._DownloadKit is None:
self._DownloadKit = DownloadKit(session=self, goal_path=self.download_path)
return self._DownloadKit
@property
def session(self):
@ -440,73 +435,6 @@ class SessionPageSetter(object):
self._page.session.mount(url, adapter)
class DownloadSetter(object):
"""用于设置下载参数的类"""
def __init__(self, page):
self._page = page
self._DownloadKit = None
self._file_exists = 'rename'
@property
def DownloadKit(self):
if self._DownloadKit is None:
self._DownloadKit = DownloadKit(session=self._page, goal_path=self._page.download_path)
return self._DownloadKit
@property
def if_file_exists(self):
"""返回用于设置存在同名文件时处理方法的对象"""
return FileExists(self)
def split(self, on_off):
"""设置是否允许拆分大文件用多线程下载
:param on_off: 是否启用多线程下载大文件
:return: None
"""
self.DownloadKit.split = on_off
def save_path(self, path):
"""设置下载保存路径
:param path: 下载保存路径
:return: None
"""
path = path if path is None else str(path)
self._page._download_path = path
self.DownloadKit.goal_path = path
class FileExists(object):
"""用于设置存在同名文件时处理方法"""
def __init__(self, setter):
"""
:param setter: DownloadSetter对象
"""
self._setter = setter
def __call__(self, mode):
if mode not in ('skip', 'rename', 'overwrite'):
raise ValueError("mode参数只能是'skip', 'rename', 'overwrite'")
self._setter.DownloadKit.file_exists = mode
self._setter._file_exists = mode
def skip(self):
"""设为跳过"""
self._setter.DownloadKit.file_exists = 'skip'
self._setter._file_exists = 'skip'
def rename(self):
"""设为重命名，文件名后加序号"""
self._setter.DownloadKit._file_exists = 'rename'
self._setter._file_exists = 'rename'
def overwrite(self):
"""设为覆盖"""
self._setter.DownloadKit._file_exists = 'overwrite'
self._setter._file_exists = 'overwrite'
def check_headers(kwargs, headers, arg) -> bool:
"""检查kwargs或headers中是否有arg所示属性"""
return arg in kwargs['headers'] or arg in headers

37
DrissionPage/session_page.pyi
View File

@ -3,7 +3,6 @@
@Author : g1879
@Contact : g1879@qq.com
"""
from pathlib import Path
from typing import Any, Union, Tuple, List
from DownloadKit import DownloadKit
@ -14,7 +13,6 @@ from requests.cookies import RequestsCookieJar
from requests.structures import CaseInsensitiveDict
from .base import BasePage
from .chromium_base import ChromiumBase
from .commons.constants import NoneElement
from .configs.session_options import SessionOptions
from .session_element import SessionElement
@ -29,7 +27,7 @@ class SessionPage(BasePage):
self._url: str = ...
self._response: Response = ...
self._download_path: str = ...
self._download_set: DownloadSetter = ...
self._DownloadKit: DownloadKit = ...
self._url_available: bool = ...
self.timeout: float = ...
self.retry_times: int = ...
@ -65,9 +63,6 @@ class SessionPage(BasePage):
@property
def download_path(self) -> str: ...
@property
def download_set(self) -> DownloadSetter: ...
def get(self,
url: str,
show_errmsg: bool | None = False,
@ -200,36 +195,6 @@ class SessionPageSetter(object):
def add_adapter(self, url: str, adapter: HTTPAdapter) -> None: ...
class DownloadSetter(object):
def __init__(self, page: Union[SessionPage, ChromiumBase]):
self._page: SessionPage = ...
self._DownloadKit: DownloadKit = ...
self._file_exists: str = ...
@property
def DownloadKit(self) -> DownloadKit: ...
@property
def if_file_exists(self) -> FileExists: ...
def split(self, on_off: bool) -> None: ...
def save_path(self, path: Union[str, Path]): ...
class FileExists(object):
def __init__(self, setter: DownloadSetter):
self._setter: DownloadSetter = ...
def __call__(self, mode: str) -> None: ...
def skip(self) -> None: ...
def rename(self) -> None: ...
def overwrite(self) -> None: ...
def check_headers(kwargs: Union[dict, CaseInsensitiveDict], headers: Union[dict, CaseInsensitiveDict],
arg: str) -> bool: ...

86
DrissionPage/web_page.py
View File

@ -42,7 +42,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
self._driver_options = None
self._session_options = None
self._response = None
# self._download_set = None
self._set = None
self._screencast = None
@ -208,23 +207,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
"""
self.set.timeouts(implicit=second)
# @property
# def download_path(self):
# """返回默认下载路径"""
# return super(SessionPage, self).download_path
#
# @property
# def download_set(self):
# """返回下载设置对象"""
# if self._download_set is None:
# self._download_set = WebPageDownloadSetter(self)
# return self._download_set
#
# @property
# def download(self):
# """返回下载器对象"""
# return self.download_set._switched_DownloadKit
@property
def set(self):
"""返回用于等待的对象"""
@ -360,8 +342,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
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, self._get_driver_cookies())
def cookies_to_browser(self):
@ -490,69 +470,3 @@ class WebPageSetter(ChromiumPageSetter):
self._session_setter.user_agent(ua)
else:
self._chromium_setter.user_agent(ua, platform)
# class WebPageDownloadSetter(BaseDownloadSetter):
# """用于设置下载参数的类"""
#
# def __init__(self, page):
# super().__init__(page)
# self._session = page.session
#
# @property
# def _switched_DownloadKit(self):
# """返回从浏览器同步cookies后的Session对象"""
# if self._page.mode == 'd':
# self._cookies_to_session()
# return self.DownloadKit
#
# def save_path(self, path):
# """设置下载路径
# :param path: 下载路径
# :return: None
# """
# path = path or ''
# path = Path(path).absolute()
# path.mkdir(parents=True, exist_ok=True)
# path = str(path)
# self._save_path = path
# self._page._download_path = path
# self.DownloadKit.goal_path = path
#
# if self._page._has_driver:
# try:
# self._page.browser_driver.Browser.setDownloadBehavior(behavior=self._behavior, downloadPath=path,
# eventsEnabled=True)
# except CDPError:
# warn('\n您的浏览器版本太低，用新标签页下载文件可能崩溃，建议升级。')
# self._page.run_cdp('Page.setDownloadBehavior', behavior=self._behavior, downloadPath=path)
#
# def by_browser(self):
# """设置使用浏览器下载文件"""
# if not self._page._has_driver:
# raise RuntimeError('浏览器未连接。')
#
# try:
# self._page.browser_driver.Browser.setDownloadBehavior(behavior='allowAndName', eventsEnabled=True,
# downloadPath=self._page.download_path)
# self._page.browser_driver.Browser.downloadWillBegin = self._download_will_begin
# self._page.browser_driver.Browser.downloadProgress = self._download_progress
#
# except CDPError:
# warn('\n您的浏览器版本太低，用新标签页下载文件可能崩溃，建议升级。')
# self._page.driver.Page.setDownloadBehavior(behavior='allowAndName', downloadPath=self._page.download_path)
# self._page.driver.Page.downloadWillBegin = self._download_will_begin
# self._page.driver.Page.downloadProgress = self._download_progress
#
# self._behavior = 'allowAndName'
#
# def by_DownloadKit(self):
# """设置使用DownloadKit下载文件"""
# if self._page._has_driver:
# try:
# self._page.browser_driver.Browser.setDownloadBehavior(behavior='deny', eventsEnabled=True)
# self._page.browser_driver.Browser.downloadWillBegin = self._download_by_DownloadKit
# # self._page.browser_driver.Browser.downloadProgress = None
# except CDPError:
# raise RuntimeError('您的浏览器版本太低，不支持此方法，请升级。')
#
# self._behavior = 'deny'