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

3.0.34修复alert时阻塞问题；改进配置类；完善WebPage的timeout设置

Browse Source
This commit is contained in:
g1879 2023-01-12 17:23:27 +08:00
parent ad579c63dd
commit edab95d554
12 changed files with 127 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
DrissionPage/chromium_base.py
View File

@ -667,6 +667,9 @@ class Timeout(object):
self.page_load = 30
self.script = 30
def __repr__(self):
return str({'implicit': self.implicit, 'page_load': self.page_load, 'script': self.script})
class PageLoadStrategy(object):
"""用于设置页面加载策略的类"""

3
DrissionPage/chromium_driver.py
View File

@ -43,6 +43,7 @@ class ChromiumDriver(object):
self.id = kwargs.get("id")
self.type = kwargs.get("type")
self.debug = getenv("DEBUG", False)
self.has_alert = False
self._websocket_url = kwargs.get("webSocketDebuggerUrl")
self._kwargs = kwargs
@ -95,6 +96,8 @@ class ChromiumDriver(object):
return self.method_results[message['id']].get(timeout=q_timeout)
except queue.Empty:
if self.has_alert:
return {'result': {'alert': True}}
if isinstance(timeout, (int, float)) and timeout <= 0:
raise TimeoutError(f"调用{message['method']}超时。")

8
DrissionPage/chromium_page.py
View File

@ -89,9 +89,9 @@ class ChromiumPage(ChromiumBase):
def _set_options(self):
"""从配置中读取设置"""
self.set_timeouts(page_load=self.options.timeouts['pageLoad'] / 1000,
script=self.options.timeouts['script'] / 1000,
implicit=self.options.timeouts['implicit'] / 1000)
self.set_timeouts(page_load=self.options.timeouts['pageLoad'],
script=self.options.timeouts['script'],
implicit=self.options.timeouts['implicit'])
self._page_load_strategy = self.options.page_load_strategy
@property
@ -339,6 +339,7 @@ class ChromiumPage(ChromiumBase):
self._alert.defaultPrompt = None
self._alert.response_accept = kwargs.get('result')
self._alert.response_text = kwargs['userInput']
self._tab_obj.has_alert = False
def _on_alert_open(self, **kwargs):
"""alert出现时触发的方法"""
@ -348,6 +349,7 @@ class ChromiumPage(ChromiumBase):
self._alert.defaultPrompt = kwargs.get('defaultPrompt', None)
self._alert.response_accept = None
self._alert.response_text = None
self._tab_obj.has_alert = True
class Alert(object):

7
DrissionPage/common.py
View File

