4.0.0b33(+)

co增加tmp_path和is_auto_port属性;
auto_port在创建对象时才确定端口和路径;
auto_port的对象在浏览器关闭时情况用户文件夹
This commit is contained in:
g1879 2024-01-02 22:51:40 +08:00
parent a20fafebd7
commit 2986e3eeb1
9 changed files with 69 additions and 35 deletions

View File

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

View File

@ -3,6 +3,8 @@
@Author : g1879 @Author : g1879
@Contact : g1879@qq.com @Contact : g1879@qq.com
""" """
from pathlib import Path
from shutil import rmtree
from time import sleep, perf_counter from time import sleep, perf_counter
from .driver import BrowserDriver, Driver from .driver import BrowserDriver, Driver
@ -196,3 +198,14 @@ class Browser(object):
def _on_quit(self): def _on_quit(self):
Browser.BROWSERS.pop(self.id, None) 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['id'] = ws_id
message_json = dumps(message) message_json = dumps(message)
if self._debug: # if self._debug:
if self._debug is True or (isinstance(self._debug, str) and # if self._debug is True or (isinstance(self._debug, str) and
message.get('method', '').startswith(self._debug)): # message.get('method', '').startswith(self._debug)):
print(f'发> {message_json}') # print(f'发> {message_json}')
elif isinstance(self._debug, (list, tuple, set)): # elif isinstance(self._debug, (list, tuple, set)):
for m in self._debug: # for m in self._debug:
if message.get('method', '').startswith(m): # if message.get('method', '').startswith(m):
print(f'发> {message_json}') # print(f'发> {message_json}')
break # break
end_time = perf_counter() + timeout if timeout is not None else None end_time = perf_counter() + timeout if timeout is not None else None
self.method_results[ws_id] = Queue() self.method_results[ws_id] = Queue()
@ -113,15 +113,15 @@ class Driver(object):
self._stop() self._stop()
return return
if self._debug: # if self._debug:
if self._debug is True or 'id' in msg or (isinstance(self._debug, str) # if self._debug is True or 'id' in msg or (isinstance(self._debug, str)
and msg.get('method', '').startswith(self._debug)): # and msg.get('method', '').startswith(self._debug)):
print(f'<收 {msg_json}') # print(f'<收 {msg_json}')
elif isinstance(self._debug, (list, tuple, set)): # elif isinstance(self._debug, (list, tuple, set)):
for m in self._debug: # for m in self._debug:
if msg.get('method', '').startswith(m): # if msg.get('method', '').startswith(m):
print(f'<收 {msg_json}') # print(f'<收 {msg_json}')
break # break
if 'method' in msg: if 'method' in msg:
if msg['method'].startswith('Page.javascriptDialog'): if msg['method'].startswith('Page.javascriptDialog'):
@ -135,8 +135,8 @@ class Driver(object):
elif msg.get('id') in self.method_results: elif msg.get('id') in self.method_results:
self.method_results[msg['id']].put(msg) self.method_results[msg['id']].put(msg)
elif self._debug: # elif self._debug:
print(f'未知信息:{msg}') # print(f'未知信息:{msg}')
def _handle_event_loop(self): def _handle_event_loop(self):
"""当接收到浏览器信息,执行已绑定的方法""" """当接收到浏览器信息,执行已绑定的方法"""
@ -266,6 +266,6 @@ class BrowserDriver(Driver):
r.close() r.close()
return r return r
def stop(self): def _stop(self):
super().stop() super()._stop()
self.browser._on_quit() self.browser._on_quit()

View File

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

View File

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

View File

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

View File

@ -10,7 +10,7 @@ from requests import get
from .._base.browser import Browser from .._base.browser import Browser
from .._functions.browser import connect_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_base import ChromiumBase, get_mhtml, Timeout
from .._pages.chromium_tab import ChromiumTab from .._pages.chromium_tab import ChromiumTab
from .._units.setter import ChromiumPageSetter from .._units.setter import ChromiumPageSetter
@ -44,6 +44,11 @@ class ChromiumPage(ChromiumBase):
self._chromium_options = ChromiumOptions(addr_or_opts) self._chromium_options = ChromiumOptions(addr_or_opts)
elif isinstance(addr_or_opts, ChromiumOptions): 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 self._chromium_options = addr_or_opts
elif isinstance(addr_or_opts, str): elif isinstance(addr_or_opts, str):

View File

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

View File

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