blackteay/DrissionPage
1
0
Fork 0
mirror of https://gitee.com/g1879/DrissionPage.git synced 2024-12-10 04:00:23 +08:00
Code Issues Projects Releases Wiki Activity

修复disconnect()的问题；MixTab默认不创建Session对象；MixTab的close()增加session参数；MixTab和MixPage的post()必返回Response对象

Browse Source
This commit is contained in:
g1879 2024-08-29 22:38:47 +08:00
parent bbf1bc7e28
commit 70bff8555c
15 changed files with 148 additions and 135 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
DrissionPage/_base/chromium.py
View File

@ -74,6 +74,7 @@ class Chromium(object):
self.retry_times = self._chromium_options.retry_times
self.retry_interval = self._chromium_options.retry_interval
self.address = self._chromium_options.address
self._disconnect_flag = False
self._driver = BrowserDriver(self.id, 'browser', self.address, self)
if ((not self._chromium_options._ua_set and self._is_headless != self._chromium_options.is_headless)
@ -223,12 +224,14 @@ class Chromium(object):
self._run_cdp('Target.activateTarget', targetId=id_ind_tab)
def reconnect(self):
self._disconnect_flag = True
self._driver.stop()
BrowserDriver.BROWSERS.pop(self.id)
self._driver = BrowserDriver(self.id, 'browser', self.address, self)
self._run_cdp('Target.setDiscoverTargets', discover=True)
self._driver.set_callback('Target.targetDestroyed', self._onTargetDestroyed)
self._driver.set_callback('Target.targetCreated', self._onTargetCreated)
self._disconnect_flag = False
def clear_cache(self, cache=True, cookies=True):
if cache:
@ -396,19 +399,20 @@ class Chromium(object):
self._all_drivers.pop(tab_id, None)
def _on_disconnect(self):
Chromium._BROWSERS.pop(self.id, None)
if self._chromium_options.is_auto_port and self._chromium_options.user_data_path:
path = Path(self._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, OSError):
pass
sleep(.03)
if not self._disconnect_flag:
Chromium._BROWSERS.pop(self.id, None)
if self._chromium_options.is_auto_port and self._chromium_options.user_data_path:
path = Path(self._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, OSError):
pass
sleep(.03)
def handle_options(addr_or_opts):
@ -468,11 +472,12 @@ def run_browser(chromium_options):
return is_headless, browser_id, is_exists
def _new_tab_by_js(browser:Chromium, url, obj, new_window):
tab = browser.get_mix_tab() if isinstance(obj, MixTab) else browser.get_tab()
def _new_tab_by_js(browser: Chromium, url, obj, new_window):
mix = isinstance(obj, MixTab)
tab = browser._get_tab(mix=mix)
url = f'"{url}"' if url else ''
new = 'target="_new"' if new_window else 'target="_blank"'
tid = browser.latest_tab.tab_id
tab.run_js(f'window.open({url}, {new})')
tid = browser.wait.new_tab(curr_tab=tid)
return browser.get_mix_tab(tid) if isinstance(obj, MixTab) else browser.get_tab(tid)
return browser._get_tab(tid, mix=mix)

1
DrissionPage/_base/chromium.pyi
View File

@ -48,6 +48,7 @@ class Chromium(object):
_auto_handle_alert: Optional[bool] = ...
_is_exists: bool = ...
_is_headless: bool = ...
_disconnect_flag: bool = ...
def __new__(cls,
addr_or_opts: Union[str, int, ChromiumOptions] = None,

2
DrissionPage/_configs/session_options.pyi
View File

@ -199,7 +199,7 @@ class SessionOptions(object):
...
def set_cert(self, cert: Union[str, Tuple[str, str], None]) -> SessionOptions:
"""SSL客户端证书文件的路径(.pem格式)，或(‘cert’, ‘key’)元组
"""SSL客户端证书文件的路径(.pem格式)，或('cert', 'key')元组
:param cert: 证书路径或元组
:return: 返回当前对象
"""

7
DrissionPage/_pages/chromium_base.py
View File

@ -61,6 +61,7 @@ class ChromiumBase(BasePage):
self._auto_handle_alert = None
self._load_end_time = 0
self._init_jss = []
self._disconnect_flag = False
self._type = 'ChromiumBase'
if not hasattr(self, '_listener'):
self._listener = None
@ -330,11 +331,11 @@ class ChromiumBase(BasePage):
@property
def tab_id(self):
return self._target_id
return self.driver.id
@property
def _target_id(self):
return self.driver.id if self.driver.is_running else ''
return self.driver.id
@property
def active_ele(self):
@ -650,8 +651,10 @@ class ChromiumBase(BasePage):
def disconnect(self):
if self._driver:
self._disconnect_flag = True
self._driver.stop()
self.browser._all_drivers.get(self._driver.id, set()).discard(self._driver)
self._disconnect_flag = False
def reconnect(self, wait=0):
t_id = self._target_id

1
DrissionPage/_pages/chromium_base.pyi
View File

@ -63,6 +63,7 @@ class ChromiumBase(BasePage):
_ready_state: Optional[str] = ...
_rect: Optional[TabRect] = ...
_console: Optional[Console] = ...
_disconnect_flag: bool = ...
_type: str = ...
def __init__(self,

3
DrissionPage/_pages/chromium_tab.py
View File

@ -67,4 +67,5 @@ class ChromiumTab(ChromiumBase):
return f'<ChromiumTab browser_id={self.browser.id} tab_id={self.tab_id}>'
def _on_disconnect(self):
ChromiumTab._TABS.pop(self.tab_id, None)
if not self._disconnect_flag:
ChromiumTab._TABS.pop(self.tab_id, None)

2
DrissionPage/_pages/chromium_tab.pyi
View File

@ -106,3 +106,5 @@ class ChromiumTab(ChromiumBase):
:return: as_pdf为True时返回bytes，否则返回文件文本
"""
...
def _on_disconnect(self): ...

9
DrissionPage/_pages/mix_page.py
View File

@ -153,8 +153,9 @@ class MixPage(SessionPage, ChromiumPage, BasePage):
# s模式转d模式
if self._d_mode:
if self._driver is None:
self._connect_browser(self._chromium_options)
if self._driver is None or not self._driver.is_running:
self._driver_init(self.tab_id)
self._get_document()
self._url = None if not self._has_driver else super(SessionPage, self).url
self._has_driver = True
@ -170,7 +171,7 @@ class MixPage(SessionPage, ChromiumPage, BasePage):
self._has_session = True
self._url = self._session_url
if self._has_driver:
if self._has_driver and self._driver.is_running:
if copy_cookies:
self.cookies_to_session()
if go and not self.get(super(SessionPage, self).url):
@ -203,7 +204,7 @@ class MixPage(SessionPage, ChromiumPage, BasePage):
return self.browser._get_tabs(title=title, url=url, tab_type=tab_type, mix=True, as_id=as_id)
def new_tab(self, url=None, new_window=False, background=False, new_context=False):
return self.browser.new_mix_tab(url=url, new_window=new_window, background=background, new_context=new_context)
return self.browser._new_tab(url=url, new_window=new_window, background=background, new_context=new_context)
def close_driver(self):
if self._has_driver:

71
DrissionPage/_pages/mix_page.pyi
View File

@ -5,6 +5,7 @@
@Copyright: (c) 2024 by g1879, Inc. All Rights Reserved.
@License : BSD 3-Clause.
"""
from http.cookiejar import CookieJar
from typing import Union, Tuple, List, Any, Optional, Literal
from requests import Session, Response
@ -135,22 +136,22 @@ class MixPage(SessionPage, ChromiumPage, BasePage):
def get(self,
url: str,
show_errmsg: bool = False,
retry: int | None = None,
interval: float | None = None,
timeout: float | None = None,
params: dict | None = ...,
data: Union[dict, str, None] = ...,
json: Union[dict, str, None] = ...,
headers: Union[dict, str, None] = ...,
cookies: Any | None = ...,
files: Any | None = ...,
auth: Any | None = ...,
allow_redirects: bool = ...,
proxies: dict | None = ...,
hooks: Any | None = ...,
stream: Any | None = ...,
verify: Any | None = ...,
cert: Any | None = ...) -> Union[bool, None]:
retry: Optional[int] = None,
interval: Optional[float] = None,
timeout: Optional[float] = None,
params: Optional[dict] = None,
data: Union[dict, str, None] = None,
json: Union[dict, str, None] = None,
headers: Optional[dict] = None,
cookies: Union[CookieJar, dict] = None,
files: Optional[Any] = None,
auth: Optional[Any] = None,
allow_redirects: bool = True,
proxies: Optional[dict] = None,
hooks: Optional[Any] = None,
stream: bool = None,
verify: Union[bool, str] = None,
cert: [str, Tuple[str, str]] = None) -> Union[bool, None]:
"""跳转到一个url
:param url: 目标url
:param show_errmsg: 是否显示和抛出异常
@ -169,30 +170,30 @@ class MixPage(SessionPage, ChromiumPage, BasePage):
:param hooks: 回调方法
:param stream: 是否使用流式传输
:param verify: 是否验证 SSL 证书
:param cert: SSL客户端证书文件的路径(.pem格式)，或(‘cert’, ‘key’)元组
:param cert: SSL客户端证书文件的路径(.pem格式)，或('cert', 'key')元组
:return: s模式时返回url是否可用，d模式时返回获取到的Response对象
"""
...
def post(self,
url: str,
data: Union[dict, str, None] = None,
show_errmsg: bool = False,
retry: int | None = None,
interval: float | None = None,
timeout: float | None = ...,
params: dict | None = ...,
json: Union[dict, str, None] = ...,
headers: Union[dict, str, None] = ...,
cookies: Any | None = ...,
files: Any | None = ...,
auth: Any | None = ...,
allow_redirects: bool = ...,
proxies: dict | None = ...,
hooks: Any | None = ...,
stream: Any | None = ...,
verify: Any | None = ...,
cert: Any | None = ...) -> Union[bool, Response]:
retry: Optional[int] = None,
interval: Optional[float] = None,
timeout: Optional[float] = None,
params: Optional[dict] = None,
data: Union[dict, str, None] = None,
json: Union[dict, str, None] = None,
headers: Optional[dict] = None,
cookies: Union[CookieJar, dict] = None,
files: Optional[Any] = None,
auth: Optional[Any] = None,
allow_redirects: bool = True,
proxies: Optional[dict] = None,
hooks: Optional[Any] = None,
stream: bool = None,
verify: Union[bool, str] = None,
cert: [str, Tuple[str, str]] = None) -> Response:
"""用post方式跳转到url
:param url: 目标url
:param show_errmsg: 是否显示和抛出异常
@ -211,8 +212,8 @@ class MixPage(SessionPage, ChromiumPage, BasePage):
:param hooks: 回调方法
:param stream: 是否使用流式传输
:param verify: 是否验证 SSL 证书
:param cert: SSL客户端证书文件的路径(.pem格式)，或(‘cert’, ‘key’)元组
:return: s模式时返回url是否可用，d模式时返回获取到的Response对象
:param cert: SSL客户端证书文件的路径(.pem格式)，或('cert', 'key')元组
:return: 获取到的Response对象
"""
...

61
DrissionPage/_pages/mix_tab.py
View File

@ -19,12 +19,13 @@ class MixTab(SessionPage, ChromiumTab, BasePage):
def __init__(self, browser, tab_id):
if Settings.singleton_tab_obj and hasattr(self, '_created'):
return
self._d_mode = True
self._has_driver = True
self._has_session = True
super().__init__(session_or_options=browser._session_options or SessionOptions(
read_file=browser._session_options is None))
self._session_options = None
self._headers = None
self._response = None
self._session = None
self._encoding = None
self._timeout = 10
super(SessionPage, self).__init__(browser=browser, tab_id=tab_id)
self._type = 'MixTab'
@ -52,15 +53,11 @@ class MixTab(SessionPage, ChromiumTab, BasePage):
@property
def raw_data(self):
if self._d_mode:
return super(SessionPage, self).html if self._has_driver else ''
return super().raw_data
return super(SessionPage, self).html if self._d_mode else super().raw_data
@property
def html(self):
if self._d_mode:
return super(SessionPage, self).html if self._has_driver else ''
return super().html
return super(SessionPage, self).html if self._d_mode else super().html
@property
def json(self):
@ -99,15 +96,16 @@ class MixTab(SessionPage, ChromiumTab, BasePage):
return super(SessionPage, self).get(url, show_errmsg, retry, interval, timeout)
if timeout is None:
timeout = self.timeouts.page_load if self._has_driver else self.timeout
timeout = self.timeouts.page_load
return super().get(url, show_errmsg, retry, interval, timeout, **kwargs)
def post(self, url, show_errmsg=False, retry=None, interval=None, **kwargs):
def post(self, url, show_errmsg=False, retry=None, interval=None, timeout=None, **kwargs):
if self.mode == 'd':
self.cookies_to_session()
super().post(url, show_errmsg, retry, interval, **kwargs)
return self.response
return super().post(url, show_errmsg, retry, interval, **kwargs)
if timeout is None:
kwargs['timeout'] = self.timeouts.page_load
super().post(url, show_errmsg, retry, interval, **kwargs)
return self.response
def ele(self, locator, index=1, timeout=None):
return super(SessionPage, self).ele(locator, index=index, timeout=timeout) if self._d_mode \
@ -133,13 +131,11 @@ class MixTab(SessionPage, ChromiumTab, BasePage):
# s模式转d模式
if self._d_mode:
if self._driver is None: # todo: 优化这里的逻辑
tabs = self.browser.tab_ids
tid = self.tab_id if self.tab_id in tabs else tabs[0]
self._connect_browser(tid)
if self._driver is None or not self._driver.is_running:
self._driver_init(self.tab_id)
self._get_document()
self._url = None if not self._has_driver else super(SessionPage, self).url
self._has_driver = True
self._url = super(SessionPage, self).url
if self._session_url:
if copy_cookies:
self.cookies_to_browser()
@ -149,9 +145,13 @@ class MixTab(SessionPage, ChromiumTab, BasePage):
return
# d模式转s模式
self._has_session = True
if self._session is None:
self._s_set_start_options(
self.browser._session_options or SessionOptions(read_file=self.browser._session_options is None))
self._create_session()
self._url = self._session_url
if self._has_driver:
if self._driver:
if copy_cookies:
self.cookies_to_session()
@ -161,7 +161,7 @@ class MixTab(SessionPage, ChromiumTab, BasePage):
self.get(url)
def cookies_to_session(self, copy_user_agent=True):
if not self._has_session:
if not self._session:
return
if copy_user_agent:
@ -171,7 +171,7 @@ class MixTab(SessionPage, ChromiumTab, BasePage):
set_session_cookies(self.session, super(SessionPage, self).cookies())
def cookies_to_browser(self):
if not self._has_driver:
if self._driver is None or not self._driver.is_running:
return
set_tab_cookies(self, super().cookies())
@ -179,11 +179,12 @@ class MixTab(SessionPage, ChromiumTab, BasePage):
return super(SessionPage, self).cookies(all_domains, all_info) if self._d_mode \
else super().cookies(all_domains, all_info)
def close(self, others=False):
def close(self, others=False, session=False):
self.browser.close_tabs(self.tab_id, others=others)
self._session.close()
if self._response is not None:
self._response.close()
if session and self._session:
self._session.close()
if self._response is not None:
self._response.close()
def _find_elements(self, locator, timeout=None, index=1, relative=False, raise_err=None):
return super(SessionPage, self)._find_elements(locator, timeout=timeout, index=index, relative=relative) \

71
DrissionPage/_pages/mix_tab.pyi
View File

@ -5,6 +5,7 @@
@Copyright: (c) 2024 by g1879, Inc. All Rights Reserved.
@License : BSD 3-Clause.
"""
from http.cookiejar import CookieJar
from typing import Union, Tuple, Any, Optional, Literal
from requests import Session, Response
@ -24,8 +25,6 @@ from .._units.waiter import MixTabWaiter
class MixTab(SessionPage, ChromiumTab):
_tab: MixTab = ...
_d_mode: bool = ...
_has_driver: bool = ...
_has_session: bool = ...
_set: MixTabSetter = ...
def __init__(self, browser: Chromium, tab_id: str):
@ -121,22 +120,22 @@ class MixTab(SessionPage, ChromiumTab):
def get(self,
url: str,
show_errmsg: bool = False,
retry: int | None = None,
interval: float | None = None,
timeout: float | None = None,
params: dict | None = ...,
data: Union[dict, str, None] = ...,
json: Union[dict, str, None] = ...,
headers: dict | None = ...,
cookies: Any | None = ...,
files: Any | None = ...,
auth: Any | None = ...,
allow_redirects: bool = ...,
proxies: dict | None = ...,
hooks: Any | None = ...,
stream: Any | None = ...,
verify: Any | None = ...,
cert: Any | None = ...) -> Union[bool, None]:
retry: Optional[int] = None,
interval: Optional[float] = None,
timeout: Optional[float] = None,
params: Optional[dict] = None,
data: Union[dict, str, None] = None,
json: Union[dict, str, None] = None,
headers: Optional[dict] = None,
cookies: Union[CookieJar, dict] = None,
files: Optional[Any] = None,
auth: Optional[Any] = None,
allow_redirects: bool = True,
proxies: Optional[dict] = None,
hooks: Optional[Any] = None,
stream: bool = None,
verify: Union[bool, str] = None,
cert: [str, Tuple[str, str]] = None) -> Union[bool, None]:
"""跳转到一个url
:param url: 目标url
:param show_errmsg: 是否显示和抛出异常
@ -155,7 +154,7 @@ class MixTab(SessionPage, ChromiumTab):
:param hooks: 回调方法
:param stream: 是否使用流式传输
:param verify: 是否验证 SSL 证书
:param cert: SSL客户端证书文件的路径(.pem格式)，或(‘cert’, ‘key’)元组
:param cert: SSL客户端证书文件的路径(.pem格式)，或('cert', 'key')元组
:return: s模式时返回url是否可用，d模式时返回获取到的Response对象
"""
...
@ -163,22 +162,22 @@ class MixTab(SessionPage, ChromiumTab):
def post(self,
url: str,
show_errmsg: bool = False,
retry: int | None = None,
interval: float | None = None,
timeout: float | None = ...,
params: dict | None = ...,
retry: Optional[int] = None,
interval: Optional[float] = None,
timeout: Optional[float] = None,
params: Optional[dict] = None,
data: Union[dict, str, None] = None,
json: Union[dict, str, None] = ...,
headers: dict | None = ...,
cookies: Any | None = ...,
files: Any | None = ...,
auth: Any | None = ...,
allow_redirects: bool = ...,
proxies: dict | None = ...,
hooks: Any | None = ...,
stream: Any | None = ...,
verify: Any | None = ...,
cert: Any | None = ...) -> Union[bool, Response]:
json: Union[dict, str, None] = None,
headers: Optional[dict] = None,
cookies: Union[CookieJar, dict] = None,
files: Optional[Any] = None,
auth: Optional[Any] = None,
allow_redirects: bool = True,
proxies: Optional[dict] = None,
hooks: Optional[Any] = None,
stream: bool = None,
verify: Union[bool, str] = None,
cert: [str, Tuple[str, str]] = None) -> Response:
"""用post方式跳转到url
:param url: 目标url
:param show_errmsg: 是否显示和抛出异常
@ -197,8 +196,8 @@ class MixTab(SessionPage, ChromiumTab):
:param hooks: 回调方法
:param stream: 是否使用流式传输
:param verify: 是否验证 SSL 证书
:param cert: SSL客户端证书文件的路径(.pem格式)，或(‘cert’, ‘key’)元组
:return: s模式时返回url是否可用，d模式时返回获取到的Response对象
:param cert: SSL客户端证书文件的路径(.pem格式)，或('cert', 'key')元组
:return: 获取到的Response对象
"""
...

