Compare commits

...

2 Commits

Author SHA1 Message Date
g1879
ecfa83dcf8 录像过程文件放到临时文件夹;微调参数 2024-01-03 00:06:23 +08:00
g1879
2986e3eeb1 4.0.0b33(+)
co增加tmp_path和is_auto_port属性;
auto_port在创建对象时才确定端口和路径;
auto_port的对象在浏览器关闭时情况用户文件夹
2024-01-02 22:51:40 +08:00
12 changed files with 86 additions and 47 deletions

View File

@ -13,4 +13,4 @@ from ._configs.chromium_options import ChromiumOptions
from ._configs.session_options import SessionOptions
__all__ = ['ChromiumPage', 'ChromiumOptions', 'SessionOptions', 'SessionPage', 'WebPage', '__version__']
__version__ = '4.0.0b32'
__version__ = '4.0.0b33'

View File

@ -3,6 +3,8 @@
@Author : g1879
@Contact : g1879@qq.com
"""
from pathlib import Path
from shutil import rmtree
from time import sleep, perf_counter
from .driver import BrowserDriver, Driver
@ -196,3 +198,14 @@ class Browser(object):
def _on_quit(self):
Browser.BROWSERS.pop(self.id, None)
if self.page._chromium_options.is_auto_port and self.page._chromium_options.user_data_path:
path = Path(self.page._chromium_options.user_data_path)
end_time = perf_counter() + 7
while perf_counter() < end_time:
if not path.exists():
break
try:
rmtree(path)
break
except (PermissionError, FileNotFoundError):
pass

View File

@ -59,15 +59,15 @@ class Driver(object):
message['id'] = ws_id
message_json = dumps(message)
if self._debug:
if self._debug is True or (isinstance(self._debug, str) and
message.get('method', '').startswith(self._debug)):
print(f'发> {message_json}')
elif isinstance(self._debug, (list, tuple, set)):
for m in self._debug:
if message.get('method', '').startswith(m):
print(f'发> {message_json}')
break
# if self._debug:
# if self._debug is True or (isinstance(self._debug, str) and
# message.get('method', '').startswith(self._debug)):
# print(f'发> {message_json}')
# elif isinstance(self._debug, (list, tuple, set)):
# for m in self._debug:
# if message.get('method', '').startswith(m):
# print(f'发> {message_json}')
# break
end_time = perf_counter() + timeout if timeout is not None else None
self.method_results[ws_id] = Queue()
@ -113,15 +113,15 @@ class Driver(object):
self._stop()
return
if self._debug:
if self._debug is True or 'id' in msg or (isinstance(self._debug, str)
and msg.get('method', '').startswith(self._debug)):
print(f'<收 {msg_json}')
elif isinstance(self._debug, (list, tuple, set)):
for m in self._debug:
if msg.get('method', '').startswith(m):
print(f'<收 {msg_json}')
break
# if self._debug:
# if self._debug is True or 'id' in msg or (isinstance(self._debug, str)
# and msg.get('method', '').startswith(self._debug)):
# print(f'<收 {msg_json}')
# elif isinstance(self._debug, (list, tuple, set)):
# for m in self._debug:
# if msg.get('method', '').startswith(m):
# print(f'<收 {msg_json}')
# break
if 'method' in msg:
if msg['method'].startswith('Page.javascriptDialog'):
@ -135,8 +135,8 @@ class Driver(object):
elif msg.get('id') in self.method_results:
self.method_results[msg['id']].put(msg)
elif self._debug:
print(f'未知信息:{msg}')
# elif self._debug:
# print(f'未知信息:{msg}')
def _handle_event_loop(self):
"""当接收到浏览器信息,执行已绑定的方法"""
@ -266,6 +266,6 @@ class BrowserDriver(Driver):
r.close()
return r
def stop(self):
super().stop()
def _stop(self):
super()._stop()
self.browser._on_quit()

View File

@ -106,6 +106,11 @@ class ChromiumOptions(object):
"""返回用户数据文件夹路径"""
return self._user_data_path
@property
def tmp_path(self):
"""返回临时文件夹路径"""
return self._tmp_path
@property
def user(self):
"""返回用户配置文件夹名称"""
@ -161,6 +166,11 @@ class ChromiumOptions(object):
"""返回是否只接管现有浏览器方式"""
return self._existing_only
@property
def is_auto_port(self):
"""返回是否使用自动端口和用户文件"""
return self._auto_port
@property
def retry_times(self):
"""返回连接失败时的重试次数"""
@ -485,14 +495,13 @@ class ChromiumOptions(object):
def auto_port(self, on_off=True, tmp_path=None):
"""自动获取可用端口
:param on_off: 是否开启自动获取端口号
:param tmp_path: 临时文件保存路径为None时保存到系统临时文件夹
:param tmp_path: 临时文件保存路径为None时保存到系统临时文件夹on_off为False时此参数无效
:return: 当前对象
"""
if on_off:
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
if tmp_path:
self._tmp_path = str(tmp_path)
else:
self._auto_port = False
return self
@ -618,7 +627,8 @@ class PortFinder(object):
"""
:param path: 临时文件保存路径为None时使用系统临时文件夹
"""
self.tmp_dir = Path(path) if path else Path(gettempdir()) / 'DrissionPage' / 'UserTempFolder'
tmp = Path(path) if path else Path(gettempdir()) / 'DrissionPage'
self.tmp_dir = tmp / 'UserTempFolder'
self.tmp_dir.mkdir(parents=True, exist_ok=True)
if not PortFinder.used_port:
clean_folder(self.tmp_dir)

View File

@ -5,7 +5,7 @@
"""
from pathlib import Path
from threading import Lock
from typing import Union, Tuple, Any, Literal
from typing import Union, Tuple, Any, Literal, Optional
class ChromiumOptions(object):
@ -43,6 +43,9 @@ class ChromiumOptions(object):
@property
def user_data_path(self) -> str: ...
@property
def tmp_path(self) -> Optional[str]: ...
@property
def user(self) -> str: ...
@ -76,6 +79,9 @@ class ChromiumOptions(object):
@property
def is_existing_only(self) -> bool: ...
@property
def is_auto_port(self) -> bool: ...
@property
def retry_times(self) -> int: ...

