mirror of
https://gitee.com/g1879/DrissionPage.git
synced 2024-12-10 04:00:23 +08:00
4.0.0b6 quit()增加超时和强制关闭参数
This commit is contained in:
g1879
parent
d24730cc2d
commit
89e1b3a29f
@ -3,9 +3,10 @@
|
||||
@Author : g1879
|
||||
@Contact : g1879@qq.com
|
||||
"""
|
||||
from time import sleep
|
||||
from time import sleep, perf_counter
|
||||
|
||||
from .chromium_driver import BrowserDriver, ChromiumDriver
|
||||
from .._commons.tools import stop_process_on_port
|
||||
from .._units.download_manager import DownloadManager
|
||||
|
||||
|
||||
@ -151,8 +152,12 @@ class Browser(object):
|
||||
"""
|
||||
return self.run_cdp('Browser.getWindowForTarget', targetId=tab_id or self.id)['bounds']
|
||||
|
||||
def quit(self):
|
||||
"""关闭浏览器"""
|
||||
def quit(self, timeout=5, force=True):
|
||||
"""关闭浏览器
|
||||
:param timeout: 等待浏览器关闭超时时间
|
||||
:param force: 关闭超时是否强制终止进程
|
||||
:return: None
|
||||
"""
|
||||
self.run_cdp('Browser.close')
|
||||
self.driver.stop()
|
||||
|
||||
@ -161,11 +166,18 @@ class Browser(object):
|
||||
from platform import system
|
||||
txt = f'tasklist | findstr {self.process_id}' if system().lower() == 'windows' \
|
||||
else f'ps -ef | grep {self.process_id}'
|
||||
while True:
|
||||
end_time = perf_counter() + timeout
|
||||
while perf_counter() < end_time:
|
||||
p = popen(txt)
|
||||
if f' {self.process_id} ' not in p.read():
|
||||
break
|
||||
return
|
||||
sleep(.2)
|
||||
|
||||
if force:
|
||||
ip, port = self.address.split(':')
|
||||
if ip not in ('127.0.0.1', 'localhost'):
|
||||
return
|
||||
stop_process_on_port(port)
|
||||
|
||||
def _on_quit(self):
|
||||
Browser.BROWSERS.pop(self.id, None)
|
||||
|
||||
@ -26,7 +26,7 @@ class Browser(object):
|
||||
|
||||
def __init__(self, address: str, browser_id: str, page: ChromiumPage): ...
|
||||
|
||||
def _get_driver(self, tab_id: str)->ChromiumDriver: ...
|
||||
def _get_driver(self, tab_id: str) -> ChromiumDriver: ...
|
||||
|
||||
def run_cdp(self, cmd, **cmd_args) -> dict: ...
|
||||
|
||||
@ -57,6 +57,6 @@ class Browser(object):
|
||||
|
||||
def _onTargetDestroyed(self, **kwargs) -> None: ...
|
||||
|
||||
def quit(self) -> None: ...
|
||||
def quit(self, timeout: float = 5, force: bool = True) -> None: ...
|
||||
|
||||
def _on_quit(self) -> None: ...
|
||||
|
||||
@ -9,6 +9,8 @@ from re import search, sub
|
||||
from shutil import rmtree
|
||||
from time import perf_counter, sleep
|
||||
|
||||
from psutil import process_iter, AccessDenied, NoSuchProcess, ZombieProcess
|
||||
|
||||
|
||||
def get_usable_path(path, is_file=True, parents=True):
|
||||
"""检查文件或文件夹是否有重名，并返回可以使用的路径
|
||||
@ -216,3 +218,23 @@ def wait_until(page, condition, timeout=10, poll=0.1, raise_err=True):
|
||||
raise TimeoutError('等待超时')
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
def stop_process_on_port(port):
|
||||
"""强制关闭某个端口内的进程
|
||||
:param port: 端口号
|
||||
:return: None
|
||||
"""
|
||||
for proc in process_iter(['pid', 'connections']):
|
||||
try:
|
||||
connections = proc.connections()
|
||||
except AccessDenied:
|
||||
continue
|
||||
for conn in connections:
|
||||
if conn.laddr.port == int(port):
|
||||
try:
|
||||
proc.terminate()
|
||||
except (NoSuchProcess, AccessDenied, ZombieProcess):
|
||||
pass
|
||||
except Exception as e:
|
||||
print(f"{proc.pid} {port}: {e}")
|
||||
|
||||
@ -36,3 +36,6 @@ 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 stop_process_on_port(port: Union[int, str]) -> None: ...
|
||||
|
||||
@ -157,9 +157,21 @@ class ChromiumPage(ChromiumBase):
|
||||
:param new_context: 是否创建新的上下文
|
||||
:return: 新标签页对象
|
||||
"""
|
||||
tab = ChromiumTab(self, tab_id=self._new_tab(new_window, background, new_context))
|
||||
if url:
|
||||
tab.get(url)
|
||||
return tab
|
||||
|
||||
def _new_tab(self, new_window=False, background=False, new_context=False):
|
||||
"""新建一个标签页
|
||||
:param new_window: 是否在新窗口打开标签页
|
||||
:param background: 是否不激活新标签页，如new_window为True则无效
|
||||
:param new_context: 是否创建新的上下文
|
||||
:return: 新标签页对象
|
||||
"""
|
||||
bid = None
|
||||
if new_context:
|
||||
bid = self.browser.run_cdp('Target.createBrowserContext', **kwargs)['browserContextId']
|
||||
bid = self.browser.run_cdp('Target.createBrowserContext')['browserContextId']
|
||||
|
||||
kwargs = {'url': ''}
|
||||
if new_window:
|
||||
@ -169,11 +181,7 @@ class ChromiumPage(ChromiumBase):
|
||||
if bid:
|
||||
kwargs['browserContextId'] = bid
|
||||
|
||||
tid = self.run_cdp('Target.createTarget', **kwargs)['targetId']
|
||||
tab = ChromiumTab(self, tab_id=tid)
|
||||
if url:
|
||||
tab.get(url)
|
||||
return tab
|
||||
return self.run_cdp('Target.createTarget', **kwargs)['targetId']
|
||||
|
||||
def close_tabs(self, tabs_or_ids=None, others=False):
|
||||
"""关闭传入的标签页，默认关闭当前页。可传入多个
|
||||
@ -214,9 +222,13 @@ class ChromiumPage(ChromiumBase):
|
||||
"""
|
||||
self.close_tabs(tabs_or_ids, True)
|
||||
|
||||
def quit(self):
|
||||
"""关闭浏览器"""
|
||||
self.browser.quit()
|
||||
def quit(self, timeout=5, force=True):
|
||||
"""关闭浏览器
|
||||
:param timeout: 等待浏览器关闭超时时间
|
||||
:param force: 关闭超时是否强制终止进程
|
||||
:return: None
|
||||
"""
|
||||
self.browser.quit(timeout, force)
|
||||
|
||||
|
||||
def get_rename(original, rename):
|
||||
|
||||
@ -63,7 +63,9 @@ class ChromiumPage(ChromiumBase):
|
||||
tab_type: Union[str, list, tuple] = None, single: bool = True) -> Union[str, List[str]]: ...
|
||||
|
||||
def new_tab(self, url: str = None, new_window: bool = False, background: bool = False,
|
||||
new_context: bool = False) -> Union[ChromiumTab, ChromiumPage]: ...
|
||||
new_context: bool = False) -> ChromiumTab: ...
|
||||
|
||||
def _new_tab(self, new_window: bool = False, background: bool = False, new_context: bool = False) -> str: ...
|
||||
|
||||
def close_tabs(self, tabs_or_ids: Union[str, ChromiumTab, List[Union[str, ChromiumTab]],
|
||||
Tuple[Union[str, ChromiumTab]]] = None, others: bool = False) -> None: ...
|
||||
@ -71,7 +73,7 @@ class ChromiumPage(ChromiumBase):
|
||||
def close_other_tabs(self, tabs_or_ids: Union[
|
||||
str, ChromiumTab, List[Union[str, ChromiumTab]], Tuple[Union[str, ChromiumTab]]] = None) -> None: ...
|
||||
|
||||
def quit(self) -> None: ...
|
||||
def quit(self, timeout: float = 5, force: bool = True) -> None: ...
|
||||
|
||||
|
||||
def get_rename(original: str, rename: str) -> str: ...
|
||||
|
||||
@ -296,13 +296,18 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
"""
|
||||
return tab_id if isinstance(tab_id, WebPageTab) else WebPageTab(self, tab_id or self.tab_id)
|
||||
|
||||
def new_tab(self, url=None, switch_to=False):
|
||||
"""新建一个标签页,该标签页在最后面
|
||||
def new_tab(self, url=None, new_window=False, background=False, new_context=False):
|
||||
"""新建一个标签页
|
||||
:param url: 新标签页跳转到的网址
|
||||
:param switch_to: 新建标签页后是否把焦点移过去
|
||||
:return: switch_to为False时返回新标签页对象，否则返回当前对象，
|
||||
:param new_window: 是否在新窗口打开标签页
|
||||
:param background: 是否不激活新标签页，如new_window为True则无效
|
||||
:param new_context: 是否创建新的上下文
|
||||
:return: 新标签页对象
|
||||
"""
|
||||
return self if switch_to else WebPageTab(self, self._new_tab(url, switch_to))
|
||||
tab = WebPageTab(self, tab_id=self._new_tab(new_window, background, new_context))
|
||||
if url:
|
||||
tab.get(url)
|
||||
return tab
|
||||
|
||||
def close_driver(self):
|
||||
"""关闭driver及浏览器"""
|
||||
@ -340,14 +345,18 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
return super(SessionPage, self)._find_elements(loc_or_ele, timeout=timeout, single=single,
|
||||
relative=relative)
|
||||
|
||||
def quit(self):
|
||||
"""关闭浏览器，关闭session"""
|
||||
def quit(self, timeout=5, force=True):
|
||||
"""关闭浏览器和Session
|
||||
:param timeout: 等待浏览器关闭超时时间
|
||||
:param force: 关闭超时是否强制终止进程
|
||||
:return: None
|
||||
"""
|
||||
if self._has_session:
|
||||
self._session.close()
|
||||
self._session = None
|
||||
self._response = None
|
||||
self._has_session = None
|
||||
if self._has_driver:
|
||||
super(SessionPage, self).quit()
|
||||
super(SessionPage, self).quit(timeout, force)
|
||||
self._driver = None
|
||||
self._has_driver = None
|
||||
|
||||
@ -121,7 +121,8 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
|
||||
def get_tab(self, tab_id: Union[str, WebPageTab] = None) -> WebPageTab: ...
|
||||
|
||||
def new_tab(self, url: str = None, switch_to: bool = False) -> Union[WebPageTab, WebPage]: ...
|
||||
def new_tab(self, url: str = None, new_window: bool = False, background: bool = False,
|
||||
new_context: bool = False) -> WebPageTab: ...
|
||||
|
||||
def close_driver(self) -> None: ...
|
||||
|
||||
@ -159,6 +160,6 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||
def _set_start_options(self, dr_opt: Union[ChromiumDriver, bool, None],
|
||||
se_opt: Union[Session, SessionOptions, bool, None]) -> None: ...
|
||||
|
||||
def quit(self) -> None: ...
|
||||
def quit(self, timeout: float = 5, force: bool = True) -> None: ...
|
||||
|
||||
def _on_download_begin(self, **kwargs): ...
|
||||
|
||||
@ -6,3 +6,4 @@ FlowViewer>=0.3.0
|
||||
websocket-client
|
||||
click
|
||||
tldextract
|
||||
psutil
|
||||
Loading…
x
Reference in New Issue
Block a user