8
DrissionPage/_pages/session_page.pyi
View File

@ -22,7 +22,7 @@ from .._units.setter import SessionPageSetter
class SessionPage(BasePage):
_headers: Optional[CaseInsensitiveDict] = ...
_session: Optional[Session] = ...
_session_options: SessionOptions = ...
_session_options: Optional[SessionOptions] = ...
_url: str = ...
_response: Optional[Response] = ...
_url_available: bool = ...
@ -115,7 +115,7 @@ class SessionPage(BasePage):
@property
def encoding(self) -> str:
"""返回设置的编码"""
"""返回设置的编码，s模式专用"""
...
@property
@ -165,7 +165,7 @@ class SessionPage(BasePage):
:param hooks: 回调方法
:param stream: 是否使用流式传输
:param verify: 是否验证 SSL 证书
:param cert: SSL客户端证书文件的路径(.pem格式)，或(‘cert’, ‘key’)元组
:param cert: SSL客户端证书文件的路径(.pem格式)，或('cert', 'key')元组
:return: s模式时返回url是否可用，d模式时返回获取到的Response对象
"""
...
@ -207,7 +207,7 @@ class SessionPage(BasePage):
:param hooks: 回调方法
:param stream: 是否使用流式传输
:param verify: 是否验证 SSL 证书
:param cert: SSL客户端证书文件的路径(.pem格式)，或(‘cert’, ‘key’)元组
:param cert: SSL客户端证书文件的路径(.pem格式)，或('cert', 'key')元组
:return: s模式时返回url是否可用，d模式时返回获取到的Response对象
"""
...

