mirror of
https://gitee.com/g1879/DrissionPage.git
synced 2024-11-29 18:47:34 +08:00
4.0.3.3(+)
修复set_browser_path()导致auto_port()失效问题; 重构wait_until(); ini文件增加'--disable-features=PrivacySandboxSettings4'; get_blob()参数改为as_bytes
This commit is contained in:
parent
aea5aace53
commit
4db2f71d15
@ -14,4 +14,4 @@ from ._configs.chromium_options import ChromiumOptions
|
||||
from ._configs.session_options import SessionOptions
|
||||
|
||||
__all__ = ['ChromiumPage', 'ChromiumOptions', 'SessionOptions', 'SessionPage', 'WebPage', '__version__']
|
||||
__version__ = '4.0.3.2'
|
||||
__version__ = '4.0.3.3'
|
||||
|
@ -448,7 +448,6 @@ class ChromiumOptions(object):
|
||||
:return: 当前对象
|
||||
"""
|
||||
self._browser_path = str(path)
|
||||
self._auto_port = False
|
||||
return self
|
||||
|
||||
def set_download_path(self, path):
|
||||
|
@ -5,7 +5,7 @@ tmp_path =
|
||||
[chromium_options]
|
||||
address = 127.0.0.1:9222
|
||||
browser_path = chrome
|
||||
arguments = ['--no-default-browser-check', '--disable-suggestions-ui', '--no-first-run', '--disable-infobars', '--disable-popup-blocking', '--hide-crash-restore-bubble']
|
||||
arguments = ['--no-default-browser-check', '--disable-suggestions-ui', '--no-first-run', '--disable-infobars', '--disable-popup-blocking', '--hide-crash-restore-bubble', '--disable-features=PrivacySandboxSettings4']
|
||||
extensions = []
|
||||
prefs = {'profile.default_content_settings.popups': 0, 'profile.default_content_setting_values': {'notifications': 2}}
|
||||
flags = {}
|
||||
|
@ -122,40 +122,21 @@ def get_chrome_hwnds_from_pid(pid, title):
|
||||
return hwnds
|
||||
|
||||
|
||||
def wait_until(page, condition, timeout=10, poll=0.1, raise_err=True):
|
||||
"""等待返回值不为False或空,直到超时
|
||||
:param page: DrissionPage对象
|
||||
:param condition: 等待条件,返回值不为False则停止等待
|
||||
def wait_until(function, kwargs=None, timeout=10):
|
||||
"""等待传入的方法返回值不为假
|
||||
:param function: 要执行的方法
|
||||
:param kwargs: 方法参数
|
||||
:param timeout: 超时时间(秒)
|
||||
:param poll: 轮询间隔
|
||||
:param raise_err: 是否抛出异常
|
||||
:return: DP Element or bool
|
||||
:return: 执行结果,超时抛出TimeoutError
|
||||
"""
|
||||
if kwargs is None:
|
||||
kwargs = {}
|
||||
end_time = perf_counter() + timeout
|
||||
if isinstance(condition, str) or isinstance(condition, tuple):
|
||||
if not callable(getattr(page, 's_ele', None)):
|
||||
raise AttributeError('page对象缺少s_ele方法')
|
||||
condition_method = lambda page: page.s_ele(condition)
|
||||
elif callable(condition):
|
||||
condition_method = condition
|
||||
else:
|
||||
raise ValueError('condition必须是函数或者字符串或者元组')
|
||||
while perf_counter() < end_time:
|
||||
try:
|
||||
value = condition_method(page)
|
||||
if value:
|
||||
return value
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
sleep(poll)
|
||||
if perf_counter() > end_time:
|
||||
break
|
||||
|
||||
if raise_err:
|
||||
raise TimeoutError(f'等待超时(等待{timeout}秒)。')
|
||||
else:
|
||||
return False
|
||||
value = function(**kwargs)
|
||||
if value:
|
||||
return value
|
||||
raise TimeoutError
|
||||
|
||||
|
||||
def stop_process_on_port(port):
|
||||
|
@ -28,7 +28,7 @@ def get_browser_progress_id(progress: Union[popen, None], address: str) -> Union
|
||||
def get_chrome_hwnds_from_pid(pid: Union[str, int], title: str) -> list: ...
|
||||
|
||||
|
||||
def wait_until(page, condition: Union[FunctionType, str, tuple], timeout: float, poll: float, raise_err: bool): ...
|
||||
def wait_until(function: callable, kwargs: dict = None, timeout: float = 10): ...
|
||||
|
||||
|
||||
def stop_process_on_port(port: Union[int, str]) -> None: ...
|
||||
|
@ -330,7 +330,13 @@ def is_cookie_in_driver(page, cookie):
|
||||
return False
|
||||
|
||||
|
||||
def get_blob(page, url, base64_to_bytes=True):
|
||||
def get_blob(page, url, as_bytes=True):
|
||||
"""获取知道blob资源
|
||||
:param page: 资源所在页面对象
|
||||
:param url: 资源url
|
||||
:param as_bytes: 是否以字节形式返回
|
||||
:return: 资源内容
|
||||
"""
|
||||
if not url.startswith('blob'):
|
||||
raise TypeError('该链接非blob类型。')
|
||||
js = """
|
||||
@ -352,6 +358,6 @@ def get_blob(page, url, base64_to_bytes=True):
|
||||
result = page.run_js(js, url)
|
||||
except:
|
||||
raise RuntimeError('无法获取该资源。')
|
||||
if base64_to_bytes:
|
||||
if as_bytes:
|
||||
from base64 import b64decode
|
||||
return b64decode(result.split(',', 1)[-1])
|
||||
|
@ -49,4 +49,4 @@ def set_browser_cookies(page: ChromiumBase, cookies: Union[RequestsCookieJar, li
|
||||
def is_cookie_in_driver(page: ChromiumBase, cookie: dict) -> bool: ...
|
||||
|
||||
|
||||
def get_blob(page: ChromiumBase, url: str, base64_to_bytes: bool = True) -> bytes: ...
|
||||
def get_blob(page: ChromiumBase, url: str, as_bytes: bool = True) -> bytes: ...
|
||||
|
Loading…
Reference in New Issue
Block a user