上传文件时支持传入相对路径;ChromiumPage也支持用DownloadKit下载

This commit is contained in:
g1879 2023-01-20 23:11:07 +08:00
parent 9e6534aa1c
commit 33ab90fe52
8 changed files with 115 additions and 31 deletions

View File

@ -571,12 +571,13 @@ class ChromiumElement(DrissionElement):
self.page.driver.Input.insertText(text=vals) self.page.driver.Input.insertText(text=vals)
def _set_file_input(self, files): def _set_file_input(self, files):
"""设置上传控件值 """往上传控件写入路径
:param files: 文件路径列表或字符串字符串时多个文件用回车分隔 :param files: 文件路径列表或字符串字符串时多个文件用回车分隔
:return: None :return: None
""" """
if isinstance(files, str): if isinstance(files, str):
files = files.split('\n') files = files.split('\n')
files = [str(Path(i).absolute()) for i in files]
self.page.run_cdp('DOM.setFileInputFiles', files=files, nodeId=self._node_id, not_change=True) self.page.run_cdp('DOM.setFileInputFiles', files=files, nodeId=self._node_id, not_change=True)
def clear(self, by_js=False): def clear(self, by_js=False):

View File

@ -7,12 +7,16 @@ from pathlib import Path
from platform import system from platform import system
from time import perf_counter, sleep from time import perf_counter, sleep
from DownloadKit import DownloadKit
from requests import Session
from .chromium_base import ChromiumBase, Timeout from .chromium_base import ChromiumBase, Timeout
from .chromium_driver import ChromiumDriver from .chromium_driver import ChromiumDriver
from .chromium_tab import ChromiumTab from .chromium_tab import ChromiumTab
from .configs.chromium_options import ChromiumOptions from .configs.chromium_options import ChromiumOptions
from .configs.driver_options import DriverOptions from .configs.driver_options import DriverOptions
from .functions.browser import connect_browser from .functions.browser import connect_browser
from .functions.web import set_session_cookies
from .session_page import DownloadSetter from .session_page import DownloadSetter
@ -26,6 +30,9 @@ class ChromiumPage(ChromiumBase):
:param timeout: 超时时间 :param timeout: 超时时间
""" """
super().__init__(addr_driver_opts, tab_id, timeout) super().__init__(addr_driver_opts, tab_id, timeout)
self._session = None
self._download_set = None
self._download_kit = None
def _connect_browser(self, addr_driver_opts=None, tab_id=None): def _connect_browser(self, addr_driver_opts=None, tab_id=None):
"""连接浏览器,在第一次时运行 """连接浏览器,在第一次时运行
@ -137,7 +144,17 @@ class ChromiumPage(ChromiumBase):
@property @property
def download_set(self): def download_set(self):
"""返回用于设置下载参数的对象""" """返回用于设置下载参数的对象"""
return ChromiumDownloadSetter(self) if self._download_set is None:
self._download_set = ChromiumDownloadSetter(self)
return self._download_set
@property
def download(self):
"""返回下载器对象"""
self.cookies_to_session()
if self._download_kit is None:
self._download_kit = DownloadKit(session=self._session, goal_path=self.download_path)
return self._download_kit
def get_tab(self, tab_id=None): def get_tab(self, tab_id=None):
"""获取一个标签页对象 """获取一个标签页对象
@ -344,6 +361,15 @@ class ChromiumPage(ChromiumBase):
"""显示浏览器窗口只在Windows系统可用""" """显示浏览器窗口只在Windows系统可用"""
show_or_hide_browser(self, hide=False) show_or_hide_browser(self, hide=False)
def cookies_to_session(self):
"""把driver对象的cookies复制到session对象"""
if self._session is None:
self._session = Session()
selenium_user_agent = self._tab_obj.Runtime.evaluate(expression='navigator.userAgent;')['result']['value']
self._session.headers.update({"User-Agent": selenium_user_agent})
set_session_cookies(self._session, self.get_cookies(as_dict=True))
def quit(self): def quit(self):
"""关闭浏览器""" """关闭浏览器"""
self._tab_obj.Browser.close() self._tab_obj.Browser.close()
@ -373,6 +399,10 @@ class ChromiumPage(ChromiumBase):
class ChromiumDownloadSetter(DownloadSetter): class ChromiumDownloadSetter(DownloadSetter):
"""用于设置下载参数的类""" """用于设置下载参数的类"""
def __init__(self, page):
super().__init__(page)
self._behavior = 'allow'
def save_path(self, path): def save_path(self, path):
"""设置下载路径 """设置下载路径
:param path: 下载路径 :param path: 下载路径
@ -388,6 +418,31 @@ class ChromiumDownloadSetter(DownloadSetter):
except: except:
self._page.run_cdp('Page.setDownloadBehavior', behavior='allow', downloadPath=path, not_change=True) self._page.run_cdp('Page.setDownloadBehavior', behavior='allow', downloadPath=path, not_change=True)
if self._page._download_kit is not None:
self._page._download_kit.goal_path = path
def use_browser(self):
"""设置使用浏览器下载文件"""
self._page.driver.Page.downloadWillBegin = None
self._page.driver.Browser.downloadWillBegin = None
self._page.driver.Browser.setDownloadBehavior(behavior='allow', downloadPath=self._page.download_path)
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']
self._page.run_cdp('Browser.cancelDownload', guid=gid, not_change=True)
url = kwargs['url']
name = kwargs['suggestedFilename']
print(f'下载:{url}')
self._page.download.add(url, goal_path=self._page.download_path, rename=name)
class Alert(object): class Alert(object):
"""用于保存alert信息的类""" """用于保存alert信息的类"""

