页面类增加download_set,WebPage的未完成,待测试

This commit is contained in:
g1879 2023-01-15 23:14:20 +08:00
parent 8b7df4675b
commit fbcb2d3b37
6 changed files with 139 additions and 63 deletions

View File

@ -11,10 +11,11 @@ from time import perf_counter, sleep
from requests import Session from requests import Session
from .chromium_base import Timeout, ChromiumBase from .chromium_base import Timeout, ChromiumBase
from .chromium_driver import ChromiumDriver
from .chromium_tab import ChromiumTab from .chromium_tab import ChromiumTab
from .common import connect_browser from .common import connect_browser
from .config import DriverOptions from .config import DriverOptions
from .chromium_driver import ChromiumDriver from .session_page import DownloadSetter
class ChromiumPage(ChromiumBase): class ChromiumPage(ChromiumBase):
@ -86,7 +87,7 @@ class ChromiumPage(ChromiumBase):
self._tab_obj.Page.javascriptDialogOpening = self._on_alert_open self._tab_obj.Page.javascriptDialogOpening = self._on_alert_open
self._tab_obj.Page.javascriptDialogClosed = self._on_alert_close self._tab_obj.Page.javascriptDialogClosed = self._on_alert_close
self._download_path = self.set_download_path(self.options.download_path) self.download_set.save_path(self.options.download_path)
def _set_options(self): def _set_options(self):
"""从配置中读取设置""" """从配置中读取设置"""
@ -131,17 +132,10 @@ class ChromiumPage(ChromiumBase):
p = self._download_path or '' p = self._download_path or ''
return str(Path(p).absolute()) return str(Path(p).absolute())
def set_download_path(self, path): @property
"""设置下载路径 \n def download_set(self):
:param path: 下载路径 """返回用于设置下载参数的对象"""
:return: None return ChromiumDownloadSetter(self)
"""
path = path or ''
path = Path(path).absolute()
path.mkdir(parents=True, exist_ok=True)
path = str(path)
self._download_path = path
self.run_cdp('Browser.setDownloadBehavior', behavior='allow', downloadPath=path, not_change=True)
def get_tab(self, tab_id=None): def get_tab(self, tab_id=None):
"""获取一个标签页对象 \n """获取一个标签页对象 \n
@ -371,6 +365,25 @@ class ChromiumPage(ChromiumBase):
self._tab_obj.has_alert = True self._tab_obj.has_alert = True
class ChromiumDownloadSetter(DownloadSetter):
"""用于设置下载参数的类"""
def save_path(self, path):
"""设置下载路径 \n
:param path: 下载路径
:return: None
"""
path = path or ''
path = Path(path).absolute()
path.mkdir(parents=True, exist_ok=True)
path = str(path)
self._page._download_path = path
try:
self._page.run_cdp('Browser.setDownloadBehavior', behavior='allow', downloadPath=path, not_change=True)
except:
self._page.run_cdp('Page.setDownloadBehavior', behavior='allow', downloadPath=path, not_change=True)
class Alert(object): class Alert(object):
"""用于保存alert信息的类""" """用于保存alert信息的类"""

View File

