This commit is contained in:
g1879 2020-06-17 23:24:03 +08:00
commit cd3207e076
4 changed files with 33 additions and 10 deletions

View File

@ -22,18 +22,20 @@ class Drission(object):
"""Drission类整合了WebDriver对象和HTLSession对象可按要求创建、关闭及同步cookies"""
def __init__(self, driver_options: Union[dict, Options] = None, session_options: dict = None,
driver_path: str = None, ini_path: str = None):
driver_path: str = None, ini_path: str = None, proxy: dict = None):
"""初始化配置信息但不生成session和driver实例
:param driver_options: chrome设置Options类或设置字典
:param session_options: session设置
:param driver_path: chromedriver路径如为空则为'chromedriver'
:param ini_path: ini文件路径'
:param proxy: 代理设置
"""
self._session = None
self._driver = None
om = OptionsManager(ini_path)
self._session_options = session_options or om.get_option('session_options')
self._driver_options = _chrome_options_to_dict(driver_options) or om.get_option('chrome_options')
self._proxy = proxy
if driver_path:
self._driver_path = driver_path
@ -43,7 +45,7 @@ class Drission(object):
self._driver_path = 'chromedriver'
@property
def session(self):
def session(self) -> HTMLSession:
"""获取HTMLSession对象"""
if self._session is None:
self._session = HTMLSession()
@ -52,17 +54,21 @@ class Drission(object):
for i in attrs:
if i in self._session_options:
exec(f'self._session.{i} = self._session_options["{i}"]')
if self._proxy:
self._session.proxies = self._proxy
return self._session
@property
def driver(self):
def driver(self) -> WebDriver:
"""获取WebDriver对象按传入配置信息初始化"""
if self._driver is None:
if isinstance(self._driver_options, dict):
options = _dict_to_chrome_options(self._driver_options)
else:
raise KeyError('Driver options invalid')
if self._proxy:
options.add_argument(f'--proxy-server={self._proxy["http"]}')
self._driver = webdriver.Chrome(self._driver_path, options=options)
@ -89,6 +95,25 @@ class Drission(object):
def session_options(self, value: dict):
self._session_options = value
@property
def proxy(self) -> Union[None, dict]:
return self._proxy
@proxy.setter
def proxy(self, proxies: dict = None):
self._proxy = proxies
if self._session:
self._session.proxies = proxies
if self._driver:
cookies = self._driver.get_cookies()
url = self._driver.current_url
self._driver.quit()
self._driver = None
self._driver = self.driver
self._driver.get(url)
for cookie in cookies:
self._driver.add_cookie(cookie)
def cookies_to_session(self, copy_user_agent: bool = False, driver: WebDriver = None, session: Session = None) \
-> None:
"""把driver的cookies复制到session"""

View File

@ -147,7 +147,7 @@ class DriverPage(object):
self.driver.switch_to.default_content()
elif loc_or_ele == 'parent':
self.driver.switch_to.parent_frame()
elif ':' not in loc_or_ele:
elif ':' not in loc_or_ele and '=' not in loc_or_ele:
# 传入id或name
self.driver.switch_to.frame(loc_or_ele)
else:

View File

@ -110,6 +110,7 @@ class MixPage(Null, SessionPage, DriverPage):
每次访问时切换到d模式主要用于独有函数及外部调用
:return:selenium的WebDriver对象
"""
# TODO: 改成每次获取drission的driver
if self._driver is None:
self._driver = self._drission.driver
self.change_mode('d')
@ -153,8 +154,7 @@ class MixPage(Null, SessionPage, DriverPage):
# ----------------重写SessionPage的函数-----------------------
def post(self, url: str, params: dict = None, data: dict = None, go_anyway: bool = False, **kwargs) \
-> Union[bool, None]:
def post(self, url: str, data: dict = None, go_anyway: bool = False, **kwargs) -> Union[bool, None]:
"""post前先转换模式但不跳转"""
self.change_mode('s', go=False)
return super().post(url, data, go_anyway, **kwargs)

View File

@ -61,7 +61,6 @@ class SessionPage(object):
@property
def html(self) -> str:
"""获取元素innerHTML如未指定元素则获取所有源代码"""
# return unescape(self.response.html.raw_html.replace(b'\x08', b'').decode()).replace('\xa0', ' ')
return self.response.html.html
def ele(self, loc_or_ele: Union[tuple, str, SessionElement], mode: str = None, show_errmsg: bool = False) \
@ -85,7 +84,7 @@ class SessionPage(object):
"""查找符合条件的所有元素"""
return self.ele(loc, 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, **kwargs) -> Union[bool, None]:
"""用get方式跳转到url调用_make_response()函数生成response对象"""
to_url = quote(url, safe='/:&?=%;#@')
if not url or (not go_anyway and self.url == to_url):
@ -97,8 +96,7 @@ class SessionPage(object):
self._url_available = True if self._response and self._response.ok else False
return self._url_available
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, **kwargs) -> Union[bool, None]:
"""用post方式跳转到url调用_make_response()函数生成response对象"""
to_url = quote(url, safe='/:&?=%;#@')
if not url or (not go_anyway and self._url == to_url):