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

连接函数增加show_errmsg参数

Browse Source
This commit is contained in:
g1879 2020-08-14 00:27:31 +08:00
parent 2ab3b7b5cb
commit 016d54561c
3 changed files with 75 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
DrissionPage/driver_page.py
View File

@ -60,10 +60,11 @@ class DriverPage(object):
"""返回网页title"""
return self.driver.title
def get(self, url: str, go_anyway: bool = False) -> Union[None, bool]:
def get(self, url: str, go_anyway: bool = False, show_errmsg: bool = False) -> Union[None, bool]:
"""访问url \n
:param url: 目标url
:param go_anyway: 若目标url与当前url一致，是否强制跳转
:param show_errmsg: 是否显示和抛出异常
:return: 目标url是否可用
"""
to_url = quote(url, safe='/:&?=%;#@')
@ -72,6 +73,8 @@ class DriverPage(object):
self._url = to_url
self.driver.get(to_url)
self._url_available = self.check_page()
if self._url_available is False and show_errmsg:
raise ConnectionError('Connect error.')
return self._url_available
def ele(self,

32
DrissionPage/mix_page.py
View File

@ -154,19 +154,30 @@ class MixPage(Null, SessionPage, DriverPage):
u = url or self.session_url
self._drission.cookies_to_driver(u)
def check_page(self) -> Union[bool, None]:
return
# if self.session_url == self.url:
# return True if self._response and self._response.ok else False
# ----------------重写SessionPage的函数-----------------------
def post(self, url: str, data: dict = None, go_anyway: bool = False, **kwargs) -> Union[bool, None]:
def post(self,
url: str,
data: dict = None,
go_anyway: bool = False,
show_errmsg: bool = False,
**kwargs) -> Union[bool, None]:
"""用post方式跳转到url \n
post前先转换模式，但不跳转
:param url: 目标url
:param data: 提交的数据
:param go_anyway: 若目标url与当前url一致，是否强制跳转
:param show_errmsg: 是否显示和抛出异常
:param kwargs: 连接参数
:return: url是否可用
"""
self.change_mode('s', go=False)
return super().post(url, data, go_anyway, **kwargs)
return super().post(url, data, go_anyway, show_errmsg, **kwargs)
def download(self,
file_url: str,
@ -174,6 +185,7 @@ class MixPage(Null, SessionPage, DriverPage):
rename: str = None,
file_exists: str = 'rename',
show_msg: bool = False,
show_errmsg: bool = False,
**kwargs) -> tuple:
"""下载一个文件 \n
d模式下下载前先同步cookies \n
@ -182,12 +194,13 @@ class MixPage(Null, SessionPage, DriverPage):
:param rename: 重命名文件，可不写扩展名
:param file_exists: 若存在同名文件，可选择 'rename', 'overwrite', 'skip' 方式处理
:param show_msg: 是否显示下载信息
:param show_errmsg: 是否显示和抛出异常
:param kwargs: 连接参数
:return: 下载是否成功（bool）和状态信息（成功时信息为文件路径）的元组
"""
if self.mode == 'd':
self.cookies_to_session()
return super().download(file_url, goal_path, rename, file_exists, show_msg, **kwargs)
return super().download(file_url, goal_path, rename, file_exists, show_msg, show_errmsg, **kwargs)
# ----------------重写DriverPage的函数-----------------------
@ -207,24 +220,19 @@ class MixPage(Null, SessionPage, DriverPage):
# ----------------以下为共用函数-----------------------
def get(self, url: str, go_anyway=False, **kwargs) -> Union[bool, None]:
def get(self, url: str, go_anyway=False, show_errmsg: bool = False, **kwargs) -> Union[bool, None]:
"""跳转到一个url \n
跳转前先同步cookies，跳转后判断目标url是否可用
:param url: 目标url
:param go_anyway: 若目标url与当前url一致，是否强制跳转
:param show_errmsg: 是否显示和抛出异常
:param kwargs: 连接参数，s模式专用
:return: url是否可用
"""
if self._mode == 'd':
if super(SessionPage, self).get(url=url, go_anyway=go_anyway) is None:
return
if self.session_url == self.url:
self._url_available = True if self._response and self._response.ok else False
else:
self._url_available = self.check_page()
return self._url_available
return super(SessionPage, self).get(url, go_anyway, show_errmsg)
elif self._mode == 's':
return None if super().get(url=url, go_anyway=go_anyway, **kwargs) is None else self._url_available
return super().get(url, go_anyway, show_errmsg, **kwargs)
def ele(self,
loc_or_ele: Union[tuple, str, DriverElement, SessionElement, Element, WebElement],

62
DrissionPage/session_page.py
View File

@ -138,10 +138,15 @@ class SessionPage(object):
raise TypeError('Type of loc_or_str can only be tuple or str.')
return self.ele(loc_or_str, mode='all', show_errmsg=True)
def get(self, url: str, go_anyway: bool = False, **kwargs) -> Union[bool, None]:
def get(self,
url: str,
go_anyway: bool = False,
show_errmsg: bool = False,
**kwargs) -> Union[bool, None]:
"""用get方式跳转到url \n
:param url: 目标url
:param go_anyway: 若目标url与当前url一致，是否强制跳转
:param show_errmsg: 是否显示和抛出异常
:param kwargs: 连接参数
:return: url是否可用
"""
@ -149,17 +154,34 @@ class SessionPage(object):
if not url or (not go_anyway and self.url == to_url):
return
self._url = to_url
self._response = self._make_response(to_url, **kwargs)[0]
if self._response:
self._response.html.encoding = self._response.encoding # 修复requests_html丢失编码方式的bug
self._url_available = True if self._response and self._response.ok else False
self._response = self._make_response(to_url, show_errmsg=show_errmsg, **kwargs)[0]
if not self._response:
self._url_available = False
else:
try:
self._response.html.encoding = self._response.encoding # 修复requests_html丢失编码方式的bug
except:
pass
if self._response.ok:
self._url_available = True
else:
if show_errmsg:
raise ConnectionError(f'Status code: {self._response.status_code}.')
self._url_available = False
return self._url_available
def post(self, url: str, data: dict = None, go_anyway: bool = True, **kwargs) -> Union[bool, None]:
def post(self,
url: str,
data: dict = None,
go_anyway: bool = True,
show_errmsg: bool = False,
**kwargs) -> Union[bool, None]:
"""用post方式跳转到url \n
:param url: 目标url
:param data: 提交的数据
:param go_anyway: 若目标url与当前url一致，是否强制跳转
:param show_errmsg: 是否显示和抛出异常
:param kwargs: 连接参数
:return: url是否可用
"""
@ -167,13 +189,21 @@ class SessionPage(object):
if not url or (not go_anyway and self._url == to_url):
return
self._url = to_url
self._response = self._make_response(to_url, mode='post', data=data, **kwargs)[0]
if self._response:
self._response = self._make_response(to_url, mode='post', data=data, show_errmsg=show_errmsg, **kwargs)[0]
if not self._response:
self._url_available = False
else:
try:
self._response.html.encoding = self._response.encoding # 修复requests_html丢失编码方式的bug
except:
pass
self._url_available = True if self._response and self._response.status_code == 200 else False
if self._response.ok:
self._url_available = True
else:
if show_errmsg:
raise ConnectionError(f'Status code: {self._response.status_code}.')
self._url_available = False
return self._url_available
def download(self,
@ -182,6 +212,7 @@ class SessionPage(object):
rename: str = None,
file_exists: str = 'rename',
show_msg: bool = False,
show_errmsg: bool = False,
**kwargs) -> tuple:
"""下载一个文件 \n
:param file_url: 文件url
@ -189,6 +220,7 @@ class SessionPage(object):
:param rename: 重命名文件，可不写扩展名
:param file_exists: 若存在同名文件，可选择 'rename', 'overwrite', 'skip' 方式处理
:param show_msg: 是否显示下载信息
:param show_errmsg: 是否抛出和显示异常
:param kwargs: 连接参数
:return: 下载是否成功（bool）和状态信息（成功时信息为文件路径）的元组
"""
@ -201,7 +233,7 @@ class SessionPage(object):
if 'timeout' not in kwargs:
kwargs['timeout'] = 20
r, info = self._make_response(file_url, mode='get', **kwargs)
r, info = self._make_response(file_url, mode='get', show_errmsg=show_errmsg, **kwargs)
if not r:
if show_msg:
print(info)
@ -280,11 +312,17 @@ class SessionPage(object):
info = f'{goal_path}\\{full_name}' if download_status else info
return download_status, info
def _make_response(self, url: str, mode: str = 'get', data: dict = None, **kwargs) -> tuple:
def _make_response(self,
url: str,
mode: str = 'get',
data: dict = None,
show_errmsg: bool = False,
**kwargs) -> tuple:
"""生成response对象 \n
:param url: 目标url
:param mode: 'get', 'post' 中选择
:param data: post方式要提交的数据
:param show_errmsg: 是否显示和抛出异常
:param kwargs: 其它参数
:return: Response对象
"""
@ -316,6 +354,8 @@ class SessionPage(object):
elif mode == 'post':
r = self.session.post(url, data=data, **kwargs)
except Exception as e:
if show_errmsg:
raise e
return None, e
else:
headers = dict(r.headers)