@ -356,7 +356,7 @@ def unzip(zip_path, to_path):
return [f.extract(f.namelist()[0], path=to_path)]
def get_exe_path_from_port(port):
def get_exe_from_port(port):
"""获取端口号第一条进程的可执行文件路径 \n
:param port: 端口号
:return: 可执行文件的绝对路径
@ -529,8 +529,7 @@ def connect_browser(option):
return None, None
if _port_is_using(ip, port):
chrome_path = get_exe_path_from_port(port) if chrome_path == 'chrome' and system_type == 'windows' \
else chrome_path
chrome_path = get_exe_from_port(port) if chrome_path == 'chrome' and system_type == 'windows' else chrome_path
return chrome_path, None
args = _get_running_args(option)
@ -539,7 +538,7 @@ def connect_browser(option):
try:
debugger = _run_browser(port, chrome_path, args)
if chrome_path == 'chrome' and system_type == 'windows':
chrome_path = get_exe_path_from_port(port)
chrome_path = get_exe_from_port(port)
# 传入的路径找不到，主动在ini文件、注册表、系统变量中找
except FileNotFoundError:

2
DrissionPage/common.pyi
View File

@ -33,7 +33,7 @@ def clean_folder(folder_path: str, ignore: list = None) -> None: ...
def unzip(zip_path: str, to_path: str) -> Union[list, None]: ...
def get_exe_path_from_port(port: Union[str, int]) -> Union[str, None]: ...
def get_exe_from_port(port: Union[str, int]) -> Union[str, None]: ...
def get_pid_from_port(port: Union[str, int]) -> Union[str, None]: ...

50
DrissionPage/config.py
View File

@ -148,7 +148,7 @@ class SessionOptions(object):
self._stream = None
self._trust_env = None
self._max_redirects = None
self.timeout = 10
self._timeout = 10
if read_file:
self.ini_path = ini_path or str(Path(__file__).parent / 'configs.ini')
@ -191,7 +191,12 @@ class SessionOptions(object):
if options_dict.get('max_redirects', None) is not None:
self._max_redirects = options_dict['max_redirects']
self.timeout = options_dict.get('timeout', 10)
self._timeout = options_dict.get('timeout', 10)
@property
def timeout(self):
"""返回timeout属性信息"""
return self._timeout
@property
def headers(self):
@ -205,7 +210,6 @@ class SessionOptions(object):
"""返回cookies设置信息"""
if self._cookies is None:
self._cookies = []
return self._cookies
@property
@ -218,7 +222,6 @@ class SessionOptions(object):
"""返回proxies设置信息"""
if self._proxies is None:
self._proxies = {}
return self._proxies
@property
@ -226,7 +229,6 @@ class SessionOptions(object):
"""返回hooks设置信息"""
if self._hooks is None:
self._hooks = {}
return self._hooks
@property
@ -266,6 +268,11 @@ class SessionOptions(object):
"""返回max_redirects设置信息"""
return self._max_redirects
@timeout.setter
def timeout(self, second):
"""返回timeout属性信息"""
self._timeout = second
@headers.setter
def headers(self, headers):
"""设置headers参数 \n
@ -362,6 +369,13 @@ class SessionOptions(object):
"""
self._max_redirects = max_redirects
def set_timeout(self, second):
"""设置超时信息
:param second: 秒数
:return: 返回当前对象
"""
self._timeout = second
def set_headers(self, headers):
"""设置headers参数 \n
:param headers: 参数值
@ -460,10 +474,7 @@ class DriverOptions(Options):
:param ini_path: ini文件路径，为None则读取默认ini文件
"""
super().__init__()
self._driver_path = None
self._user_data_path = None
self.ini_path = None
self.timeouts = {'implicit': 10000, 'pageLoad': 30000, 'script': 30000}
if read_file:
self.ini_path = ini_path or str(Path(__file__).parent / 'configs.ini')
@ -484,9 +495,12 @@ class DriverOptions(Options):
break
self.timeouts = options_dict.get('timeouts', {'implicit': 10, 'pageLoad': 30, 'script': 30})
self.timeouts['implicit'] *= 1000
self.timeouts['pageLoad'] *= 1000
self.timeouts['script'] *= 1000
return
self._driver_path = None
self.ini_path = None
self.timeouts = {'implicit': 10, 'pageLoad': 30, 'script': 30}
self._debugger_address = '127.0.0.1:9222'
@property
def driver_path(self):
@ -649,14 +663,12 @@ class DriverOptions(Options):
:param script: 脚本运行超时时间
:return: 当前对象
"""
# timeouts = self._caps.get('timeouts', {'implicit': 10, 'pageLoad': 3000, 'script': 3000})
if implicit is not None:
self.timeouts['implicit'] = implicit
if pageLoad is not None:
self.timeouts['pageLoad'] = pageLoad * 1000
self.timeouts['pageLoad'] = pageLoad
if script is not None:
self.timeouts['script'] = script * 1000
# self.timeouts = timeouts
self.timeouts['script'] = script
return self
@ -778,7 +790,7 @@ def chrome_options_to_dict(options):
re_dict = dict()
attrs = ['debugger_address', 'binary_location', 'arguments', 'extensions', 'experimental_options', 'driver_path',
'set_window_rect', 'page_load_strategy']
'page_load_strategy']
options_dir = options.__dir__()
for attr in attrs:
@ -789,9 +801,6 @@ def chrome_options_to_dict(options):
if 'timeouts' in options_dir and 'timeouts' in options._caps:
timeouts = options.__getattribute__('timeouts')
timeouts['implicit'] /= 1000
timeouts['pageLoad'] /= 1000
timeouts['script'] /= 1000
re_dict['timeouts'] = timeouts
return re_dict
@ -809,7 +818,8 @@ def session_options_to_dict(options):
return options
re_dict = dict()
attrs = ['headers', 'proxies', 'hooks', 'params', 'verify', 'stream', 'trust_env', 'max_redirects'] # 'adapters',
attrs = ['headers', 'proxies', 'hooks', 'params', 'verify', 'stream', 'trust_env',
'max_redirects', 'timeout'] # 'adapters',
cookies = options.__getattribute__('_cookies')

18
DrissionPage/config.pyi
View File

