diff --git a/DrissionPage/chromium_base.py b/DrissionPage/chromium_base.py index 6e08aba..b07db5f 100644 --- a/DrissionPage/chromium_base.py +++ b/DrissionPage/chromium_base.py @@ -64,6 +64,7 @@ class ChromiumBase(BasePage): def _set_runtime_settings(self): self._timeouts = Timeout(self) self._page_load_strategy = 'normal' + self._wait_download_flag = None def _connect_browser(self, tab_id=None): """连接浏览器,在第一次时运行 @@ -244,7 +245,10 @@ class ChromiumBase(BasePage): self._upload_list = None def _onDownloadWillBegin(self, **kwargs): + if self._wait_download_flag is False: + self._page.run_cdp('Browser.cancelDownload', guid=kwargs['guid']) self._page._dl_mgr.add_mission(kwargs['guid'], self.download_path, kwargs['suggestedFilename']) + self._wait_download_flag = {'url': kwargs['url'], 'name': kwargs['suggestedFilename']} def __call__(self, loc_or_str, timeout=None): """在内部查找元素 diff --git a/DrissionPage/chromium_base.pyi b/DrissionPage/chromium_base.pyi index 94463f1..433e7a2 100644 --- a/DrissionPage/chromium_base.pyi +++ b/DrissionPage/chromium_base.pyi @@ -45,6 +45,7 @@ class ChromiumBase(BasePage): self._set: ChromiumBaseSetter = ... self._screencast: Screencast = ... self._listener: NetworkListener = ... + self._wait_download_flag: bool = ... def _connect_browser(self, tab_id: str = None) -> None: ... diff --git a/DrissionPage/chromium_page.py b/DrissionPage/chromium_page.py index 8d7e52b..9700c26 100644 --- a/DrissionPage/chromium_page.py +++ b/DrissionPage/chromium_page.py @@ -10,6 +10,7 @@ from .chromium_base import ChromiumBase, Timeout from .chromium_driver import ChromiumDriver from .chromium_tab import ChromiumTab from .commons.browser import connect_browser +from .commons.tools import get_usable_path from .configs.chromium_options import ChromiumOptions from .errors import BrowserConnectError from .setter import ChromiumPageSetter @@ -448,17 +449,16 @@ class BrowserDownloadManager(object): self._missions = {} def add_mission(self, guid, path, name): - print(name) self._missions[guid] = {'path': path, 'name': name} def _onDownloadProgress(self, **kwargs): - # todo: 处理同名文件、处理后缀 if kwargs['state'] == 'completed' and kwargs['guid'] in self._missions: guid = kwargs['guid'] path = self._missions[guid]['path'] name = self._missions[guid]['name'] form_path = f'{self._page.download_path}\\{guid}' - to_path = f'{path}\\{name}' + to_path = get_usable_path(f'{path}\\{name}') + move(form_path, to_path) self._missions.pop(guid) diff --git a/DrissionPage/waiter.py b/DrissionPage/waiter.py index 0df8fb0..89cddee 100644 --- a/DrissionPage/waiter.py +++ b/DrissionPage/waiter.py @@ -78,6 +78,26 @@ class ChromiumBaseWaiter(object): while self._driver._upload_list: sleep(.01) + def browser_download_begin(self, timeout=None, cancel=False): + """等待浏览器下载开始,可将其拦截 + :param timeout: 超时时间,None使用页面对象超时时间 + :param cancel: 是否取消该任务 + :return: 成功返回任务信息dict,失败返回False + """ + self._driver._wait_download_flag = False if cancel else True + if timeout is None: + timeout = self._driver.timeout + + r = False + end_time = perf_counter() + timeout + while perf_counter() < end_time: + if not isinstance(self._driver._wait_download_flag, bool): + r = self._driver._wait_download_flag + break + + self._driver._wait_download_flag = None + return r + def url_change(self, text, exclude=False, timeout=None, raise_err=None): """等待url变成包含或不包含指定文本 :param text: 用于识别的文本 @@ -173,6 +193,10 @@ class ChromiumPageWaiter(ChromiumBaseWaiter): else: return False + def browser_downloads_complete(self): + """等待所有下载任务结束""" + pass + class ChromiumElementWaiter(object): """等待元素在dom中某种状态,如删除、显示、隐藏""" diff --git a/DrissionPage/waiter.pyi b/DrissionPage/waiter.pyi index 548c167..2a659ce 100644 --- a/DrissionPage/waiter.pyi +++ b/DrissionPage/waiter.pyi @@ -3,7 +3,7 @@ @Author : g1879 @Contact : g1879@qq.com """ -from typing import Union +from typing import Union, Optional from .chromium_base import ChromiumBase from .chromium_element import ChromiumElement @@ -35,6 +35,8 @@ class ChromiumBaseWaiter(object): def upload_paths_inputted(self) -> None: ... + def browser_download_begin(self, timeout: float = None, cancel: bool = False) -> Union[dict, bool]: ... + def url_change(self, text: str, exclude: bool = False, timeout: float = None, raise_err: bool = None) -> bool: ... def title_change(self, text: str, exclude: bool = False, timeout: float = None, raise_err: bool = None) -> bool: ...