mirror of
https://gitee.com/g1879/DrissionPage.git
synced 2024-12-10 04:00:23 +08:00
ChromiumBase删除session属性；内置控制连接每次请求后关闭
This commit is contained in:
g1879
parent
453b0fed85
commit
a5d67dc628
@ -19,7 +19,7 @@ from .chromium_element import ChromiumScroll, ChromiumElement, run_js, make_chro
|
||||
from .commons.constants import HANDLE_ALERT_METHOD, ERROR, NoneElement
|
||||
from .commons.locator import get_loc
|
||||
from .commons.tools import get_usable_path, clean_folder
|
||||
from .commons.web import set_browser_cookies, set_session_cookies
|
||||
from .commons.web import set_browser_cookies
|
||||
from .errors import ContextLossError, ElementLossError, AlertExistsError, CDPError, TabClosedError, \
|
||||
NoRectError, BrowserConnectError, GetDocumentError
|
||||
from .network_listener import NetworkListener
|
||||
@ -45,7 +45,6 @@ class ChromiumBase(BasePage):
|
||||
self._listener = None
|
||||
self._DownloadKit = None
|
||||
self._download_path = None
|
||||
self._session = None
|
||||
|
||||
if isinstance(address, int) or (isinstance(address, str) and address.isdigit()):
|
||||
address = f'127.0.0.1:{address}'
|
||||
@ -75,7 +74,9 @@ class ChromiumBase(BasePage):
|
||||
"""
|
||||
self._chromium_init()
|
||||
if not tab_id:
|
||||
json = self._control_session.get(f'http://{self.address}/json').json()
|
||||
u = f'http://{self.address}/json'
|
||||
json = self._control_session.get(u).json()
|
||||
self._control_session.get(u, headers={'Connection': 'close'})
|
||||
tab_id = [i['id'] for i in json if i['type'] == 'page']
|
||||
if not tab_id:
|
||||
raise BrowserConnectError('浏览器连接失败，可能是浏览器版本原因。')
|
||||
@ -88,6 +89,7 @@ class ChromiumBase(BasePage):
|
||||
"""浏览器初始设置"""
|
||||
self._control_session = Session()
|
||||
self._control_session.keep_alive = False
|
||||
self._control_session.proxies = {'http': None, 'https': None}
|
||||
self._first_run = True
|
||||
self._is_reading = False
|
||||
self._upload_list = None
|
||||
@ -392,17 +394,6 @@ class ChromiumBase(BasePage):
|
||||
self._DownloadKit = DownloadKit(session=self, goal_path=self.download_path)
|
||||
return self._DownloadKit
|
||||
|
||||
@property
|
||||
def session(self):
|
||||
"""返回用于DownloadKit的Session对象"""
|
||||
if self._session is None:
|
||||
self._session = Session()
|
||||
ua = self.run_cdp('Runtime.evaluate', expression='navigator.userAgent;')['result']['value']
|
||||
self._session.headers.update({"User-Agent": ua})
|
||||
|
||||
set_session_cookies(self._session, self.get_cookies(as_dict=False, all_info=False))
|
||||
return self._session
|
||||
|
||||
def run_cdp(self, cmd, **cmd_args):
|
||||
"""执行Chrome DevTools Protocol语句
|
||||
:param cmd: 协议项目
|
||||
|
||||
@ -18,7 +18,6 @@ from .chromium_frame import ChromiumFrame
|
||||
from .commons.constants import NoneElement
|
||||
from .network_listener import NetworkListener
|
||||
from .session_element import SessionElement
|
||||
from .session_page import DownloadSetter
|
||||
|
||||
|
||||
class ChromiumBase(BasePage):
|
||||
@ -47,7 +46,6 @@ class ChromiumBase(BasePage):
|
||||
self._listener: NetworkListener = ...
|
||||
self._download_path: str = ...
|
||||
self._DownloadKit: DownloadKit = ...
|
||||
self._session: Session = ...
|
||||
|
||||
def _connect_browser(self, tab_id: str = None) -> None: ...
|
||||
|
||||
@ -144,9 +142,6 @@ class ChromiumBase(BasePage):
|
||||
@property
|
||||
def download(self) -> DownloadKit: ...
|
||||
|
||||
@property
|
||||
def session(self) -> Session: ...
|
||||
|
||||
def run_js(self, script: str, *args: Any, as_expr: bool = False) -> Any: ...
|
||||
|
||||
def run_js_loaded(self, script: str, *args: Any, as_expr: bool = False) -> Any: ...
|
||||
@ -225,22 +220,6 @@ class ChromiumBase(BasePage):
|
||||
timeout: float = None) -> Union[bool, None]: ...
|
||||
|
||||
|
||||
class BaseDownloadSetter(DownloadSetter):
|
||||
def __init__(self, page: ChromiumBase):
|
||||
self._page: ChromiumBase = ...
|
||||
self._session: Session = ...
|
||||
self._show_msg: bool = ...
|
||||
|
||||
@property
|
||||
def session(self) -> Session: ...
|
||||
|
||||
def save_path(self, path: Union[str, Path]) -> None: ...
|
||||
|
||||
def show_msg(self, on_off: bool = True) -> None: ...
|
||||
|
||||
def _cookies_to_session(self) -> None: ...
|
||||
|
||||
|
||||
class ChromiumBaseWaiter(object):
|
||||
def __init__(self, page: ChromiumBase):
|
||||
self._driver: ChromiumBase = ...
|
||||
|
||||
@ -69,7 +69,9 @@ class ChromiumFrame(ChromiumBase):
|
||||
try:
|
||||
super()._driver_init(tab_id)
|
||||
except:
|
||||
self._control_session.get(f'http://{self.address}/json')
|
||||
u = f'http://{self.address}/json'
|
||||
self._control_session.get(u)
|
||||
self._control_session.get(u, headers={'Connection': 'close'})
|
||||
super()._driver_init(tab_id)
|
||||
|
||||
def _reload(self):
|
||||
|
||||
@ -73,7 +73,9 @@ class ChromiumPage(ChromiumBase):
|
||||
if not self._tab_obj: # 不是传入driver的情况
|
||||
connect_browser(self._driver_options)
|
||||
if not tab_id:
|
||||
json = self._control_session.get(f'http://{self.address}/json').json()
|
||||
u = f'http://{self.address}/json'
|
||||
json = self._control_session.get(u).json()
|
||||
self._control_session.get(u, headers={'Connection': 'close'})
|
||||
tab_id = [i['id'] for i in json if i['type'] == 'page']
|
||||
if not tab_id:
|
||||
raise BrowserConnectError('浏览器连接失败，可能是浏览器版本原因。')
|
||||
@ -87,7 +89,9 @@ class ChromiumPage(ChromiumBase):
|
||||
|
||||
def _page_init(self):
|
||||
"""页面相关设置"""
|
||||
ws = self._control_session.get(f'http://{self.address}/json/version').json()['webSocketDebuggerUrl']
|
||||
u = f'http://{self.address}/json/version'
|
||||
ws = self._control_session.get(u).json()['webSocketDebuggerUrl']
|
||||
self._control_session.get(u, headers={'Connection': 'close'})
|
||||
self._browser_driver = ChromiumDriver(ws.split('/')[-1], 'browser', self.address)
|
||||
self._browser_driver.start()
|
||||
|
||||
@ -124,7 +128,9 @@ class ChromiumPage(ChromiumBase):
|
||||
@property
|
||||
def tabs(self):
|
||||
"""返回所有标签页id组成的列表"""
|
||||
j = self._control_session.get(f'http://{self.address}/json').json() # 不要改用cdp
|
||||
u = f'http://{self.address}/json'
|
||||
j = self._control_session.get(u).json() # 不要改用cdp
|
||||
self._control_session.get(u, headers={'Connection': 'close'})
|
||||
return [i['id'] for i in j if i['type'] == 'page']
|
||||
|
||||
@property
|
||||
@ -195,7 +201,9 @@ class ChromiumPage(ChromiumBase):
|
||||
:param single: 是否返回首个结果的id，为False返回所有信息
|
||||
:return: tab id或tab dict
|
||||
"""
|
||||
tabs = self._control_session.get(f'http://{self.address}/json').json() # 不要改用cdp
|
||||
u = f'http://{self.address}/json'
|
||||
tabs = self._control_session.get(u).json() # 不要改用cdp
|
||||
self._control_session.get(u, headers={'Connection': 'close'})
|
||||
if isinstance(tab_type, str):
|
||||
tab_type = {tab_type}
|
||||
elif isinstance(tab_type, (list, tuple, set)):
|
||||
|
||||
@ -145,7 +145,9 @@ def test_connect(ip, port):
|
||||
end_time = perf_counter() + 30
|
||||
while perf_counter() < end_time:
|
||||
try:
|
||||
tabs = requests_get(f'http://{ip}:{port}/json', timeout=10).json()
|
||||
u = f'http://{ip}:{port}/json'
|
||||
tabs = requests_get(u, timeout=10, proxies={'http': None, 'https': None}).json()
|
||||
requests_get(u, headers={'Connection': 'close'}, proxies={'http': None, 'https': None})
|
||||
for tab in tabs:
|
||||
if tab['type'] == 'page':
|
||||
return
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user