@ -58,7 +58,10 @@ class SessionOptions(object):
self._stream: bool = ...
self._trust_env: bool = ...
self._max_redirects: int = ...
self.timeout: float = ...
self._timeout: float = ...
@property
def timeout(self) -> Union[int, float]: ...
@property
def headers(self) -> dict: ...
@ -96,6 +99,9 @@ class SessionOptions(object):
@property
def max_redirects(self) -> int: ...
@timeout.setter
def timeout(self, second: Union[int, float]) -> None: ...
@headers.setter
def headers(self, headers: dict) -> None: ...
@ -132,13 +138,15 @@ class SessionOptions(object):
@max_redirects.setter
def max_redirects(self, max_redirects: int) -> None: ...
def set_headers(self, headers: dict) -> 'SessionOptions': ...
def set_timeout(self, second: Union[int, float]) -> SessionOptions: ...
def set_a_header(self, attr: str, value: str) -> 'SessionOptions': ...
def set_headers(self, headers: dict) -> SessionOptions: ...
def remove_a_header(self, attr: str) -> 'SessionOptions': ...
def set_a_header(self, attr: str, value: str) -> SessionOptions: ...
def set_proxies(self, proxies: dict) -> 'SessionOptions': ...
def remove_a_header(self, attr: str) -> SessionOptions: ...
def set_proxies(self, proxies: dict) -> SessionOptions: ...
def save(self, path: str = None) -> str: ...

4
DrissionPage/mix_page.py
View File

@ -39,8 +39,8 @@ class MixPage(SessionPage, DriverPage, BasePage):
if self._mode == 'd':
try:
timeouts = self.drission.driver_options.timeouts
t = timeout if timeout is not None else timeouts['implicit'] / 1000
self.set_timeouts(t, timeouts['pageLoad'] / 1000, timeouts['script'] / 1000)
t = timeout if timeout is not None else timeouts['implicit']
self.set_timeouts(t, timeouts['pageLoad'], timeouts['script'])
except Exception:
self.timeout = timeout if timeout is not None else 10

88
DrissionPage/web_page.py
View File

