优化cookies处理逻辑,待测试

This commit is contained in:
g1879 2023-03-11 23:19:56 +08:00
parent 8fc8719547
commit 3832e75a31
6 changed files with 48 additions and 61 deletions

View File

@ -8,11 +8,9 @@ from json import loads, JSONDecodeError
from os import sep
from pathlib import Path
from time import perf_counter, sleep, time
from urllib.parse import urlparse
from warnings import warn
from requests import Session
from tldextract import extract
from .base import BasePage
from .chromium_driver import ChromiumDriver
@ -20,7 +18,7 @@ from .chromium_element import ChromiumScroll, ChromiumElement, run_js, make_chro
from .commons.constants import HANDLE_ALERT_METHOD, ERROR, NoneElement
from .commons.locator import get_loc
from .commons.tools import get_usable_path, clean_folder
from .commons.web import cookies_to_tuple
from .commons.web import set_browser_cookies
from .errors import ContextLossError, ElementLossError, AlertExistsError, CallMethodError, TabClosedError, \
NoRectError, BrowserConnectError
from .session_element import make_session_ele
@ -264,6 +262,11 @@ class ChromiumBase(BasePage):
"""返回当前页面url"""
return self.run_cdp_loaded('Target.getTargetInfo', targetId=self.tab_id)['targetInfo']['url']
@property
def _browser_url(self):
"""用于被WebPage覆盖"""
return self.url
@property
def html(self):
"""返回当前页面html文本"""
@ -923,29 +926,7 @@ class ChromiumBaseSetter(object):
:param cookies: cookies信息
:return: None
"""
cookies = cookies_to_tuple(cookies)
result_cookies = []
for cookie in cookies:
# todo: 须要吗?
# if 'expiry' in cookie:
# cookie['expiry'] = int(cookie['expiry'])
if not cookie.get('domain', None):
print(cookie)
ex_url = extract(self._page.url)
cookie['domain'] = f'{ex_url.domain}.{ex_url.suffix}' if ex_url.suffix else ex_url.domain
# netloc = urlparse(self._page.url).netloc
# if netloc.replace('.', '').isdigit(): # ip
# cookie['domain'] = netloc
# else: # 域名
# u = netloc.split('.')
# cookie['domain'] = f'.{u[-2]}.{u[-1]}' if len(u) > 1 else netloc
result_cookies.append({'value': '' if cookie['value'] is None else cookie['value'],
'name': cookie['name'],
'domain': cookie['domain']})
self._page.run_cdp_loaded('Network.setCookies', cookies=result_cookies)
set_browser_cookies(self._page, cookies)
def upload_files(self, files):
"""等待上传的文件路径

View File

