可指定临时文件夹(+)

auto_port()增加tmp_path参数;
ini增加tmp_path项;
co增加set_tmp_path()方法
This commit is contained in:
g1879 2023-12-31 11:02:27 +08:00
parent c0f50e2bbf
commit d1a87654a1
4 changed files with 34 additions and 11 deletions

View File

@ -32,6 +32,7 @@ class ChromiumOptions(object):
options = om.chromium_options
self._download_path = om.paths.get('download_path', None) or None
self._tmp_path = om.paths.get('tmp_path', None) or None
self._arguments = options.get('arguments', [])
self._browser_path = options.get('browser_path', '')
self._extensions = options.get('extensions', [])
@ -76,6 +77,7 @@ class ChromiumOptions(object):
self._browser_path = "chrome"
self._arguments = []
self._download_path = None
self._tmp_path = None
self._extensions = []
self._prefs = {}
self._flags = {}
@ -445,6 +447,14 @@ class ChromiumOptions(object):
self._download_path = str(path)
return self
def set_tmp_path(self, path):
"""设置临时文件文件保存路径
:param path: 下载路径
:return: 当前对象
"""
self._tmp_path = str(path)
return self
def set_user_data_path(self, path):
"""设置用户文件夹路径
:param path: 用户文件夹路径
@ -472,13 +482,15 @@ class ChromiumOptions(object):
self._system_user_path = on_off
return self
def auto_port(self, on_off=True):
def auto_port(self, on_off=True, tmp_path=None):
"""自动获取可用端口
:param on_off: 是否开启自动获取端口号
:param tmp_path: 临时文件保存路径为None时保存到系统临时文件夹
:return: 当前对象
"""
if on_off:
port, path = PortFinder().get_port()
tmp_path = tmp_path or self._tmp_path
port, path = PortFinder(tmp_path).get_port()
self.set_paths(local_port=port, user_data_path=path)
self._auto_port = True
else:
@ -527,6 +539,7 @@ class ChromiumOptions(object):
om.set_item('proxies', 'https', self._proxy)
# 设置路径
om.set_item('paths', 'download_path', self._download_path or '')
om.set_item('paths', 'tmp_path', self._tmp_path or '')
# 设置timeout
om.set_item('timeouts', 'base', self._timeouts['base'])
om.set_item('timeouts', 'page_load', self._timeouts['pageLoad'])
@ -546,6 +559,9 @@ class ChromiumOptions(object):
"""保存当前配置到默认ini文件"""
return self.save('default')
def __repr__(self):
return f'<ChromiumOptions at {id(self)}>'
# ---------------即将废弃--------------
@property
@ -593,16 +609,16 @@ class ChromiumOptions(object):
on_off = None if on_off else False
return self.set_argument('--mute-audio', on_off)
def __repr__(self):
return f'<ChromiumOptions at {id(self)}>'
class PortFinder(object):
used_port = {}
lock = Lock()
def __init__(self):
self.tmp_dir = Path(gettempdir()) / 'DrissionPage' / 'TempFolder'
def __init__(self, path=None):
"""
:param path: 临时文件保存路径为None时使用系统临时文件夹
"""
self.tmp_dir = Path(path) if path else Path(gettempdir()) / 'DrissionPage' / 'UserTempFolder'
self.tmp_dir.mkdir(parents=True, exist_ok=True)
if not PortFinder.used_port:
clean_folder(self.tmp_dir)

View File

@ -14,6 +14,7 @@ class ChromiumOptions(object):
self._driver_path: str = ...
self._user_data_path: str = ...
self._download_path: str = ...
self._tmp_path: str = ...
self._arguments: list = ...
self._browser_path: str = ...
self._user: str = ...
@ -132,6 +133,8 @@ class ChromiumOptions(object):
def set_download_path(self, path: Union[str, Path]) -> ChromiumOptions: ...
def set_tmp_path(self, path: Union[str, Path]) -> ChromiumOptions: ...
def set_user_data_path(self, path: Union[str, Path]) -> ChromiumOptions: ...
def set_cache_path(self, path: Union[str, Path]) -> ChromiumOptions: ...
@ -142,7 +145,7 @@ class ChromiumOptions(object):
def use_system_user_path(self, on_off: bool = True) -> ChromiumOptions: ...
def auto_port(self, on_off: bool = True) -> ChromiumOptions: ...
def auto_port(self, on_off: bool = True, tmp_path: Union[str, Path] = None) -> ChromiumOptions: ...
def existing_only(self, on_off: bool = True) -> ChromiumOptions: ...
@ -154,6 +157,9 @@ class ChromiumOptions(object):
class PortFinder(object):
used_port: dict = ...
lock: Lock = ...
tmp_dir: Path = ...
def __init__(self, path: Union[str, Path] = None): ...
@staticmethod
def get_port() -> Tuple[int, str]: ...

View File

@ -1,5 +1,6 @@
[paths]
download_path =
download_path =
tmp_path =
[chromium_options]
address = 127.0.0.1:9222

View File

@ -65,12 +65,12 @@ class ChromiumPage(ChromiumBase):
try:
ws = get(f'http://{self._chromium_options.address}/json/version', headers={'Connection': 'close'})
if not ws:
raise BrowserConnectError('\n浏览器连接失败,请检查是否启用全局代理。如是须设置不代理127.0.0.1地址。')
raise BrowserConnectError('\n浏览器连接失败,如使用全局代理须设置不代理127.0.0.1地址。')
ws = ws.json()['webSocketDebuggerUrl'].split('/')[-1]
except KeyError:
raise BrowserConnectError('浏览器版本太旧,请升级。')
except:
raise BrowserConnectError('\n浏览器连接失败,请检查是否启用全局代理。如是须设置不代理127.0.0.1地址。')
raise BrowserConnectError('\n浏览器连接失败,如使用全局代理须设置不代理127.0.0.1地址。')
self._browser = Browser(self._chromium_options.address, ws, self)
if (is_exist and self._chromium_options._headless is False and