@ -37,8 +37,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
self._tab_obj = None
self._is_loading = False
self.timeouts = Timeout(self)
self._set_session_options(session_or_options)
self._set_driver_options(driver_or_options)
self._set_both_options(driver_or_options, session_or_options)
self._setting_tab_id = tab_id
self._has_driver, self._has_session = (None, True) if self._mode == 's' else (True, None)
self._response = None
@ -383,58 +382,61 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
elif self._mode == 'd':
return super(SessionPage, self)._ele(loc_or_ele, timeout=timeout, single=single, relative=relative)
def _set_driver_options(self, driver_or_Options):
"""处理driver设置
:param driver_or_Options: ChromiumDriver对象或DriverOptions对象
def _set_both_options(self, dr_opt, se_opt):
"""处理两种模式的设置
:param dr_opt: ChromiumDriver或DriverOptions对象，为None则从ini读取，为False用默认信息创建
:param se_opt: Session、SessionOptions对象或配置信息，为None则从ini读取，为False用默认信息创建
:return: None
"""
if isinstance(driver_or_Options, ChromiumDriver):
self._connect_browser(driver_or_Options)
if isinstance(dr_opt, ChromiumDriver):
self._connect_browser(dr_opt)
self._has_driver = True
return
if driver_or_Options is None:
self._driver_options = DriverOptions()
elif driver_or_Options is False:
self._driver_options = DriverOptions(read_file=False)
elif isinstance(driver_or_Options, DriverOptions):
self._driver_options = driver_or_Options
self._driver_options = None
dr_opt = False
else:
raise TypeError('driver_or_options参数只能接收WebDriver, Options, DriverOptions或False。')
if dr_opt is None:
self._driver_options = DriverOptions()
timeouts = self._driver_options.timeouts
self.set_timeouts(timeouts['implicit'], timeouts['pageLoad'], timeouts['script'])
elif dr_opt is False:
self._driver_options = DriverOptions(read_file=False)
def _set_session_options(self, Session_or_Options):
"""处理session设置
:param Session_or_Options: Session对象或SessionOptions对象
:return: None
"""
if isinstance(Session_or_Options, Session):
self._session = Session_or_Options
elif isinstance(dr_opt, DriverOptions):
self._driver_options = dr_opt
else:
raise TypeError('driver_or_options参数只能接收ChromiumDriver, DriverOptions、None或False。')
if isinstance(se_opt, Session):
self._session = se_opt
self._has_session = True
return
if Session_or_Options is None:
so = SessionOptions()
elif Session_or_Options is False:
so = SessionOptions(read_file=False)
elif isinstance(Session_or_Options, SessionOptions):
so = Session_or_Options
elif isinstance(Session_or_Options, dict):
so = Session_or_Options
self._session_options = None
se_opt = False
else:
raise TypeError('session_or_options参数只能接收Session, dict, SessionOptions或False。')
if se_opt is None:
so = SessionOptions().as_dict()
self._session_options = so.as_dict()
self.set_timeouts(implicit=so.timeout)
elif se_opt is False:
so = SessionOptions(read_file=False).as_dict()
elif isinstance(se_opt, SessionOptions):
so = se_opt.as_dict()
elif isinstance(se_opt, dict):
so = se_opt
else:
raise TypeError('session_or_options参数只能接收Session, dict, SessionOptions、None或False。')
self._session_options = so
if se_opt is not False:
self.set_timeouts(implicit=self._session_options.get('timeout', 10))
if dr_opt is not False:
t = self._driver_options.timeouts
self.set_timeouts(t['implicit'], t['pageLoad'], t['script'])
def quit(self):
"""关闭浏览器，关闭session"""

5
DrissionPage/web_page.pyi
View File

@ -22,7 +22,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
def __init__(self,
mode: str = 'd',
timeout: float = 10,
timeout: float = None,
tab_id: str = None,
driver_or_options: Union[ChromiumDriver, DriverOptions, bool] = None,
session_or_options: Union[Session, SessionOptions, bool] = None) -> None:
@ -159,6 +159,9 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
-> Union[ChromiumElement, SessionElement, ChromiumFrame, str, None, List[Union[SessionElement, str]], List[
Union[ChromiumElement, str, ChromiumFrame]]]: ...
def _set_both_options(self, dr_opt: Union[ChromiumDriver, DriverOptions],
se_opt: Union[Session, SessionOptions, dict, bool, None]) -> None: ...
def _set_driver_options(self, driver_or_Options: Union[ChromiumDriver, DriverOptions]) -> None: ...
def _set_session_options(self, Session_or_Options: Union[Session, SessionOptions]) -> None: ...

23
README.md
View File

@ -1,5 +1,3 @@
#
DrissionPage 是一个基于 python 的网页自动化工具。
它既能控制浏览器，也能收发数据包，甚至能把两者合而为一，
@ -10,9 +8,21 @@ DrissionPage 是一个基于 python 的网页自动化工具。
它的语法简洁而优雅，代码量少，对新手友好。
**使用文档：** 📒[点击打开](http://g1879.gitee.io/drissionpage)
***
**QQ群：** 897838127
支持系统：Windows、Linux、Mac
python 版本：3.6 及以上
支持浏览器：Chromium 内核浏览器（如 Chrome 和 edge）
***
<a href='https://gitee.com/g1879/DrissionPage/stargazers'><img src='https://gitee.com/g1879/DrissionPage/badge/star.svg?theme=dark' alt='star'></img></a> <a href='https://gitee.com/g1879/DrissionPage/members'><img src='https://gitee.com/g1879/DrissionPage/badge/fork.svg?theme=dark' alt='fork'></img></a>
项目地址：[gitee](https://gitee.com/g1879/DrissionPage) | [github](https://github.com/g1879/DrissionPage)
**交流QQ群：** 897838127
**联系邮箱：** g1879@qq.com
@ -22,8 +32,9 @@ DrissionPage 是一个基于 python 的网页自动化工具。
使用浏览器，可以很大程度上绕过这些坑，但浏览器运行效率不高。
因此，这个库设计初衷，是将它们合而为一，能够在不同须要时切换相应模式，并提供一种人性化的使用方法，提高开发和运行效率。
除了合并两者，本库还以网页为单位封装了常用功能，提供非常简便的操作和语句，在用于网页自动化操作时，减少考虑细节，专注功能实现，使用更方便。
一切从简，尽量提供简单直接的使用方法，对新手更友好。
除了合并两者，本库还以网页为单位封装了常用功能，提供非常简便的操作和语句，在用于网页自动化操作时，减少考虑细节，专注功能实现，使用更方便。 一切从简，尽量提供简单直接的使用方法，使代码更优雅。
以前的版本是对 selenium 进行重新封装实现的。从 3.0 开始，作者另起炉灶，对底层进行了重新开发，摆脱对 selenium 的依赖，增强了功能，提升了运行效率。
# 💡 理念

2
setup.py
View File

@ -6,7 +6,7 @@ with open("README.md", "r", encoding='utf-8') as fh:
setup(
name="DrissionPage",
version="3.0.33",
version="3.0.34",
author="g1879",
author_email="g1879@qq.com",
description="A module that integrates selenium and requests session, encapsulates common page operations.",