@ -8,9 +8,9 @@ from pathlib import Path
from typing import Union, Tuple, List from typing import Union, Tuple, List
from .chromium_base import ChromiumBase from .chromium_base import ChromiumBase
from .chromium_driver import ChromiumDriver
from .chromium_tab import ChromiumTab from .chromium_tab import ChromiumTab
from .config import DriverOptions from .config import DriverOptions
from .chromium_driver import ChromiumDriver
class ChromiumPage(ChromiumBase): class ChromiumPage(ChromiumBase):
@ -53,13 +53,11 @@ class ChromiumPage(ChromiumBase):
def download_list(self) -> list: ... def download_list(self) -> list: ...
@property @property
def set_download(self) -> DownloadSetter: ... def download_set(self) -> ChromiumDownloadSetter: ...
@property @property
def download_path(self) -> str: ... def download_path(self) -> str: ...
def set_download_path(self, path: Union[str, Path]) -> None: ...
def get_tab(self, tab_id: str = None) -> ChromiumTab: ... def get_tab(self, tab_id: str = None) -> ChromiumTab: ...
def get_screenshot(self, path: [str, Path] = None, def get_screenshot(self, path: [str, Path] = None,
@ -95,6 +93,13 @@ class ChromiumPage(ChromiumBase):
def _on_alert_open(self, **kwargs): ... def _on_alert_open(self, **kwargs): ...
class ChromiumDownloadSetter(object):
def __init__(self, page: ChromiumPage):
self._page: ChromiumPage = ...
def save_path(self, path: Union[str, Path]): ...
class Alert(object): class Alert(object):
def __init__(self): def __init__(self):

View File

@ -120,14 +120,10 @@ class SessionPage(BasePage):
"""返回下载路径""" """返回下载路径"""
return self._download_path return self._download_path
def set_download_path(self, path): @property
"""设置下载路径 \n def download_set(self):
:param path: 下载路径 """返回用于设置下载参数的对象"""
:return: None return DownloadSetter(self)
"""
self._download_path = str(path)
if self._download_kit is not None:
self._download_kit.goal_path = self._download_path
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):
"""用get方式跳转到url \n """用get方式跳转到url \n
@ -327,6 +323,23 @@ class SessionPage(BasePage):
return r, f'状态码:{r.status_code}' return r, f'状态码:{r.status_code}'
class DownloadSetter(object):
"""用于设置下载参数的类"""
def __init__(self, page):
self._page = page
def save_path(self, path):
"""设置下载保存路径 \n
:param path: 下载保存路径
:return: None
"""
path = path if path is None else str(path)
self._page._download_path = path
if self._page._download_kit is not None:
self._page._download_kit.goal_path = path
def check_headers(kwargs, headers, arg) -> bool: def check_headers(kwargs, headers, arg) -> bool:
"""检查kwargs或headers中是否有arg所示属性""" """检查kwargs或headers中是否有arg所示属性"""
return arg in kwargs['headers'] or arg in headers return arg in kwargs['headers'] or arg in headers

View File

@ -12,8 +12,8 @@ from requests.cookies import RequestsCookieJar
from requests.structures import CaseInsensitiveDict from requests.structures import CaseInsensitiveDict
from .base import BasePage from .base import BasePage
from .session_element import SessionElement
from .config import SessionOptions from .config import SessionOptions
from .session_element import SessionElement
class SessionPage(BasePage): class SessionPage(BasePage):
@ -56,7 +56,8 @@ class SessionPage(BasePage):
@property @property
def download_path(self) -> str: ... def download_path(self) -> str: ...
def set_download_path(self, path: Union[str, Path]) -> None: ... @property
def download_set(self) -> DownloadSetter: ...
def get(self, def get(self,
url: str, url: str,
@ -149,6 +150,13 @@ class SessionPage(BasePage):
**kwargs) -> tuple: ... **kwargs) -> tuple: ...
class DownloadSetter(object):
def __init__(self, page: SessionPage):
self._page: SessionPage = ...
def save_path(self, path: Union[str, Path]): ...
def check_headers(kwargs: Union[dict, CaseInsensitiveDict], headers: Union[dict, CaseInsensitiveDict], def check_headers(kwargs: Union[dict, CaseInsensitiveDict], headers: Union[dict, CaseInsensitiveDict],
arg: str) -> bool: ... arg: str) -> bool: ...

View File

@ -3,6 +3,7 @@
@Author : g1879 @Author : g1879
@Contact : g1879@qq.com @Contact : g1879@qq.com
""" """
from pathlib import Path
from time import sleep from time import sleep
from requests import Session from requests import Session
@ -10,10 +11,10 @@ from tldextract import extract
from .base import BasePage from .base import BasePage
from .chromium_base import ChromiumBase, Timeout from .chromium_base import ChromiumBase, Timeout
from .chromium_page import ChromiumPage from .chromium_driver import ChromiumDriver
from .chromium_page import ChromiumPage, ChromiumDownloadSetter
from .config import DriverOptions, SessionOptions, cookies_to_tuple from .config import DriverOptions, SessionOptions, cookies_to_tuple
from .session_page import SessionPage from .session_page import SessionPage
from .chromium_driver import ChromiumDriver
class WebPage(SessionPage, ChromiumPage, BasePage): class WebPage(SessionPage, ChromiumPage, BasePage):
@ -42,6 +43,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
self._setting_tab_id = tab_id self._setting_tab_id = tab_id
self._response = None self._response = None
self._download_kit = None self._download_kit = None
self._download_set = None
if self._mode == 'd': if self._mode == 'd':
self._to_d_mode() self._to_d_mode()
@ -159,28 +161,12 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
"""返回默认下载路径""" """返回默认下载路径"""
return super(SessionPage, self).download_path return super(SessionPage, self).download_path
def set_download_tool(self, use_browser=False): @property
"""设置下载释是否使用浏览器 \n def download_set(self):
:param use_browser: 是否使用浏览器 """返回下载设置对象"""
:return: None if self._download_set is None:
""" self._download_set = WebPageDownloadSetter(self)
if use_browser: return self._download_set
self._tab_obj.Browser.setDownloadBehavior(behavior='allow')
self._tab_obj.Page.downloadWillBegin = None
else:
self._tab_obj.Page.downloadWillBegin = self._on_download_begin
self._tab_obj.Browser.downloadWillBegin = self._on_download_begin
self._tab_obj.Browser.setDownloadBehavior(behavior='deny')
def set_download_path(self, path):
"""设置默认下载路径
:param path: 下载路径
:return: None
"""
if self._has_driver:
super(SessionPage, self).set_download_path(path)
else:
super().set_download_path(path)
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 \n """跳转到一个url \n
@ -480,9 +466,53 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
self._tab_obj = None self._tab_obj = None
self._has_driver = None self._has_driver = None
def _on_download_begin(self, **kwargs):
class WebPageDownloadSetter(ChromiumDownloadSetter):
"""用于设置下载参数的类"""
def __init__(self, page):
super().__init__(page)
self._behavior = 'allow'
def save_path(self, path):
"""设置下载路径 \n
:param path: 下载路径
:return: None
"""
# todo: 设置时判断模式,初始设置
path = path or ''
path = Path(path).absolute()
path.mkdir(parents=True, exist_ok=True)
path = str(path)
self._page._download_path = path
if self._page._has_driver:
try:
self._page.run_cdp('Browser.setDownloadBehavior', behavior=self._behavior, downloadPath=path,
not_change=True)
except:
self._page.run_cdp('Page.setDownloadBehavior', behavior=self._behavior, downloadPath=path,
not_change=True)
def use_browser(self):
"""设置使用浏览器下载文件"""
if not self._page._has_driver:
raise RuntimeError('浏览器未连接。')
self._page.driver.Page.downloadWillBegin = None
self._page.driver.Browser.downloadWillBegin = None
self._page.driver.Browser.setDownloadBehavior(behavior='allow')
self._behavior = 'allow'
def use_DownloadKit(self):
"""设置使用DownloadKit下载文件"""
self._page.driver.Page.downloadWillBegin = self._download_by_DownloadKit
self._page.driver.Browser.downloadWillBegin = self._download_by_DownloadKit
self._page.driver.Browser.setDownloadBehavior(behavior='deny')
self._behavior = 'deny'
def _download_by_DownloadKit(self, **kwargs):
gid = kwargs['guid'] gid = kwargs['guid']
self._tab_obj.Browser.cancelDownload(guid=gid) self._page.run_cdp('Browser.cancelDownload', guid=gid, not_change=True)
url = kwargs['url'] url = kwargs['url']
name = kwargs['suggestedFilename'] name = kwargs['suggestedFilename']
self.download(url, goal_path=self.download_path, rename=name) print(f'开始下载:{url}')
self._page.download.add(url, goal_path=self._page.download_path, rename=name)

View File

@ -3,20 +3,19 @@
@Author : g1879 @Author : g1879
@Contact : g1879@qq.com @Contact : g1879@qq.com
""" """
from pathlib import Path
from typing import Union, Tuple, List, Any from typing import Union, Tuple, List, Any
from DownloadKit import DownloadKit from DownloadKit import DownloadKit
from requests import Session, Response from requests import Session, Response
from .base import BasePage from .base import BasePage
from .chromium_driver import ChromiumDriver
from .chromium_element import ChromiumElement from .chromium_element import ChromiumElement
from .chromium_frame import ChromiumFrame from .chromium_frame import ChromiumFrame
from .chromium_page import ChromiumPage from .chromium_page import ChromiumPage, ChromiumDownloadSetter
from .config import DriverOptions, SessionOptions from .config import DriverOptions, SessionOptions
from .session_element import SessionElement from .session_element import SessionElement
from .session_page import SessionPage from .session_page import SessionPage
from .chromium_driver import ChromiumDriver
class WebPage(SessionPage, ChromiumPage, BasePage): class WebPage(SessionPage, ChromiumPage, BasePage):
@ -33,6 +32,8 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
self._session_options: dict = ... self._session_options: dict = ...
self._driver_options: DriverOptions = ... self._driver_options: DriverOptions = ...
self._setting_tab_id: str = ... self._setting_tab_id: str = ...
self._download_kit: DownloadKit = ...
self._download_set: WebPageDownloadSetter = ...
def __call__(self, def __call__(self,
loc_or_str: Union[Tuple[str, str], str, ChromiumElement, SessionElement], loc_or_str: Union[Tuple[str, str], str, ChromiumElement, SessionElement],
@ -84,7 +85,8 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
@property @property
def download_path(self) -> str: ... def download_path(self) -> str: ...
def set_download_tool(self, use_browser:bool=False) -> None: ... @property
def download_set(self) -> WebPageDownloadSetter: ...
def get(self, def get(self,
url: str, url: str,
@ -177,10 +179,15 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
def _on_download_begin(self, **kwargs): ... def _on_download_begin(self, **kwargs): ...
class DownloadSetter(object): class WebPageDownloadSetter(ChromiumDownloadSetter):
def __init__(self, page: ChromiumPage): def __init__(self, page: WebPage):
self._page: ChromiumPage = ... self._page: WebPage = ...
self._behavior: str = ...
def deny(self) -> None: ... def save_path(self, path) -> None: ...
def save_path(self, path: Union[str, Path] = '') -> None: ... def use_browser(self) -> None: ...
def use_DownloadKit(self) -> None: ...
def _download_by_DownloadKit(self, **kwargs) -> None: ...