@ -86,6 +86,9 @@ class ChromiumBase(BasePage):
@property
def url(self) -> str: ...
@property
def _browser_url(self) -> str: ...
@property
def html(self) -> str: ...
@ -138,7 +141,8 @@ class ChromiumBase(BasePage):
interval: float = None,
timeout: float = None) -> Union[None, bool]: ...
def get_cookies(self, as_dict: bool = False, all_domains: bool =False, all_info: bool = False) -> Union[list, dict]: ...
def get_cookies(self, as_dict: bool = False, all_domains: bool = False, all_info: bool = False) -> Union[
list, dict]: ...
def ele(self,
loc_or_ele: Union[Tuple[str, str], str, ChromiumElement, ChromiumFrame],

View File

@ -5,10 +5,8 @@
"""
from copy import copy
from tldextract import extract
from .chromium_base import ChromiumBase, ChromiumBaseSetter
from .commons.web import set_session_cookies
from .commons.web import set_session_cookies, set_browser_cookies
from .session_page import SessionPage, SessionPageSetter, DownloadSetter
@ -296,18 +294,7 @@ class WebPageTab(SessionPage, ChromiumTab):
if not self._has_driver:
return
ex_url = extract(self._session_url)
domain = f'{ex_url.domain}.{ex_url.suffix}' if ex_url.suffix else ex_url.domain
cookies = []
for cookie in super().get_cookies():
if not cookie.get('domain', None):
cookie['domain'] = domain
if domain in cookie['domain']:
cookies.append(cookie)
self.run_cdp_loaded('Network.setCookies', cookies=cookies)
set_browser_cookies(self, super().get_cookies()) # todo: cookies的选择
def get_cookies(self, as_dict=False, all_domains=False, all_info=False):
"""返回cookies

View File

@ -9,6 +9,7 @@ from re import sub
from urllib.parse import urlparse, urljoin, urlunparse
from requests.cookies import RequestsCookieJar
from tldextract import extract
def get_ele_txt(e):
@ -237,3 +238,30 @@ def set_session_cookies(session, cookies):
kwargs['expires'] = cookie['expiry']
session.cookies.set(cookie['name'], cookie['value'], **kwargs)
def set_browser_cookies(page, cookies):
"""设置cookies值
:param page: 页面对象
:param cookies: cookies信息
:return: None
"""
cookies = cookies_to_tuple(cookies)
for cookie in cookies:
if 'expiry' in cookie:
cookie['expires'] = int(cookie['expiry'])
cookie.pop('expiry')
if 'expires' in cookie:
cookie['expires'] = int(cookie['expires'])
if not cookie.get('domain', None):
ex_url = extract(page._browser_url)
cookie['domain'] = f'{ex_url.domain}.{ex_url.suffix}' if ex_url.suffix else ex_url.domain
if cookie['value'] is None:
cookie['value'] = ''
try:
page.run_cdp_loaded('Network.setCookie', **cookie)
except Exception as e:
pass

View File

@ -11,6 +11,7 @@ from requests.cookies import RequestsCookieJar
from DrissionPage.base import DrissionElement, BasePage
from DrissionPage.chromium_element import ChromiumElement
from DrissionPage.chromium_base import ChromiumBase
def get_ele_txt(e: DrissionElement) -> str: ...
@ -38,3 +39,6 @@ def cookies_to_tuple(cookies: Union[RequestsCookieJar, list, tuple, str, dict])
def set_session_cookies(session: Session, cookies: Union[RequestsCookieJar, list, tuple, str, dict]) -> None: ...
def set_browser_cookies(page: ChromiumBase, cookies: Union[RequestsCookieJar, list, tuple, str, dict]) -> None: ...

View File

@ -7,14 +7,13 @@ from pathlib import Path
from warnings import warn
from requests import Session
from tldextract import extract
from .commons.web import set_session_cookies
from .base import BasePage
from .chromium_base import ChromiumBase, Timeout
from .chromium_driver import ChromiumDriver
from .chromium_page import ChromiumPage, ChromiumDownloadSetter, ChromiumPageSetter
from .chromium_tab import WebPageTab
from .commons.web import set_session_cookies, set_browser_cookies
from .configs.chromium_options import ChromiumOptions
from .configs.session_options import SessionOptions
from .errors import CallMethodError
@ -370,23 +369,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
if not self._has_driver:
return
ex_url = extract(self._session_url)
domain = f'{ex_url.domain}.{ex_url.suffix}' if ex_url.suffix else ex_url.domain
cookies = []
for cookie in super().get_cookies(): # 带域名list
if not cookie.get('domain', None):
cookie['domain'] = domain
if domain in cookie['domain']:
cookies.append(cookie)
# self.run_cdp_loaded('Network.setCookies', cookies=cookies)
for c in cookies:
try:
self.run_cdp_loaded('Network.setCookie', name=c['name'], value=c['value'], domain=c['domain'])
except Exception as e:
print(e)
set_browser_cookies(self, super().get_cookies(all_domains=True)) # todo: cookies的选择
def get_cookies(self, as_dict=False, all_domains=False, all_info=False):
"""返回cookies