diff --git a/DrissionPage/driver_page.py b/DrissionPage/driver_page.py index b3aaef9..e6993c7 100644 --- a/DrissionPage/driver_page.py +++ b/DrissionPage/driver_page.py @@ -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, diff --git a/DrissionPage/mix_page.py b/DrissionPage/mix_page.py index 10f7eb7..3967003 100644 --- a/DrissionPage/mix_page.py +++ b/DrissionPage/mix_page.py @@ -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], diff --git a/DrissionPage/session_page.py b/DrissionPage/session_page.py index 9170cce..f22dca8 100644 --- a/DrissionPage/session_page.py +++ b/DrissionPage/session_page.py @@ -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)