View File

@ -753,29 +753,29 @@ class ChromiumBase(BasePage):
return self._get_screenshot(path=path, name=name, as_bytes=as_bytes, as_base64=as_base64,
full_page=full_page, left_top=left_top, right_bottom=right_bottom)
def add_init_js(self, js):
def add_init_js(self, script):
"""添加初始化脚本,在页面加载任何脚本前执行
:param js: js文本
:param script: js文本
:return: 添加的脚本的id
"""
js_id = self.run_cdp('Page.addScriptToEvaluateOnNewDocument', source=js,
js_id = self.run_cdp('Page.addScriptToEvaluateOnNewDocument', source=script,
includeCommandLineAPI=True)['identifier']
self._init_jss.append(js_id)
return js_id
def remove_init_js(self, js_id=None):
def remove_init_js(self, script_id=None):
"""删除初始化脚本js_id传入None时删除所有
:param js_id: 脚本的id
:param script_id: 脚本的id
:return: None
"""
if js_id is None:
if script_id is None:
for js_id in self._init_jss:
self.run_cdp('Page.removeScriptToEvaluateOnNewDocument', identifier=js_id)
self._init_jss.clear()
elif js_id in self._init_jss:
self.run_cdp('Page.removeScriptToEvaluateOnNewDocument', identifier=js_id)
self._init_jss.remove(js_id)
elif script_id in self._init_jss:
self.run_cdp('Page.removeScriptToEvaluateOnNewDocument', identifier=script_id)
self._init_jss.remove(script_id)
def clear_cache(self, session_storage=True, local_storage=True, cache=True, cookies=True):
"""清除缓存,可选要清除的项

View File

@ -214,9 +214,9 @@ class ChromiumBase(BasePage):
def get_local_storage(self, item: str = None) -> Union[str, dict, None]: ...
def add_init_js(self, js: str) -> str: ...
def add_init_js(self, script: str) -> str: ...
def remove_init_js(self, js_id: str = None) -> None: ...
def remove_init_js(self, script_id: str = None) -> None: ...
def get_screenshot(self, path: [str, Path] = None, name: str = None, as_bytes: PIC_TYPE = None,
as_base64: PIC_TYPE = None, full_page: bool = False, left_top: Tuple[int, int] = None,

View File

@ -105,7 +105,7 @@ class ChromiumFrame(ChromiumBase):
def _reload(self):
"""重新获取document"""
self._is_loading = True
d_debug = self.driver._debug
# d_debug = self.driver._debug
self._reloading = True
self._doc_got = False
@ -131,7 +131,7 @@ class ChromiumFrame(ChromiumBase):
if self._listener:
self._listener._to_target(self._target_page.tab_id, self.address, self)
super().__init__(self.address, self._target_page.tab_id, self._target_page.timeout)
self.driver._debug = d_debug
# self.driver._debug = d_debug
else:
self._is_diff_domain = True
@ -154,7 +154,8 @@ class ChromiumFrame(ChromiumBase):
# print(f'获取doc失败重试 {e}')
# else:
# raise GetDocumentError
self.driver._debug = d_debug
# self.driver._debug = d_debug
self._is_loading = False
self._reloading = False

View File

@ -10,7 +10,7 @@ from requests import get
from .._base.browser import Browser
from .._functions.browser import connect_browser
from .._configs.chromium_options import ChromiumOptions
from .._configs.chromium_options import ChromiumOptions, PortFinder
from .._pages.chromium_base import ChromiumBase, get_mhtml, Timeout
from .._pages.chromium_tab import ChromiumTab
from .._units.setter import ChromiumPageSetter
@ -44,6 +44,11 @@ class ChromiumPage(ChromiumBase):
self._chromium_options = ChromiumOptions(addr_or_opts)
elif isinstance(addr_or_opts, ChromiumOptions):
if addr_or_opts.is_auto_port:
port, path = PortFinder(addr_or_opts.tmp_path).get_port()
addr_or_opts.set_address(f'127.0.0.1:{port}')
addr_or_opts.set_user_data_path(path)
addr_or_opts.auto_port()
self._chromium_options = addr_or_opts
elif isinstance(addr_or_opts, str):

View File

@ -283,7 +283,6 @@ class Actions:
if character in ('\ue009', '\ue008', '\ue00a', '\ue03d'):
modifiers.append(character)
else:
sleep(.01)
self.key_up(character)
for m in modifiers:
self.key_up(m)

View File

@ -8,6 +8,7 @@ from os.path import sep
from pathlib import Path
from random import randint
from shutil import rmtree
from tempfile import gettempdir
from threading import Thread
from time import sleep, time
@ -36,7 +37,11 @@ class Screencast(object):
raise ValueError('save_path必须设置。')
if self._mode in ('frugal_video', 'video'):
self._tmp_path = self._path / f'screencast_tmp_{time()}_{randint(0, 100)}'
if self._page.browser.page._chromium_options.tmp_path:
self._tmp_path = Path(
self._page.browser.page._chromium_options.tmp_path) / f'screencast_tmp_{time()}_{randint(0, 100)}'
else:
self._tmp_path = Path(gettempdir()) / 'DrissionPage' / f'screencast_tmp_{time()}_{randint(0, 100)}'
self._tmp_path.mkdir(parents=True, exist_ok=True)
if self._mode.startswith('frugal'):

View File

@ -6,7 +6,7 @@ with open("README.md", "r", encoding='utf-8') as fh:
setup(
name="DrissionPage",
version="4.0.0b32",
version="4.0.0b33",
author="g1879",
author_email="g1879@qq.com",
description="Python based web automation tool. It can control the browser and send and receive data packets.",