6
DrissionPage/_units/clicker.py
View File

@ -103,8 +103,7 @@ class Clicker(object):
tid = self._ele.tab.browser.wait.new_tab(curr_tab=curr_tid)
if not tid:
raise RuntimeError('没有出现新标签页。')
return (self._ele.tab.browser.get_mix_tab(tid) if self._ele.tab._type == 'MixTab'
else self._ele.tab.browser.get_tab(tid))
return self._ele.tab.browser._get_tab(tid, mix=self._ele.tab._type == 'MixTab')
def at(self, offset_x=None, offset_y=None, button='left', count=1):
self._ele.owner.scroll.to_see(self._ele)
@ -141,8 +140,7 @@ class Clicker(object):
tid = self._ele.tab.browser.wait.new_tab(timeout=timeout, curr_tab=curr_tid)
if not tid:
raise RuntimeError('没有出现新标签页。')
return (self._ele.tab.browser.get_mix_tab(tid) if self._ele.tab._type == 'MixTab'
else self._ele.tab.browser.get_tab(tid))
return self._ele.tab.browser._get_tab(tid, mix=self._ele.tab._type == 'MixTab')
def for_url_change(self, text=None, exclude=False, by_js=False, timeout=None):
if text is None:

2
DrissionPage/_units/setter.pyi
View File

@ -164,7 +164,7 @@ class SessionPageSetter(BaseSetter):
...
def cert(self, cert: Union[str, Tuple[str, str], None]) -> None:
"""SSL客户端证书文件的路径(.pem格式)，或(‘cert’, ‘key’)元组
"""SSL客户端证书文件的路径(.pem格式)，或('cert', 'key')元组
:param cert: 证书路径或元组
:return: None
"""

2
DrissionPage/common.py
View File

@ -49,5 +49,5 @@ def from_playwright(page_or_browser):
port = con_info.laddr.port
break
else:
raise RuntimeError('获取失败。')
raise RuntimeError('获取失败，请用管理员权限运行。')
return Chromium(f'127.0.0.1:{port}')