View File

@ -7,6 +7,10 @@ from os import popen
from pathlib import Path from pathlib import Path
from typing import Union, Tuple, List from typing import Union, Tuple, List
from DownloadKit import DownloadKit
from requests import Session
from .configs.chromium_options import ChromiumOptions
from .chromium_base import ChromiumBase from .chromium_base import ChromiumBase
from .chromium_driver import ChromiumDriver from .chromium_driver import ChromiumDriver
from .chromium_tab import ChromiumTab from .chromium_tab import ChromiumTab
@ -25,6 +29,9 @@ class ChromiumPage(ChromiumBase):
self._main_tab: str = ... self._main_tab: str = ...
self._alert: Alert = ... self._alert: Alert = ...
self._download_path: str = ... self._download_path: str = ...
self._session: Session = ...
self._download_set: ChromiumDownloadSetter = ...
self._download_kit: DownloadKit = ...
def _connect_browser(self, def _connect_browser(self,
addr_driver_opts: Union[str, ChromiumDriver, DriverOptions] = None, addr_driver_opts: Union[str, ChromiumDriver, DriverOptions] = None,
@ -54,6 +61,9 @@ class ChromiumPage(ChromiumBase):
@property @property
def download_set(self) -> ChromiumDownloadSetter: ... def download_set(self) -> ChromiumDownloadSetter: ...
@property
def download(self) -> DownloadKit: ...
@property @property
def download_path(self) -> str: ... def download_path(self) -> str: ...
@ -85,6 +95,8 @@ class ChromiumPage(ChromiumBase):
def show_browser(self) -> None: ... def show_browser(self) -> None: ...
def cookies_to_session(self) -> None: ...
def quit(self) -> None: ... def quit(self) -> None: ...
def _on_alert_close(self, **kwargs): ... def _on_alert_close(self, **kwargs): ...
@ -95,8 +107,15 @@ class ChromiumPage(ChromiumBase):
class ChromiumDownloadSetter(object): class ChromiumDownloadSetter(object):
def __init__(self, page: ChromiumPage): def __init__(self, page: ChromiumPage):
self._page: ChromiumPage = ... self._page: ChromiumPage = ...
self._behavior: str = ...
def save_path(self, path: Union[str, Path]): ... def save_path(self, path) -> None: ...
def use_browser(self) -> None: ...
def use_DownloadKit(self) -> None: ...
def _download_by_DownloadKit(self, **kwargs) -> None: ...
class Alert(object): class Alert(object):

View File

@ -216,3 +216,24 @@ def cookies_to_tuple(cookies):
raise TypeError('cookies参数必须为RequestsCookieJar、list、tuple、str或dict类型。') raise TypeError('cookies参数必须为RequestsCookieJar、list、tuple、str或dict类型。')
return cookies return cookies
def set_session_cookies(session, cookies):
"""设置Session对象的cookies
:param session: Session对象
:param cookies: cookies信息
:return: None
"""
cookies = cookies_to_tuple(cookies)
for cookie in cookies:
if cookie['value'] is None:
cookie['value'] = ''
kwargs = {x: cookie[x] for x in cookie
if x.lower() in ('version', 'port', 'domain', 'path', 'secure',
'expires', 'discard', 'comment', 'comment_url', 'rest')}
if 'expiry' in cookie:
kwargs['expires'] = cookie['expiry']
session.cookies.set(cookie['name'], cookie['value'], **kwargs)

View File

@ -6,6 +6,7 @@
from http.cookiejar import Cookie from http.cookiejar import Cookie
from typing import Union from typing import Union
from requests import Session
from requests.cookies import RequestsCookieJar from requests.cookies import RequestsCookieJar
from DrissionPage.base import DrissionElement, BasePage from DrissionPage.base import DrissionElement, BasePage
@ -34,3 +35,6 @@ def cookie_to_dict(cookie: Union[Cookie, str, dict]) -> dict: ...
def cookies_to_tuple(cookies: Union[RequestsCookieJar, list, tuple, str, dict]) -> tuple: ... def cookies_to_tuple(cookies: Union[RequestsCookieJar, list, tuple, str, dict]) -> tuple: ...
def set_session_cookies(session: Session, cookies: Union[RequestsCookieJar, list, tuple, str, dict]) -> None: ...

View File

@ -14,7 +14,7 @@ from tldextract import extract
from .base import BasePage from .base import BasePage
from .configs.session_options import SessionOptions from .configs.session_options import SessionOptions
from .functions.web import cookies_to_tuple, cookie_to_dict from .functions.web import cookie_to_dict, set_session_cookies
from .session_element import SessionElement, make_session_ele from .session_element import SessionElement, make_session_ele
@ -71,23 +71,15 @@ class SessionPage(BasePage):
def _set_options(self): def _set_options(self):
"""设置WebPage中与d模式共用的配置便于WebPage覆盖掉""" """设置WebPage中与d模式共用的配置便于WebPage覆盖掉"""
self._timeouts = self._session_options.timeout self._timeout = self._session_options.timeout
self._download_path = self._session_options.download_path self._download_path = self._session_options.download_path
def set_cookies(self, cookies): def set_cookies(self, cookies):
cookies = cookies_to_tuple(cookies) """为Session对象设置cookies
for cookie in cookies: :param cookies: cookies信息
if cookie['value'] is None: :return: None
cookie['value'] = '' """
set_session_cookies(self.session, cookies)
kwargs = {x: cookie[x] for x in cookie
if x.lower() in ('version', 'port', 'domain', 'path', 'secure',
'expires', 'discard', 'comment', 'comment_url', 'rest')}
if 'expiry' in cookie:
kwargs['expires'] = cookie['expiry']
self.session.cookies.set(cookie['name'], cookie['value'], **kwargs)
def set_headers(self, headers): def set_headers(self, headers):
"""设置通用的headers设置的headers值回逐个覆盖原有的不会清理原来的 """设置通用的headers设置的headers值回逐个覆盖原有的不会清理原来的

View File

@ -154,7 +154,7 @@ class SessionPage(BasePage):
class DownloadSetter(object): class DownloadSetter(object):
def __init__(self, page: SessionPage): def __init__(self, page: BasePage):
self._page: SessionPage = ... self._page: SessionPage = ...
def save_path(self, path: Union[str, Path]): ... def save_path(self, path: Union[str, Path]): ...

View File

@ -487,10 +487,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
class WebPageDownloadSetter(ChromiumDownloadSetter): class WebPageDownloadSetter(ChromiumDownloadSetter):
"""用于设置下载参数的类""" """用于设置下载参数的类"""
def __init__(self, page):
super().__init__(page)
self._behavior = 'allow'
def save_path(self, path): def save_path(self, path):
"""设置下载路径 """设置下载路径
:param path: 下载路径 :param path: 下载路径
@ -501,6 +497,10 @@ class WebPageDownloadSetter(ChromiumDownloadSetter):
path.mkdir(parents=True, exist_ok=True) path.mkdir(parents=True, exist_ok=True)
path = str(path) path = str(path)
self._page._download_path = path self._page._download_path = path
if self._page._download_kit is not None:
self._page._download_kit.goal_path = path
if self._page._has_driver: if self._page._has_driver:
try: try:
self._page.run_cdp('Browser.setDownloadBehavior', behavior=self._behavior, downloadPath=path, self._page.run_cdp('Browser.setDownloadBehavior', behavior=self._behavior, downloadPath=path,
@ -525,11 +525,3 @@ class WebPageDownloadSetter(ChromiumDownloadSetter):
self._page.driver.Browser.downloadWillBegin = self._download_by_DownloadKit self._page.driver.Browser.downloadWillBegin = self._download_by_DownloadKit
self._page.driver.Browser.setDownloadBehavior(behavior='deny') self._page.driver.Browser.setDownloadBehavior(behavior='deny')
self._behavior = 'deny' self._behavior = 'deny'
def _download_by_DownloadKit(self, **kwargs):
gid = kwargs['guid']
self._page.run_cdp('Browser.cancelDownload', guid=gid, not_change=True)
url = kwargs['url']
name = kwargs['suggestedFilename']
print(f'下载:{url}')
self._page.download.add(url, goal_path=self._page.download_path, rename=name)