mirror of
https://gitee.com/g1879/DrissionPage.git
synced 2024-12-10 04:00:23 +08:00
Setter重构到专属文件,待测试
This commit is contained in:
parent
b62c3cb6a1
commit
0fd4d724f6
@ -18,11 +18,11 @@ from .chromium_element import ChromiumScroll, ChromiumElement, run_js, make_chro
|
|||||||
from .commons.constants import HANDLE_ALERT_METHOD, ERROR, NoneElement
|
from .commons.constants import HANDLE_ALERT_METHOD, ERROR, NoneElement
|
||||||
from .commons.locator import get_loc
|
from .commons.locator import get_loc
|
||||||
from .commons.tools import get_usable_path, clean_folder
|
from .commons.tools import get_usable_path, clean_folder
|
||||||
from .commons.web import set_browser_cookies
|
|
||||||
from .errors import ContextLossError, ElementLossError, AlertExistsError, CDPError, TabClosedError, \
|
from .errors import ContextLossError, ElementLossError, AlertExistsError, CDPError, TabClosedError, \
|
||||||
NoRectError, BrowserConnectError, GetDocumentError
|
NoRectError, BrowserConnectError, GetDocumentError
|
||||||
from .network_listener import NetworkListener
|
from .network_listener import NetworkListener
|
||||||
from .session_element import make_session_ele
|
from .session_element import make_session_ele
|
||||||
|
from .setter import ChromiumBaseSetter
|
||||||
from .waiter import ChromiumBaseWaiter
|
from .waiter import ChromiumBaseWaiter
|
||||||
|
|
||||||
|
|
||||||
@ -890,112 +890,6 @@ class ChromiumBase(BasePage):
|
|||||||
return str(path.absolute())
|
return str(path.absolute())
|
||||||
|
|
||||||
|
|
||||||
class ChromiumBaseSetter(object):
|
|
||||||
def __init__(self, page):
|
|
||||||
self._page = page
|
|
||||||
|
|
||||||
@property
|
|
||||||
def load_strategy(self):
|
|
||||||
"""返回用于设置页面加载策略的对象"""
|
|
||||||
return PageLoadStrategy(self._page)
|
|
||||||
|
|
||||||
@property
|
|
||||||
def scroll(self):
|
|
||||||
"""返回用于设置页面滚动设置的对象"""
|
|
||||||
return PageScrollSetter(self._page.scroll)
|
|
||||||
|
|
||||||
def retry_times(self, times):
|
|
||||||
"""设置连接失败重连次数"""
|
|
||||||
self._page.retry_times = times
|
|
||||||
|
|
||||||
def retry_interval(self, interval):
|
|
||||||
"""设置连接失败重连间隔"""
|
|
||||||
self._page.retry_interval = interval
|
|
||||||
|
|
||||||
def timeouts(self, implicit=None, page_load=None, script=None):
|
|
||||||
"""设置超时时间,单位为秒
|
|
||||||
:param implicit: 查找元素超时时间
|
|
||||||
:param page_load: 页面加载超时时间
|
|
||||||
:param script: 脚本运行超时时间
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if implicit is not None:
|
|
||||||
self._page.timeouts.implicit = implicit
|
|
||||||
|
|
||||||
if page_load is not None:
|
|
||||||
self._page.timeouts.page_load = page_load
|
|
||||||
|
|
||||||
if script is not None:
|
|
||||||
self._page.timeouts.script = script
|
|
||||||
|
|
||||||
def user_agent(self, ua, platform=None):
|
|
||||||
"""为当前tab设置user agent,只在当前tab有效
|
|
||||||
:param ua: user agent字符串
|
|
||||||
:param platform: platform字符串
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
keys = {'userAgent': ua}
|
|
||||||
if platform:
|
|
||||||
keys['platform'] = platform
|
|
||||||
self._page.run_cdp('Emulation.setUserAgentOverride', **keys)
|
|
||||||
|
|
||||||
def session_storage(self, item, value):
|
|
||||||
"""设置或删除某项sessionStorage信息
|
|
||||||
:param item: 要设置的项
|
|
||||||
:param value: 项的值,设置为False时,删除该项
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
js = f'sessionStorage.removeItem("{item}");' if item is False else f'sessionStorage.setItem("{item}","{value}");'
|
|
||||||
return self._page.run_js_loaded(js, as_expr=True)
|
|
||||||
|
|
||||||
def local_storage(self, item, value):
|
|
||||||
"""设置或删除某项localStorage信息
|
|
||||||
:param item: 要设置的项
|
|
||||||
:param value: 项的值,设置为False时,删除该项
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
js = f'localStorage.removeItem("{item}");' if item is False else f'localStorage.setItem("{item}","{value}");'
|
|
||||||
return self._page.run_js_loaded(js, as_expr=True)
|
|
||||||
|
|
||||||
def cookie(self, cookie):
|
|
||||||
"""设置单个cookie
|
|
||||||
:param cookie: cookie信息
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if isinstance(cookie, str):
|
|
||||||
self.cookies(cookie)
|
|
||||||
else:
|
|
||||||
self.cookies([cookie])
|
|
||||||
|
|
||||||
def cookies(self, cookies):
|
|
||||||
"""设置多个cookie,注意不要传入单个
|
|
||||||
:param cookies: cookies信息
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
set_browser_cookies(self._page, cookies)
|
|
||||||
|
|
||||||
def upload_files(self, files):
|
|
||||||
"""等待上传的文件路径
|
|
||||||
:param files: 文件路径列表或字符串,字符串时多个文件用回车分隔
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if not self._page._upload_list:
|
|
||||||
self._page.driver.Page.fileChooserOpened = self._page._onFileChooserOpened
|
|
||||||
self._page.run_cdp('Page.setInterceptFileChooserDialog', enabled=True)
|
|
||||||
|
|
||||||
if isinstance(files, str):
|
|
||||||
files = files.split('\n')
|
|
||||||
self._page._upload_list = [str(Path(i).absolute()) for i in files]
|
|
||||||
|
|
||||||
def headers(self, headers: dict) -> None:
|
|
||||||
"""设置固定发送的headers
|
|
||||||
:param headers: dict格式的headers数据
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.run_cdp('Network.enable')
|
|
||||||
self._page.run_cdp('Network.setExtraHTTPHeaders', headers=headers)
|
|
||||||
|
|
||||||
|
|
||||||
class ChromiumPageScroll(ChromiumScroll):
|
class ChromiumPageScroll(ChromiumScroll):
|
||||||
def __init__(self, page):
|
def __init__(self, page):
|
||||||
"""
|
"""
|
||||||
@ -1055,62 +949,6 @@ class Timeout(object):
|
|||||||
return str({'implicit': self.implicit, 'page_load': self.page_load, 'script': self.script})
|
return str({'implicit': self.implicit, 'page_load': self.page_load, 'script': self.script})
|
||||||
|
|
||||||
|
|
||||||
class PageLoadStrategy(object):
|
|
||||||
"""用于设置页面加载策略的类"""
|
|
||||||
|
|
||||||
def __init__(self, page):
|
|
||||||
"""
|
|
||||||
:param page: ChromiumBase对象
|
|
||||||
"""
|
|
||||||
self._page = page
|
|
||||||
|
|
||||||
def __call__(self, value):
|
|
||||||
"""设置加载策略
|
|
||||||
:param value: 可选 'normal', 'eager', 'none'
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if value.lower() not in ('normal', 'eager', 'none'):
|
|
||||||
raise ValueError("只能选择 'normal', 'eager', 'none'。")
|
|
||||||
self._page._page_load_strategy = value
|
|
||||||
|
|
||||||
def normal(self):
|
|
||||||
"""设置页面加载策略为normal"""
|
|
||||||
self._page._page_load_strategy = 'normal'
|
|
||||||
|
|
||||||
def eager(self):
|
|
||||||
"""设置页面加载策略为eager"""
|
|
||||||
self._page._page_load_strategy = 'eager'
|
|
||||||
|
|
||||||
def none(self):
|
|
||||||
"""设置页面加载策略为none"""
|
|
||||||
self._page._page_load_strategy = 'none'
|
|
||||||
|
|
||||||
|
|
||||||
class PageScrollSetter(object):
|
|
||||||
def __init__(self, scroll):
|
|
||||||
self._scroll = scroll
|
|
||||||
|
|
||||||
def wait_complete(self, on_off=True):
|
|
||||||
"""设置滚动命令后是否等待完成
|
|
||||||
:param on_off: 开或关
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if not isinstance(on_off, bool):
|
|
||||||
raise TypeError('on_off必须为bool。')
|
|
||||||
self._scroll._wait_complete = on_off
|
|
||||||
|
|
||||||
def smooth(self, on_off=True):
|
|
||||||
"""设置页面滚动是否平滑滚动
|
|
||||||
:param on_off: 开或关
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if not isinstance(on_off, bool):
|
|
||||||
raise TypeError('on_off必须为bool。')
|
|
||||||
b = 'smooth' if on_off else 'auto'
|
|
||||||
self._scroll._driver.run_js(f'document.documentElement.style.setProperty("scroll-behavior","{b}");')
|
|
||||||
self._scroll._wait_complete = on_off
|
|
||||||
|
|
||||||
|
|
||||||
class Screencast(object):
|
class Screencast(object):
|
||||||
def __init__(self, page):
|
def __init__(self, page):
|
||||||
self._page = page
|
self._page = page
|
||||||
|
@ -8,9 +8,7 @@ from typing import Union, Tuple, List, Any
|
|||||||
|
|
||||||
from DataRecorder import Recorder
|
from DataRecorder import Recorder
|
||||||
from requests import Session
|
from requests import Session
|
||||||
from requests.cookies import RequestsCookieJar
|
|
||||||
|
|
||||||
from .waiter import ChromiumBaseWaiter
|
|
||||||
from .base import BasePage
|
from .base import BasePage
|
||||||
from .chromium_driver import ChromiumDriver
|
from .chromium_driver import ChromiumDriver
|
||||||
from .chromium_element import ChromiumElement, ChromiumScroll
|
from .chromium_element import ChromiumElement, ChromiumScroll
|
||||||
@ -18,6 +16,8 @@ from .chromium_frame import ChromiumFrame
|
|||||||
from .commons.constants import NoneElement
|
from .commons.constants import NoneElement
|
||||||
from .network_listener import NetworkListener
|
from .network_listener import NetworkListener
|
||||||
from .session_element import SessionElement
|
from .session_element import SessionElement
|
||||||
|
from .setter import ChromiumBaseSetter
|
||||||
|
from .waiter import ChromiumBaseWaiter
|
||||||
|
|
||||||
|
|
||||||
class ChromiumBase(BasePage):
|
class ChromiumBase(BasePage):
|
||||||
@ -222,37 +222,6 @@ class ChromiumPageScroll(ChromiumScroll):
|
|||||||
def _to_see(self, ele: ChromiumElement, center: Union[bool, None]) -> None: ...
|
def _to_see(self, ele: ChromiumElement, center: Union[bool, None]) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
class ChromiumBaseSetter(object):
|
|
||||||
def __init__(self, page):
|
|
||||||
self._page: ChromiumBase = ...
|
|
||||||
|
|
||||||
@property
|
|
||||||
def load_strategy(self) -> PageLoadStrategy: ...
|
|
||||||
|
|
||||||
@property
|
|
||||||
def scroll(self) -> PageScrollSetter: ...
|
|
||||||
|
|
||||||
def retry_times(self, times: int) -> None: ...
|
|
||||||
|
|
||||||
def retry_interval(self, interval: float) -> None: ...
|
|
||||||
|
|
||||||
def timeouts(self, implicit: float = None, page_load: float = None, script: float = None) -> None: ...
|
|
||||||
|
|
||||||
def user_agent(self, ua: str, platform: str = None) -> None: ...
|
|
||||||
|
|
||||||
def session_storage(self, item: str, value: Union[str, bool]) -> None: ...
|
|
||||||
|
|
||||||
def local_storage(self, item: str, value: Union[str, bool]) -> None: ...
|
|
||||||
|
|
||||||
def cookie(self, cookies: Union[RequestsCookieJar, str, dict]) -> None: ...
|
|
||||||
|
|
||||||
def cookies(self, cookies: Union[RequestsCookieJar, list, tuple, str, dict]) -> None: ...
|
|
||||||
|
|
||||||
def headers(self, headers: dict) -> None: ...
|
|
||||||
|
|
||||||
def upload_files(self, files: Union[str, list, tuple]) -> None: ...
|
|
||||||
|
|
||||||
|
|
||||||
class Timeout(object):
|
class Timeout(object):
|
||||||
|
|
||||||
def __init__(self, page: ChromiumBase, implicit=None, page_load=None, script=None):
|
def __init__(self, page: ChromiumBase, implicit=None, page_load=None, script=None):
|
||||||
@ -262,28 +231,6 @@ class Timeout(object):
|
|||||||
self.script: float = ...
|
self.script: float = ...
|
||||||
|
|
||||||
|
|
||||||
class PageLoadStrategy(object):
|
|
||||||
def __init__(self, page: ChromiumBase):
|
|
||||||
self._page: ChromiumBase = ...
|
|
||||||
|
|
||||||
def __call__(self, value: str) -> None: ...
|
|
||||||
|
|
||||||
def normal(self) -> None: ...
|
|
||||||
|
|
||||||
def eager(self) -> None: ...
|
|
||||||
|
|
||||||
def none(self) -> None: ...
|
|
||||||
|
|
||||||
|
|
||||||
class PageScrollSetter(object):
|
|
||||||
def __init__(self, scroll: ChromiumPageScroll):
|
|
||||||
self._scroll: ChromiumPageScroll = ...
|
|
||||||
|
|
||||||
def wait_complete(self, on_off: bool = True): ...
|
|
||||||
|
|
||||||
def smooth(self, on_off: bool = True): ...
|
|
||||||
|
|
||||||
|
|
||||||
class Screencast(object):
|
class Screencast(object):
|
||||||
def __init__(self, page: ChromiumBase):
|
def __init__(self, page: ChromiumBase):
|
||||||
self._page: ChromiumBase = ...
|
self._page: ChromiumBase = ...
|
||||||
|
@ -16,6 +16,7 @@ from .commons.web import make_absolute_link, get_ele_txt, format_html, is_js_fun
|
|||||||
from .errors import ContextLossError, ElementLossError, JavaScriptError, NoRectError, ElementNotFoundError, \
|
from .errors import ContextLossError, ElementLossError, JavaScriptError, NoRectError, ElementNotFoundError, \
|
||||||
CDPError, NoResourceError, CanNotClickError
|
CDPError, NoResourceError, CanNotClickError
|
||||||
from .session_element import make_session_ele
|
from .session_element import make_session_ele
|
||||||
|
from .setter import ChromiumElementSetter
|
||||||
from .waiter import ChromiumElementWaiter
|
from .waiter import ChromiumElementWaiter
|
||||||
|
|
||||||
|
|
||||||
@ -1467,38 +1468,6 @@ class ShadowRootStates(object):
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
class ChromiumElementSetter(object):
|
|
||||||
def __init__(self, ele):
|
|
||||||
"""
|
|
||||||
:param ele: ChromiumElement
|
|
||||||
"""
|
|
||||||
self._ele = ele
|
|
||||||
|
|
||||||
def attr(self, attr, value):
|
|
||||||
"""设置元素attribute属性
|
|
||||||
:param attr: 属性名
|
|
||||||
:param value: 属性值
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._ele.page.run_cdp('DOM.setAttributeValue', nodeId=self._ele.ids.node_id, name=attr, value=str(value))
|
|
||||||
|
|
||||||
def prop(self, prop, value):
|
|
||||||
"""设置元素property属性
|
|
||||||
:param prop: 属性名
|
|
||||||
:param value: 属性值
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
value = value.replace('"', r'\"')
|
|
||||||
self._ele.run_js(f'this.{prop}="{value}";')
|
|
||||||
|
|
||||||
def innerHTML(self, html):
|
|
||||||
"""设置元素innerHTML
|
|
||||||
:param html: html文本
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self.prop('innerHTML', html)
|
|
||||||
|
|
||||||
|
|
||||||
class Locations(object):
|
class Locations(object):
|
||||||
def __init__(self, ele):
|
def __init__(self, ele):
|
||||||
"""
|
"""
|
||||||
|
@ -12,6 +12,7 @@ from .chromium_frame import ChromiumFrame
|
|||||||
from .chromium_page import ChromiumPage
|
from .chromium_page import ChromiumPage
|
||||||
from .commons.constants import NoneElement
|
from .commons.constants import NoneElement
|
||||||
from .session_element import SessionElement
|
from .session_element import SessionElement
|
||||||
|
from .setter import ChromiumElementSetter
|
||||||
from .waiter import ChromiumElementWaiter
|
from .waiter import ChromiumElementWaiter
|
||||||
from .web_page import WebPage
|
from .web_page import WebPage
|
||||||
|
|
||||||
@ -383,17 +384,6 @@ def send_enter(ele: ChromiumElement) -> None: ...
|
|||||||
def send_key(ele: ChromiumElement, modifier: int, key: str) -> None: ...
|
def send_key(ele: ChromiumElement, modifier: int, key: str) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
class ChromiumElementSetter(object):
|
|
||||||
def __init__(self, ele: ChromiumElement):
|
|
||||||
self._ele: ChromiumElement = ...
|
|
||||||
|
|
||||||
def attr(self, attr: str, value: str) -> None: ...
|
|
||||||
|
|
||||||
def prop(self, prop: str, value: str) -> None: ...
|
|
||||||
|
|
||||||
def innerHTML(self, html: str) -> None: ...
|
|
||||||
|
|
||||||
|
|
||||||
class ShadowRootStates(object):
|
class ShadowRootStates(object):
|
||||||
def __init__(self, ele: ChromiumShadowRoot):
|
def __init__(self, ele: ChromiumShadowRoot):
|
||||||
"""
|
"""
|
||||||
|
@ -7,10 +7,11 @@ from re import search
|
|||||||
from threading import Thread
|
from threading import Thread
|
||||||
from time import sleep, perf_counter
|
from time import sleep, perf_counter
|
||||||
|
|
||||||
from .chromium_base import ChromiumBase, ChromiumPageScroll, ChromiumBaseSetter
|
from .chromium_base import ChromiumBase, ChromiumPageScroll
|
||||||
from .chromium_element import ChromiumElement
|
from .chromium_element import ChromiumElement
|
||||||
from .commons.tools import get_usable_path
|
from .commons.tools import get_usable_path
|
||||||
from .errors import ContextLossError
|
from .errors import ContextLossError
|
||||||
|
from .setter import ChromiumFrameSetter
|
||||||
from .waiter import FrameWaiter
|
from .waiter import FrameWaiter
|
||||||
|
|
||||||
|
|
||||||
@ -650,14 +651,3 @@ class ChromiumFrameScroll(ChromiumPageScroll):
|
|||||||
"""
|
"""
|
||||||
ele = loc_or_ele if isinstance(loc_or_ele, ChromiumElement) else self._driver._ele(loc_or_ele)
|
ele = loc_or_ele if isinstance(loc_or_ele, ChromiumElement) else self._driver._ele(loc_or_ele)
|
||||||
self._to_see(ele, center)
|
self._to_see(ele, center)
|
||||||
|
|
||||||
|
|
||||||
class ChromiumFrameSetter(ChromiumBaseSetter):
|
|
||||||
def attr(self, attr, value):
|
|
||||||
"""设置frame元素attribute属性
|
|
||||||
:param attr: 属性名
|
|
||||||
:param value: 属性值
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page._check_ok()
|
|
||||||
self._page.frame_ele.set.attr(attr, value)
|
|
||||||
|
@ -6,9 +6,10 @@
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Union, Tuple, List, Any
|
from typing import Union, Tuple, List, Any
|
||||||
|
|
||||||
from .waiter import FrameWaiter
|
from .chromium_base import ChromiumBase, ChromiumPageScroll
|
||||||
from .chromium_base import ChromiumBase, ChromiumPageScroll, ChromiumBaseSetter
|
|
||||||
from .chromium_element import ChromiumElement, Locations, ChromiumElementStates
|
from .chromium_element import ChromiumElement, Locations, ChromiumElementStates
|
||||||
|
from .setter import ChromiumFrameSetter
|
||||||
|
from .waiter import FrameWaiter
|
||||||
|
|
||||||
|
|
||||||
class ChromiumFrame(ChromiumBase):
|
class ChromiumFrame(ChromiumBase):
|
||||||
@ -205,9 +206,3 @@ class ChromiumFrameScroll(ChromiumPageScroll):
|
|||||||
def __init__(self, frame: ChromiumFrame) -> None: ...
|
def __init__(self, frame: ChromiumFrame) -> None: ...
|
||||||
|
|
||||||
def to_see(self, loc_or_ele: Union[str, tuple, ChromiumElement], center: Union[None, bool] = None) -> None: ...
|
def to_see(self, loc_or_ele: Union[str, tuple, ChromiumElement], center: Union[None, bool] = None) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
class ChromiumFrameSetter(ChromiumBaseSetter):
|
|
||||||
_page: ChromiumFrame = ...
|
|
||||||
|
|
||||||
def attr(self, attr: str, value: str) -> None: ...
|
|
||||||
|
@ -3,16 +3,16 @@
|
|||||||
@Author : g1879
|
@Author : g1879
|
||||||
@Contact : g1879@qq.com
|
@Contact : g1879@qq.com
|
||||||
"""
|
"""
|
||||||
from platform import system
|
|
||||||
from time import perf_counter, sleep
|
from time import perf_counter, sleep
|
||||||
|
|
||||||
from .chromium_base import ChromiumBase, Timeout, ChromiumBaseSetter
|
from .chromium_base import ChromiumBase, Timeout
|
||||||
from .chromium_driver import ChromiumDriver
|
from .chromium_driver import ChromiumDriver
|
||||||
from .chromium_tab import ChromiumTab
|
from .chromium_tab import ChromiumTab
|
||||||
from .commons.browser import connect_browser
|
from .commons.browser import connect_browser
|
||||||
from .commons.tools import port_is_using
|
from .commons.tools import port_is_using
|
||||||
from .configs.chromium_options import ChromiumOptions
|
from .configs.chromium_options import ChromiumOptions
|
||||||
from .errors import BrowserConnectError
|
from .errors import BrowserConnectError
|
||||||
|
from .setter import ChromiumPageSetter
|
||||||
from .waiter import ChromiumPageWaiter
|
from .waiter import ChromiumPageWaiter
|
||||||
|
|
||||||
|
|
||||||
@ -689,191 +689,6 @@ class Alert(object):
|
|||||||
self.response_text = None
|
self.response_text = None
|
||||||
|
|
||||||
|
|
||||||
class WindowSetter(object):
|
|
||||||
"""用于设置窗口大小的类"""
|
|
||||||
|
|
||||||
def __init__(self, page):
|
|
||||||
"""
|
|
||||||
:param page: 页面对象
|
|
||||||
"""
|
|
||||||
self._page = page
|
|
||||||
self._window_id = self._get_info()['windowId']
|
|
||||||
|
|
||||||
def maximized(self):
|
|
||||||
"""窗口最大化"""
|
|
||||||
s = self._get_info()['bounds']['windowState']
|
|
||||||
if s in ('fullscreen', 'minimized'):
|
|
||||||
self._perform({'windowState': 'normal'})
|
|
||||||
self._perform({'windowState': 'maximized'})
|
|
||||||
|
|
||||||
def minimized(self):
|
|
||||||
"""窗口最小化"""
|
|
||||||
s = self._get_info()['bounds']['windowState']
|
|
||||||
if s == 'fullscreen':
|
|
||||||
self._perform({'windowState': 'normal'})
|
|
||||||
self._perform({'windowState': 'minimized'})
|
|
||||||
|
|
||||||
def fullscreen(self):
|
|
||||||
"""设置窗口为全屏"""
|
|
||||||
s = self._get_info()['bounds']['windowState']
|
|
||||||
if s == 'minimized':
|
|
||||||
self._perform({'windowState': 'normal'})
|
|
||||||
self._perform({'windowState': 'fullscreen'})
|
|
||||||
|
|
||||||
def normal(self):
|
|
||||||
"""设置窗口为常规模式"""
|
|
||||||
s = self._get_info()['bounds']['windowState']
|
|
||||||
if s == 'fullscreen':
|
|
||||||
self._perform({'windowState': 'normal'})
|
|
||||||
self._perform({'windowState': 'normal'})
|
|
||||||
|
|
||||||
def size(self, width=None, height=None):
|
|
||||||
"""设置窗口大小
|
|
||||||
:param width: 窗口宽度
|
|
||||||
:param height: 窗口高度
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if width or height:
|
|
||||||
s = self._get_info()['bounds']['windowState']
|
|
||||||
if s != 'normal':
|
|
||||||
self._perform({'windowState': 'normal'})
|
|
||||||
info = self._get_info()['bounds']
|
|
||||||
width = width - 16 if width else info['width']
|
|
||||||
height = height + 7 if height else info['height']
|
|
||||||
self._perform({'width': width, 'height': height})
|
|
||||||
|
|
||||||
def location(self, x=None, y=None):
|
|
||||||
"""设置窗口在屏幕中的位置,相对左上角坐标
|
|
||||||
:param x: 距离顶部距离
|
|
||||||
:param y: 距离左边距离
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if x is not None or y is not None:
|
|
||||||
self.normal()
|
|
||||||
info = self._get_info()['bounds']
|
|
||||||
x = x if x is not None else info['left']
|
|
||||||
y = y if y is not None else info['top']
|
|
||||||
self._perform({'left': x - 8, 'top': y})
|
|
||||||
|
|
||||||
def hide(self):
|
|
||||||
"""隐藏浏览器窗口,只在Windows系统可用"""
|
|
||||||
show_or_hide_browser(self._page, hide=True)
|
|
||||||
|
|
||||||
def show(self):
|
|
||||||
"""显示浏览器窗口,只在Windows系统可用"""
|
|
||||||
show_or_hide_browser(self._page, hide=False)
|
|
||||||
|
|
||||||
def _get_info(self):
|
|
||||||
"""获取窗口位置及大小信息"""
|
|
||||||
return self._page.run_cdp('Browser.getWindowForTarget')
|
|
||||||
|
|
||||||
def _perform(self, bounds):
|
|
||||||
"""执行改变窗口大小操作
|
|
||||||
:param bounds: 控制数据
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.run_cdp('Browser.setWindowBounds', windowId=self._window_id, bounds=bounds)
|
|
||||||
|
|
||||||
|
|
||||||
class ChromiumPageSetter(ChromiumBaseSetter):
|
|
||||||
def main_tab(self, tab_id=None):
|
|
||||||
"""设置主tab
|
|
||||||
:param tab_id: 标签页id,不传入则设置当前tab
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page._main_tab = tab_id or self._page.tab_id
|
|
||||||
|
|
||||||
@property
|
|
||||||
def window(self):
|
|
||||||
"""返回用于设置浏览器窗口的对象"""
|
|
||||||
return WindowSetter(self._page)
|
|
||||||
|
|
||||||
def tab_to_front(self, tab_or_id=None):
|
|
||||||
"""激活标签页使其处于最前面
|
|
||||||
:param tab_or_id: 标签页对象或id,为None表示当前标签页
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if not tab_or_id:
|
|
||||||
tab_or_id = self._page.tab_id
|
|
||||||
elif isinstance(tab_or_id, ChromiumTab):
|
|
||||||
tab_or_id = tab_or_id.tab_id
|
|
||||||
self._page._control_session.get(f'http://{self._page.address}/json/activate/{tab_or_id}')
|
|
||||||
|
|
||||||
|
|
||||||
def show_or_hide_browser(page, hide=True):
|
|
||||||
"""执行显示或隐藏浏览器窗口
|
|
||||||
:param page: ChromePage对象
|
|
||||||
:param hide: 是否隐藏
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if not page.address.startswith(('127.0.0.1', 'localhost')):
|
|
||||||
return
|
|
||||||
|
|
||||||
if system().lower() != 'windows':
|
|
||||||
raise OSError('该方法只能在Windows系统使用。')
|
|
||||||
|
|
||||||
try:
|
|
||||||
from win32gui import ShowWindow
|
|
||||||
from win32con import SW_HIDE, SW_SHOW
|
|
||||||
except ImportError:
|
|
||||||
raise ImportError('请先安装:pip install pypiwin32')
|
|
||||||
|
|
||||||
pid = page.process_id
|
|
||||||
if not pid:
|
|
||||||
return None
|
|
||||||
hds = get_chrome_hwnds_from_pid(pid, page.title)
|
|
||||||
sw = SW_HIDE if hide else SW_SHOW
|
|
||||||
for hd in hds:
|
|
||||||
ShowWindow(hd, sw)
|
|
||||||
|
|
||||||
|
|
||||||
def get_browser_progress_id(progress, address):
|
|
||||||
"""获取浏览器进程id
|
|
||||||
:param progress: 已知的进程对象,没有时传入None
|
|
||||||
:param address: 浏览器管理地址,含端口
|
|
||||||
:return: 进程id或None
|
|
||||||
"""
|
|
||||||
if progress:
|
|
||||||
return progress.pid
|
|
||||||
|
|
||||||
from os import popen
|
|
||||||
port = address.split(':')[-1]
|
|
||||||
txt = ''
|
|
||||||
progresses = popen(f'netstat -nao | findstr :{port}').read().split('\n')
|
|
||||||
for progress in progresses:
|
|
||||||
if 'LISTENING' in progress:
|
|
||||||
txt = progress
|
|
||||||
break
|
|
||||||
if not txt:
|
|
||||||
return None
|
|
||||||
|
|
||||||
return txt.split(' ')[-1]
|
|
||||||
|
|
||||||
|
|
||||||
def get_chrome_hwnds_from_pid(pid, title):
|
|
||||||
"""通过PID查询句柄ID
|
|
||||||
:param pid: 进程id
|
|
||||||
:param title: 窗口标题
|
|
||||||
:return: 进程句柄组成的列表
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
from win32gui import IsWindow, GetWindowText, EnumWindows
|
|
||||||
from win32process import GetWindowThreadProcessId
|
|
||||||
except ImportError:
|
|
||||||
raise ImportError('请先安装win32gui,pip install pypiwin32')
|
|
||||||
|
|
||||||
def callback(hwnd, hds):
|
|
||||||
if IsWindow(hwnd) and title in GetWindowText(hwnd):
|
|
||||||
_, found_pid = GetWindowThreadProcessId(hwnd)
|
|
||||||
if str(found_pid) == str(pid):
|
|
||||||
hds.append(hwnd)
|
|
||||||
return True
|
|
||||||
|
|
||||||
hwnds = []
|
|
||||||
EnumWindows(callback, hwnds)
|
|
||||||
return hwnds
|
|
||||||
|
|
||||||
|
|
||||||
def get_rename(original, rename):
|
def get_rename(original, rename):
|
||||||
if '.' in rename:
|
if '.' in rename:
|
||||||
return rename
|
return rename
|
||||||
|
@ -3,10 +3,10 @@
|
|||||||
@Author : g1879
|
@Author : g1879
|
||||||
@Contact : g1879@qq.com
|
@Contact : g1879@qq.com
|
||||||
"""
|
"""
|
||||||
from os import popen
|
|
||||||
from typing import Union, Tuple, List
|
from typing import Union, Tuple, List
|
||||||
|
|
||||||
from .chromium_base import ChromiumBase, ChromiumBaseSetter
|
from .setter import ChromiumPageSetter
|
||||||
|
from .chromium_base import ChromiumBase
|
||||||
from .chromium_driver import ChromiumDriver
|
from .chromium_driver import ChromiumDriver
|
||||||
from .chromium_tab import ChromiumTab
|
from .chromium_tab import ChromiumTab
|
||||||
from .configs.chromium_options import ChromiumOptions
|
from .configs.chromium_options import ChromiumOptions
|
||||||
@ -21,7 +21,7 @@ class ChromiumPage(ChromiumBase):
|
|||||||
timeout: float = None):
|
timeout: float = None):
|
||||||
self._driver_options: ChromiumOptions = ...
|
self._driver_options: ChromiumOptions = ...
|
||||||
self._process_id: str = ...
|
self._process_id: str = ...
|
||||||
self._window_setter: WindowSetter = ...
|
# self._window_setter: WindowSetter = ...
|
||||||
self._main_tab: str = ...
|
self._main_tab: str = ...
|
||||||
self._alert: Alert = ...
|
self._alert: Alert = ...
|
||||||
self._browser_driver: ChromiumDriver = ...
|
self._browser_driver: ChromiumDriver = ...
|
||||||
@ -197,51 +197,4 @@ class Alert(object):
|
|||||||
self.response_text: str = ...
|
self.response_text: str = ...
|
||||||
|
|
||||||
|
|
||||||
class WindowSetter(object):
|
|
||||||
|
|
||||||
def __init__(self, page: ChromiumPage):
|
|
||||||
self._page: ChromiumPage = ...
|
|
||||||
self._window_id: str = ...
|
|
||||||
|
|
||||||
def maximized(self) -> None: ...
|
|
||||||
|
|
||||||
def minimized(self) -> None: ...
|
|
||||||
|
|
||||||
def fullscreen(self) -> None: ...
|
|
||||||
|
|
||||||
def normal(self) -> None: ...
|
|
||||||
|
|
||||||
def size(self, width: int = None, height: int = None) -> None: ...
|
|
||||||
|
|
||||||
def location(self, x: int = None, y: int = None) -> None: ...
|
|
||||||
|
|
||||||
def hide(self) -> None: ...
|
|
||||||
|
|
||||||
def show(self) -> None: ...
|
|
||||||
|
|
||||||
def _get_info(self) -> dict: ...
|
|
||||||
|
|
||||||
def _perform(self, bounds: dict) -> None: ...
|
|
||||||
|
|
||||||
|
|
||||||
def show_or_hide_browser(page: ChromiumPage, hide: bool = True) -> None: ...
|
|
||||||
|
|
||||||
|
|
||||||
def get_browser_progress_id(progress: Union[popen, None], address: str) -> Union[str, None]: ...
|
|
||||||
|
|
||||||
|
|
||||||
def get_chrome_hwnds_from_pid(pid: Union[str, int], title: str) -> list: ...
|
|
||||||
|
|
||||||
|
|
||||||
class ChromiumPageSetter(ChromiumBaseSetter):
|
|
||||||
_page: ChromiumPage = ...
|
|
||||||
|
|
||||||
def main_tab(self, tab_id: str = None) -> None: ...
|
|
||||||
|
|
||||||
@property
|
|
||||||
def window(self) -> WindowSetter: ...
|
|
||||||
|
|
||||||
def tab_to_front(self, tab_or_id: Union[str, ChromiumTab] = None) -> None: ...
|
|
||||||
|
|
||||||
|
|
||||||
def get_rename(original: str, rename: str) -> str: ...
|
def get_rename(original: str, rename: str) -> str: ...
|
||||||
|
@ -5,9 +5,10 @@
|
|||||||
"""
|
"""
|
||||||
from copy import copy
|
from copy import copy
|
||||||
|
|
||||||
from .chromium_base import ChromiumBase, ChromiumBaseSetter
|
from .chromium_base import ChromiumBase
|
||||||
from .commons.web import set_session_cookies, set_browser_cookies
|
from .commons.web import set_session_cookies, set_browser_cookies
|
||||||
from .session_page import SessionPage, SessionPageSetter
|
from .session_page import SessionPage
|
||||||
|
from .setter import WebPageTabSetter
|
||||||
|
|
||||||
|
|
||||||
class ChromiumTab(ChromiumBase):
|
class ChromiumTab(ChromiumBase):
|
||||||
@ -327,39 +328,3 @@ class WebPageTab(SessionPage, ChromiumTab):
|
|||||||
elif self._mode == 'd':
|
elif self._mode == 'd':
|
||||||
return super(SessionPage, self)._find_elements(loc_or_ele, timeout=timeout, single=single,
|
return super(SessionPage, self)._find_elements(loc_or_ele, timeout=timeout, single=single,
|
||||||
relative=relative)
|
relative=relative)
|
||||||
|
|
||||||
|
|
||||||
class WebPageTabSetter(ChromiumBaseSetter):
|
|
||||||
def __init__(self, page):
|
|
||||||
super().__init__(page)
|
|
||||||
self._session_setter = SessionPageSetter(self._page)
|
|
||||||
self._chromium_setter = ChromiumBaseSetter(self._page)
|
|
||||||
|
|
||||||
def cookies(self, cookies):
|
|
||||||
"""添加多个cookies信息到浏览器或session对象,注意不要传入单个
|
|
||||||
:param cookies: 可以接收`CookieJar`、`list`、`tuple`、`str`、`dict`格式的`cookies`
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if self._page.mode == 'd' and self._page._has_driver:
|
|
||||||
self._chromium_setter.cookies(cookies)
|
|
||||||
elif self._page.mode == 's' and self._page._has_session:
|
|
||||||
self._session_setter.cookies(cookies)
|
|
||||||
|
|
||||||
def headers(self, headers) -> None:
|
|
||||||
"""设置固定发送的headers
|
|
||||||
:param headers: dict格式的headers数据
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if self._page._has_session:
|
|
||||||
self._session_setter.headers(headers)
|
|
||||||
if self._page._has_driver:
|
|
||||||
self._chromium_setter.headers(headers)
|
|
||||||
|
|
||||||
def user_agent(self, ua, platform=None):
|
|
||||||
"""设置user agent,d模式下只有当前tab有效"""
|
|
||||||
if self._page._has_session:
|
|
||||||
self._session_setter.user_agent(ua)
|
|
||||||
if self._page._has_driver:
|
|
||||||
self._chromium_setter.user_agent(ua, platform)
|
|
||||||
|
|
||||||
|
|
||||||
|
@ -7,12 +7,13 @@ from typing import Union, Tuple, Any, List
|
|||||||
|
|
||||||
from requests import Session, Response
|
from requests import Session, Response
|
||||||
|
|
||||||
from .chromium_base import ChromiumBase, ChromiumBaseSetter
|
from .chromium_base import ChromiumBase
|
||||||
from .chromium_element import ChromiumElement
|
from .chromium_element import ChromiumElement
|
||||||
from .chromium_frame import ChromiumFrame
|
from .chromium_frame import ChromiumFrame
|
||||||
from .chromium_page import ChromiumPage, ChromiumTabRect
|
from .chromium_page import ChromiumPage, ChromiumTabRect
|
||||||
from .session_element import SessionElement
|
from .session_element import SessionElement
|
||||||
from .session_page import SessionPage, SessionPageSetter
|
from .session_page import SessionPage
|
||||||
|
from .setter import WebPageTabSetter
|
||||||
from .web_page import WebPage
|
from .web_page import WebPage
|
||||||
|
|
||||||
|
|
||||||
@ -149,15 +150,3 @@ class WebPageTab(SessionPage, ChromiumTab):
|
|||||||
timeout: float = None, single: bool = True, relative: bool = False, raise_err: bool = None) \
|
timeout: float = None, single: bool = True, relative: bool = False, raise_err: bool = None) \
|
||||||
-> Union[ChromiumElement, SessionElement, ChromiumFrame, str, None, List[Union[SessionElement, str]], List[
|
-> Union[ChromiumElement, SessionElement, ChromiumFrame, str, None, List[Union[SessionElement, str]], List[
|
||||||
Union[ChromiumElement, str, ChromiumFrame]]]: ...
|
Union[ChromiumElement, str, ChromiumFrame]]]: ...
|
||||||
|
|
||||||
|
|
||||||
class WebPageTabSetter(ChromiumBaseSetter):
|
|
||||||
_page: WebPage = ...
|
|
||||||
_session_setter: SessionPageSetter = ...
|
|
||||||
_chromium_setter: ChromiumBaseSetter = ...
|
|
||||||
|
|
||||||
def user_agent(self, ua: str, platform: str = None) -> None: ...
|
|
||||||
|
|
||||||
def headers(self, headers: dict) -> None: ...
|
|
||||||
|
|
||||||
def cookies(self, cookies) -> None: ...
|
|
||||||
|
@ -3,6 +3,7 @@
|
|||||||
@Author : g1879
|
@Author : g1879
|
||||||
@Contact : g1879@qq.com
|
@Contact : g1879@qq.com
|
||||||
"""
|
"""
|
||||||
|
from platform import system
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from re import search, sub
|
from re import search, sub
|
||||||
from shutil import rmtree
|
from shutil import rmtree
|
||||||
@ -101,6 +102,80 @@ def clean_folder(folder_path, ignore=None):
|
|||||||
elif f.is_dir():
|
elif f.is_dir():
|
||||||
rmtree(f, True)
|
rmtree(f, True)
|
||||||
|
|
||||||
|
|
||||||
|
def show_or_hide_browser(page, hide=True):
|
||||||
|
"""执行显示或隐藏浏览器窗口
|
||||||
|
:param page: ChromePage对象
|
||||||
|
:param hide: 是否隐藏
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if not page.address.startswith(('127.0.0.1', 'localhost')):
|
||||||
|
return
|
||||||
|
|
||||||
|
if system().lower() != 'windows':
|
||||||
|
raise OSError('该方法只能在Windows系统使用。')
|
||||||
|
|
||||||
|
try:
|
||||||
|
from win32gui import ShowWindow
|
||||||
|
from win32con import SW_HIDE, SW_SHOW
|
||||||
|
except ImportError:
|
||||||
|
raise ImportError('请先安装:pip install pypiwin32')
|
||||||
|
|
||||||
|
pid = page.process_id
|
||||||
|
if not pid:
|
||||||
|
return None
|
||||||
|
hds = get_chrome_hwnds_from_pid(pid, page.title)
|
||||||
|
sw = SW_HIDE if hide else SW_SHOW
|
||||||
|
for hd in hds:
|
||||||
|
ShowWindow(hd, sw)
|
||||||
|
|
||||||
|
|
||||||
|
def get_browser_progress_id(progress, address):
|
||||||
|
"""获取浏览器进程id
|
||||||
|
:param progress: 已知的进程对象,没有时传入None
|
||||||
|
:param address: 浏览器管理地址,含端口
|
||||||
|
:return: 进程id或None
|
||||||
|
"""
|
||||||
|
if progress:
|
||||||
|
return progress.pid
|
||||||
|
|
||||||
|
from os import popen
|
||||||
|
port = address.split(':')[-1]
|
||||||
|
txt = ''
|
||||||
|
progresses = popen(f'netstat -nao | findstr :{port}').read().split('\n')
|
||||||
|
for progress in progresses:
|
||||||
|
if 'LISTENING' in progress:
|
||||||
|
txt = progress
|
||||||
|
break
|
||||||
|
if not txt:
|
||||||
|
return None
|
||||||
|
|
||||||
|
return txt.split(' ')[-1]
|
||||||
|
|
||||||
|
|
||||||
|
def get_chrome_hwnds_from_pid(pid, title):
|
||||||
|
"""通过PID查询句柄ID
|
||||||
|
:param pid: 进程id
|
||||||
|
:param title: 窗口标题
|
||||||
|
:return: 进程句柄组成的列表
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
from win32gui import IsWindow, GetWindowText, EnumWindows
|
||||||
|
from win32process import GetWindowThreadProcessId
|
||||||
|
except ImportError:
|
||||||
|
raise ImportError('请先安装win32gui,pip install pypiwin32')
|
||||||
|
|
||||||
|
def callback(hwnd, hds):
|
||||||
|
if IsWindow(hwnd) and title in GetWindowText(hwnd):
|
||||||
|
_, found_pid = GetWindowThreadProcessId(hwnd)
|
||||||
|
if str(found_pid) == str(pid):
|
||||||
|
hds.append(hwnd)
|
||||||
|
return True
|
||||||
|
|
||||||
|
hwnds = []
|
||||||
|
EnumWindows(callback, hwnds)
|
||||||
|
return hwnds
|
||||||
|
|
||||||
# def get_exe_from_port(port):
|
# def get_exe_from_port(port):
|
||||||
# """获取端口号第一条进程的可执行文件路径
|
# """获取端口号第一条进程的可执行文件路径
|
||||||
# :param port: 端口号
|
# :param port: 端口号
|
||||||
|
@ -3,9 +3,12 @@
|
|||||||
@Author : g1879
|
@Author : g1879
|
||||||
@Contact : g1879@qq.com
|
@Contact : g1879@qq.com
|
||||||
"""
|
"""
|
||||||
|
from os import popen
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Union
|
from typing import Union
|
||||||
|
|
||||||
|
from chromium_page import ChromiumPage
|
||||||
|
|
||||||
|
|
||||||
# def get_exe_from_port(port: Union[str, int]) -> Union[str, None]: ...
|
# def get_exe_from_port(port: Union[str, int]) -> Union[str, None]: ...
|
||||||
|
|
||||||
@ -26,3 +29,12 @@ def port_is_using(ip: str, port: Union[str, int]) -> bool: ...
|
|||||||
|
|
||||||
|
|
||||||
def clean_folder(folder_path: Union[str, Path], ignore: Union[tuple, list] = None) -> None: ...
|
def clean_folder(folder_path: Union[str, Path], ignore: Union[tuple, list] = None) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
|
def show_or_hide_browser(page: ChromiumPage, hide: bool = True) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
|
def get_browser_progress_id(progress: Union[popen, None], address: str) -> Union[str, None]: ...
|
||||||
|
|
||||||
|
|
||||||
|
def get_chrome_hwnds_from_pid(pid: Union[str, int], title: str) -> list: ...
|
||||||
|
@ -12,9 +12,10 @@ from requests.structures import CaseInsensitiveDict
|
|||||||
from tldextract import extract
|
from tldextract import extract
|
||||||
|
|
||||||
from .base import BasePage
|
from .base import BasePage
|
||||||
from .commons.web import cookie_to_dict, set_session_cookies
|
from .commons.web import cookie_to_dict
|
||||||
from .configs.session_options import SessionOptions
|
from .configs.session_options import SessionOptions
|
||||||
from .session_element import SessionElement, make_session_ele
|
from .session_element import SessionElement, make_session_ele
|
||||||
|
from .setter import SessionPageSetter
|
||||||
|
|
||||||
|
|
||||||
class SessionPage(BasePage):
|
class SessionPage(BasePage):
|
||||||
@ -304,137 +305,6 @@ class SessionPage(BasePage):
|
|||||||
return r, f'状态码:{r.status_code}'
|
return r, f'状态码:{r.status_code}'
|
||||||
|
|
||||||
|
|
||||||
class SessionPageSetter(object):
|
|
||||||
def __init__(self, page):
|
|
||||||
self._page = page
|
|
||||||
|
|
||||||
def retry_times(self, times):
|
|
||||||
"""设置连接失败时重连次数"""
|
|
||||||
self._page.retry_times = times
|
|
||||||
|
|
||||||
def retry_interval(self, interval):
|
|
||||||
"""设置连接失败时重连间隔"""
|
|
||||||
self._page.retry_interval = interval
|
|
||||||
|
|
||||||
def timeout(self, second):
|
|
||||||
"""设置连接超时时间
|
|
||||||
:param second: 秒数
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.timeout = second
|
|
||||||
|
|
||||||
def cookie(self, cookie):
|
|
||||||
"""为Session对象设置单个cookie
|
|
||||||
:param cookie: cookie信息
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if isinstance(cookie, str):
|
|
||||||
self.cookies(cookie)
|
|
||||||
else:
|
|
||||||
self.cookies([cookie])
|
|
||||||
|
|
||||||
def cookies(self, cookies):
|
|
||||||
"""为Session对象设置多个cookie,注意不要传入单个
|
|
||||||
:param cookies: cookies信息
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
set_session_cookies(self._page.session, cookies)
|
|
||||||
|
|
||||||
def headers(self, headers):
|
|
||||||
"""设置通用的headers
|
|
||||||
:param headers: dict形式的headers
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.session.headers = CaseInsensitiveDict(headers)
|
|
||||||
|
|
||||||
def header(self, attr, value):
|
|
||||||
"""设置headers中一个项
|
|
||||||
:param attr: 设置名称
|
|
||||||
:param value: 设置值
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.session.headers[attr.lower()] = value
|
|
||||||
|
|
||||||
def user_agent(self, ua):
|
|
||||||
"""设置user agent
|
|
||||||
:param ua: user agent
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.session.headers['user-agent'] = ua
|
|
||||||
|
|
||||||
def proxies(self, http=None, https=None):
|
|
||||||
"""设置proxies参数
|
|
||||||
:param http: http代理地址
|
|
||||||
:param https: https代理地址
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.session.proxies = {'http': http, 'https': https}
|
|
||||||
|
|
||||||
def auth(self, auth):
|
|
||||||
"""设置认证元组或对象
|
|
||||||
:param auth: 认证元组或对象
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.session.auth = auth
|
|
||||||
|
|
||||||
def hooks(self, hooks):
|
|
||||||
"""设置回调方法
|
|
||||||
:param hooks: 回调方法
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.session.hooks = hooks
|
|
||||||
|
|
||||||
def params(self, params):
|
|
||||||
"""设置查询参数字典
|
|
||||||
:param params: 查询参数字典
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.session.params = params
|
|
||||||
|
|
||||||
def verify(self, on_off):
|
|
||||||
"""设置是否验证SSL证书
|
|
||||||
:param on_off: 是否验证 SSL 证书
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.session.verify = on_off
|
|
||||||
|
|
||||||
def cert(self, cert):
|
|
||||||
"""SSL客户端证书文件的路径(.pem格式),或(‘cert’, ‘key’)元组
|
|
||||||
:param cert: 证书路径或元组
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.session.cert = cert
|
|
||||||
|
|
||||||
def stream(self, on_off):
|
|
||||||
"""设置是否使用流式响应内容
|
|
||||||
:param on_off: 是否使用流式响应内容
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.session.stream = on_off
|
|
||||||
|
|
||||||
def trust_env(self, on_off):
|
|
||||||
"""设置是否信任环境
|
|
||||||
:param on_off: 是否信任环境
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.session.trust_env = on_off
|
|
||||||
|
|
||||||
def max_redirects(self, times):
|
|
||||||
"""设置最大重定向次数
|
|
||||||
:param times: 最大重定向次数
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.session.max_redirects = times
|
|
||||||
|
|
||||||
def add_adapter(self, url, adapter):
|
|
||||||
"""添加适配器
|
|
||||||
:param url: 适配器对应url
|
|
||||||
:param adapter: 适配器对象
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self._page.session.mount(url, adapter)
|
|
||||||
|
|
||||||
|
|
||||||
def check_headers(kwargs, headers, arg) -> bool:
|
def check_headers(kwargs, headers, arg) -> bool:
|
||||||
"""检查kwargs或headers中是否有arg所示属性"""
|
"""检查kwargs或headers中是否有arg所示属性"""
|
||||||
return arg in kwargs['headers'] or arg in headers
|
return arg in kwargs['headers'] or arg in headers
|
||||||
|
@ -3,20 +3,16 @@
|
|||||||
@Author : g1879
|
@Author : g1879
|
||||||
@Contact : g1879@qq.com
|
@Contact : g1879@qq.com
|
||||||
"""
|
"""
|
||||||
from http.cookiejar import Cookie
|
|
||||||
from typing import Any, Union, Tuple, List
|
from typing import Any, Union, Tuple, List
|
||||||
|
|
||||||
# from DownloadKit import DownloadKit
|
|
||||||
from requests import Session, Response
|
from requests import Session, Response
|
||||||
from requests.adapters import HTTPAdapter
|
|
||||||
from requests.auth import HTTPBasicAuth
|
|
||||||
from requests.cookies import RequestsCookieJar
|
|
||||||
from requests.structures import CaseInsensitiveDict
|
from requests.structures import CaseInsensitiveDict
|
||||||
|
|
||||||
from .base import BasePage
|
from .base import BasePage
|
||||||
from .commons.constants import NoneElement
|
from .commons.constants import NoneElement
|
||||||
from .configs.session_options import SessionOptions
|
from .configs.session_options import SessionOptions
|
||||||
from .session_element import SessionElement
|
from .session_element import SessionElement
|
||||||
|
from .setter import SessionPageSetter
|
||||||
|
|
||||||
|
|
||||||
class SessionPage(BasePage):
|
class SessionPage(BasePage):
|
||||||
@ -160,47 +156,6 @@ class SessionPage(BasePage):
|
|||||||
**kwargs) -> tuple: ...
|
**kwargs) -> tuple: ...
|
||||||
|
|
||||||
|
|
||||||
class SessionPageSetter(object):
|
|
||||||
def __init__(self, page: SessionPage):
|
|
||||||
self._page: SessionPage = ...
|
|
||||||
|
|
||||||
def retry_times(self, times: int) -> None: ...
|
|
||||||
|
|
||||||
def retry_interval(self, interval: float) -> None: ...
|
|
||||||
|
|
||||||
def timeout(self, second: float) -> None: ...
|
|
||||||
|
|
||||||
def cookie(self, cookie: Union[Cookie, str, dict]) -> None: ...
|
|
||||||
|
|
||||||
def cookies(self, cookies: Union[RequestsCookieJar, list, tuple, str, dict]) -> None: ...
|
|
||||||
|
|
||||||
def headers(self, headers: dict) -> None: ...
|
|
||||||
|
|
||||||
def header(self, attr: str, value: str) -> None: ...
|
|
||||||
|
|
||||||
def user_agent(self, ua: str) -> None: ...
|
|
||||||
|
|
||||||
def proxies(self, http: str = None, https: str = None) -> None: ...
|
|
||||||
|
|
||||||
def auth(self, auth: Union[Tuple[str, str], HTTPBasicAuth, None]) -> None: ...
|
|
||||||
|
|
||||||
def hooks(self, hooks: Union[dict, None]) -> None: ...
|
|
||||||
|
|
||||||
def params(self, params: Union[dict, None]) -> None: ...
|
|
||||||
|
|
||||||
def verify(self, on_off: Union[bool, None]) -> None: ...
|
|
||||||
|
|
||||||
def cert(self, cert: Union[str, Tuple[str, str], None]) -> None: ...
|
|
||||||
|
|
||||||
def stream(self, on_off: Union[bool, None]) -> None: ...
|
|
||||||
|
|
||||||
def trust_env(self, on_off: Union[bool, None]) -> None: ...
|
|
||||||
|
|
||||||
def max_redirects(self, times: Union[int, None]) -> None: ...
|
|
||||||
|
|
||||||
def add_adapter(self, url: str, adapter: HTTPAdapter) -> None: ...
|
|
||||||
|
|
||||||
|
|
||||||
def check_headers(kwargs: Union[dict, CaseInsensitiveDict], headers: Union[dict, CaseInsensitiveDict],
|
def check_headers(kwargs: Union[dict, CaseInsensitiveDict], headers: Union[dict, CaseInsensitiveDict],
|
||||||
arg: str) -> bool: ...
|
arg: str) -> bool: ...
|
||||||
|
|
||||||
|
526
DrissionPage/setter.py
Normal file
526
DrissionPage/setter.py
Normal file
@ -0,0 +1,526 @@
|
|||||||
|
# -*- coding:utf-8 -*-
|
||||||
|
"""
|
||||||
|
@Author : g1879
|
||||||
|
@Contact : g1879@qq.com
|
||||||
|
"""
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from requests.structures import CaseInsensitiveDict
|
||||||
|
|
||||||
|
from .commons.tools import show_or_hide_browser
|
||||||
|
from .commons.web import set_browser_cookies, set_session_cookies
|
||||||
|
|
||||||
|
|
||||||
|
class ChromiumBaseSetter(object):
|
||||||
|
def __init__(self, page):
|
||||||
|
self._page = page
|
||||||
|
|
||||||
|
@property
|
||||||
|
def load_strategy(self):
|
||||||
|
"""返回用于设置页面加载策略的对象"""
|
||||||
|
return PageLoadStrategy(self._page)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def scroll(self):
|
||||||
|
"""返回用于设置页面滚动设置的对象"""
|
||||||
|
return PageScrollSetter(self._page.scroll)
|
||||||
|
|
||||||
|
def retry_times(self, times):
|
||||||
|
"""设置连接失败重连次数"""
|
||||||
|
self._page.retry_times = times
|
||||||
|
|
||||||
|
def retry_interval(self, interval):
|
||||||
|
"""设置连接失败重连间隔"""
|
||||||
|
self._page.retry_interval = interval
|
||||||
|
|
||||||
|
def timeouts(self, implicit=None, page_load=None, script=None):
|
||||||
|
"""设置超时时间,单位为秒
|
||||||
|
:param implicit: 查找元素超时时间
|
||||||
|
:param page_load: 页面加载超时时间
|
||||||
|
:param script: 脚本运行超时时间
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if implicit is not None:
|
||||||
|
self._page.timeouts.implicit = implicit
|
||||||
|
|
||||||
|
if page_load is not None:
|
||||||
|
self._page.timeouts.page_load = page_load
|
||||||
|
|
||||||
|
if script is not None:
|
||||||
|
self._page.timeouts.script = script
|
||||||
|
|
||||||
|
def user_agent(self, ua, platform=None):
|
||||||
|
"""为当前tab设置user agent,只在当前tab有效
|
||||||
|
:param ua: user agent字符串
|
||||||
|
:param platform: platform字符串
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
keys = {'userAgent': ua}
|
||||||
|
if platform:
|
||||||
|
keys['platform'] = platform
|
||||||
|
self._page.run_cdp('Emulation.setUserAgentOverride', **keys)
|
||||||
|
|
||||||
|
def session_storage(self, item, value):
|
||||||
|
"""设置或删除某项sessionStorage信息
|
||||||
|
:param item: 要设置的项
|
||||||
|
:param value: 项的值,设置为False时,删除该项
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
js = f'sessionStorage.removeItem("{item}");' if item is False else f'sessionStorage.setItem("{item}","{value}");'
|
||||||
|
return self._page.run_js_loaded(js, as_expr=True)
|
||||||
|
|
||||||
|
def local_storage(self, item, value):
|
||||||
|
"""设置或删除某项localStorage信息
|
||||||
|
:param item: 要设置的项
|
||||||
|
:param value: 项的值,设置为False时,删除该项
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
js = f'localStorage.removeItem("{item}");' if item is False else f'localStorage.setItem("{item}","{value}");'
|
||||||
|
return self._page.run_js_loaded(js, as_expr=True)
|
||||||
|
|
||||||
|
def cookie(self, cookie):
|
||||||
|
"""设置单个cookie
|
||||||
|
:param cookie: cookie信息
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if isinstance(cookie, str):
|
||||||
|
self.cookies(cookie)
|
||||||
|
else:
|
||||||
|
self.cookies([cookie])
|
||||||
|
|
||||||
|
def cookies(self, cookies):
|
||||||
|
"""设置多个cookie,注意不要传入单个
|
||||||
|
:param cookies: cookies信息
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
set_browser_cookies(self._page, cookies)
|
||||||
|
|
||||||
|
def upload_files(self, files):
|
||||||
|
"""等待上传的文件路径
|
||||||
|
:param files: 文件路径列表或字符串,字符串时多个文件用回车分隔
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if not self._page._upload_list:
|
||||||
|
self._page.driver.Page.fileChooserOpened = self._page._onFileChooserOpened
|
||||||
|
self._page.run_cdp('Page.setInterceptFileChooserDialog', enabled=True)
|
||||||
|
|
||||||
|
if isinstance(files, str):
|
||||||
|
files = files.split('\n')
|
||||||
|
self._page._upload_list = [str(Path(i).absolute()) for i in files]
|
||||||
|
|
||||||
|
def headers(self, headers: dict) -> None:
|
||||||
|
"""设置固定发送的headers
|
||||||
|
:param headers: dict格式的headers数据
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.run_cdp('Network.enable')
|
||||||
|
self._page.run_cdp('Network.setExtraHTTPHeaders', headers=headers)
|
||||||
|
|
||||||
|
|
||||||
|
class ChromiumPageSetter(ChromiumBaseSetter):
|
||||||
|
def main_tab(self, tab_id=None):
|
||||||
|
"""设置主tab
|
||||||
|
:param tab_id: 标签页id,不传入则设置当前tab
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page._main_tab = tab_id or self._page.tab_id
|
||||||
|
|
||||||
|
@property
|
||||||
|
def window(self):
|
||||||
|
"""返回用于设置浏览器窗口的对象"""
|
||||||
|
return WindowSetter(self._page)
|
||||||
|
|
||||||
|
def tab_to_front(self, tab_or_id=None):
|
||||||
|
"""激活标签页使其处于最前面
|
||||||
|
:param tab_or_id: 标签页对象或id,为None表示当前标签页
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if not tab_or_id:
|
||||||
|
tab_or_id = self._page.tab_id
|
||||||
|
elif 'ChromiumTab' in str(type(tab_or_id)):
|
||||||
|
tab_or_id = tab_or_id.tab_id
|
||||||
|
self._page._control_session.get(f'http://{self._page.address}/json/activate/{tab_or_id}')
|
||||||
|
|
||||||
|
|
||||||
|
class SessionPageSetter(object):
|
||||||
|
def __init__(self, page):
|
||||||
|
self._page = page
|
||||||
|
|
||||||
|
def retry_times(self, times):
|
||||||
|
"""设置连接失败时重连次数"""
|
||||||
|
self._page.retry_times = times
|
||||||
|
|
||||||
|
def retry_interval(self, interval):
|
||||||
|
"""设置连接失败时重连间隔"""
|
||||||
|
self._page.retry_interval = interval
|
||||||
|
|
||||||
|
def timeout(self, second):
|
||||||
|
"""设置连接超时时间
|
||||||
|
:param second: 秒数
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.timeout = second
|
||||||
|
|
||||||
|
def cookie(self, cookie):
|
||||||
|
"""为Session对象设置单个cookie
|
||||||
|
:param cookie: cookie信息
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if isinstance(cookie, str):
|
||||||
|
self.cookies(cookie)
|
||||||
|
else:
|
||||||
|
self.cookies([cookie])
|
||||||
|
|
||||||
|
def cookies(self, cookies):
|
||||||
|
"""为Session对象设置多个cookie,注意不要传入单个
|
||||||
|
:param cookies: cookies信息
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
set_session_cookies(self._page.session, cookies)
|
||||||
|
|
||||||
|
def headers(self, headers):
|
||||||
|
"""设置通用的headers
|
||||||
|
:param headers: dict形式的headers
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.session.headers = CaseInsensitiveDict(headers)
|
||||||
|
|
||||||
|
def header(self, attr, value):
|
||||||
|
"""设置headers中一个项
|
||||||
|
:param attr: 设置名称
|
||||||
|
:param value: 设置值
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.session.headers[attr.lower()] = value
|
||||||
|
|
||||||
|
def user_agent(self, ua):
|
||||||
|
"""设置user agent
|
||||||
|
:param ua: user agent
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.session.headers['user-agent'] = ua
|
||||||
|
|
||||||
|
def proxies(self, http=None, https=None):
|
||||||
|
"""设置proxies参数
|
||||||
|
:param http: http代理地址
|
||||||
|
:param https: https代理地址
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.session.proxies = {'http': http, 'https': https}
|
||||||
|
|
||||||
|
def auth(self, auth):
|
||||||
|
"""设置认证元组或对象
|
||||||
|
:param auth: 认证元组或对象
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.session.auth = auth
|
||||||
|
|
||||||
|
def hooks(self, hooks):
|
||||||
|
"""设置回调方法
|
||||||
|
:param hooks: 回调方法
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.session.hooks = hooks
|
||||||
|
|
||||||
|
def params(self, params):
|
||||||
|
"""设置查询参数字典
|
||||||
|
:param params: 查询参数字典
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.session.params = params
|
||||||
|
|
||||||
|
def verify(self, on_off):
|
||||||
|
"""设置是否验证SSL证书
|
||||||
|
:param on_off: 是否验证 SSL 证书
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.session.verify = on_off
|
||||||
|
|
||||||
|
def cert(self, cert):
|
||||||
|
"""SSL客户端证书文件的路径(.pem格式),或(‘cert’, ‘key’)元组
|
||||||
|
:param cert: 证书路径或元组
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.session.cert = cert
|
||||||
|
|
||||||
|
def stream(self, on_off):
|
||||||
|
"""设置是否使用流式响应内容
|
||||||
|
:param on_off: 是否使用流式响应内容
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.session.stream = on_off
|
||||||
|
|
||||||
|
def trust_env(self, on_off):
|
||||||
|
"""设置是否信任环境
|
||||||
|
:param on_off: 是否信任环境
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.session.trust_env = on_off
|
||||||
|
|
||||||
|
def max_redirects(self, times):
|
||||||
|
"""设置最大重定向次数
|
||||||
|
:param times: 最大重定向次数
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.session.max_redirects = times
|
||||||
|
|
||||||
|
def add_adapter(self, url, adapter):
|
||||||
|
"""添加适配器
|
||||||
|
:param url: 适配器对应url
|
||||||
|
:param adapter: 适配器对象
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.session.mount(url, adapter)
|
||||||
|
|
||||||
|
|
||||||
|
class WebPageSetter(ChromiumPageSetter):
|
||||||
|
def __init__(self, page):
|
||||||
|
super().__init__(page)
|
||||||
|
self._session_setter = SessionPageSetter(self._page)
|
||||||
|
self._chromium_setter = ChromiumPageSetter(self._page)
|
||||||
|
|
||||||
|
def cookies(self, cookies):
|
||||||
|
"""添加cookies信息到浏览器或session对象
|
||||||
|
:param cookies: 可以接收`CookieJar`、`list`、`tuple`、`str`、`dict`格式的`cookies`
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if self._page.mode == 'd' and self._page._has_driver:
|
||||||
|
self._chromium_setter.cookies(cookies)
|
||||||
|
elif self._page.mode == 's' and self._page._has_session:
|
||||||
|
self._session_setter.cookies(cookies)
|
||||||
|
|
||||||
|
def headers(self, headers) -> None:
|
||||||
|
"""设置固定发送的headers
|
||||||
|
:param headers: dict格式的headers数据
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if self._page.mode == 's':
|
||||||
|
self._session_setter.headers(headers)
|
||||||
|
else:
|
||||||
|
self._chromium_setter.headers(headers)
|
||||||
|
|
||||||
|
def user_agent(self, ua, platform=None):
|
||||||
|
"""设置user agent,d模式下只有当前tab有效"""
|
||||||
|
if self._page.mode == 's':
|
||||||
|
self._session_setter.user_agent(ua)
|
||||||
|
else:
|
||||||
|
self._chromium_setter.user_agent(ua, platform)
|
||||||
|
|
||||||
|
|
||||||
|
class WebPageTabSetter(ChromiumBaseSetter):
|
||||||
|
def __init__(self, page):
|
||||||
|
super().__init__(page)
|
||||||
|
self._session_setter = SessionPageSetter(self._page)
|
||||||
|
self._chromium_setter = ChromiumBaseSetter(self._page)
|
||||||
|
|
||||||
|
def cookies(self, cookies):
|
||||||
|
"""添加多个cookies信息到浏览器或session对象,注意不要传入单个
|
||||||
|
:param cookies: 可以接收`CookieJar`、`list`、`tuple`、`str`、`dict`格式的`cookies`
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if self._page.mode == 'd' and self._page._has_driver:
|
||||||
|
self._chromium_setter.cookies(cookies)
|
||||||
|
elif self._page.mode == 's' and self._page._has_session:
|
||||||
|
self._session_setter.cookies(cookies)
|
||||||
|
|
||||||
|
def headers(self, headers) -> None:
|
||||||
|
"""设置固定发送的headers
|
||||||
|
:param headers: dict格式的headers数据
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if self._page._has_session:
|
||||||
|
self._session_setter.headers(headers)
|
||||||
|
if self._page._has_driver:
|
||||||
|
self._chromium_setter.headers(headers)
|
||||||
|
|
||||||
|
def user_agent(self, ua, platform=None):
|
||||||
|
"""设置user agent,d模式下只有当前tab有效"""
|
||||||
|
if self._page._has_session:
|
||||||
|
self._session_setter.user_agent(ua)
|
||||||
|
if self._page._has_driver:
|
||||||
|
self._chromium_setter.user_agent(ua, platform)
|
||||||
|
|
||||||
|
|
||||||
|
class ChromiumElementSetter(object):
|
||||||
|
def __init__(self, ele):
|
||||||
|
"""
|
||||||
|
:param ele: ChromiumElement
|
||||||
|
"""
|
||||||
|
self._ele = ele
|
||||||
|
|
||||||
|
def attr(self, attr, value):
|
||||||
|
"""设置元素attribute属性
|
||||||
|
:param attr: 属性名
|
||||||
|
:param value: 属性值
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._ele.page.run_cdp('DOM.setAttributeValue', nodeId=self._ele.ids.node_id, name=attr, value=str(value))
|
||||||
|
|
||||||
|
def prop(self, prop, value):
|
||||||
|
"""设置元素property属性
|
||||||
|
:param prop: 属性名
|
||||||
|
:param value: 属性值
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
value = value.replace('"', r'\"')
|
||||||
|
self._ele.run_js(f'this.{prop}="{value}";')
|
||||||
|
|
||||||
|
def innerHTML(self, html):
|
||||||
|
"""设置元素innerHTML
|
||||||
|
:param html: html文本
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self.prop('innerHTML', html)
|
||||||
|
|
||||||
|
|
||||||
|
class ChromiumFrameSetter(ChromiumBaseSetter):
|
||||||
|
def attr(self, attr, value):
|
||||||
|
"""设置frame元素attribute属性
|
||||||
|
:param attr: 属性名
|
||||||
|
:param value: 属性值
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page._check_ok()
|
||||||
|
self._page.frame_ele.set.attr(attr, value)
|
||||||
|
|
||||||
|
|
||||||
|
class PageLoadStrategy(object):
|
||||||
|
"""用于设置页面加载策略的类"""
|
||||||
|
|
||||||
|
def __init__(self, page):
|
||||||
|
"""
|
||||||
|
:param page: ChromiumBase对象
|
||||||
|
"""
|
||||||
|
self._page = page
|
||||||
|
|
||||||
|
def __call__(self, value):
|
||||||
|
"""设置加载策略
|
||||||
|
:param value: 可选 'normal', 'eager', 'none'
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if value.lower() not in ('normal', 'eager', 'none'):
|
||||||
|
raise ValueError("只能选择 'normal', 'eager', 'none'。")
|
||||||
|
self._page._page_load_strategy = value
|
||||||
|
|
||||||
|
def normal(self):
|
||||||
|
"""设置页面加载策略为normal"""
|
||||||
|
self._page._page_load_strategy = 'normal'
|
||||||
|
|
||||||
|
def eager(self):
|
||||||
|
"""设置页面加载策略为eager"""
|
||||||
|
self._page._page_load_strategy = 'eager'
|
||||||
|
|
||||||
|
def none(self):
|
||||||
|
"""设置页面加载策略为none"""
|
||||||
|
self._page._page_load_strategy = 'none'
|
||||||
|
|
||||||
|
|
||||||
|
class PageScrollSetter(object):
|
||||||
|
def __init__(self, scroll):
|
||||||
|
self._scroll = scroll
|
||||||
|
|
||||||
|
def wait_complete(self, on_off=True):
|
||||||
|
"""设置滚动命令后是否等待完成
|
||||||
|
:param on_off: 开或关
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if not isinstance(on_off, bool):
|
||||||
|
raise TypeError('on_off必须为bool。')
|
||||||
|
self._scroll._wait_complete = on_off
|
||||||
|
|
||||||
|
def smooth(self, on_off=True):
|
||||||
|
"""设置页面滚动是否平滑滚动
|
||||||
|
:param on_off: 开或关
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if not isinstance(on_off, bool):
|
||||||
|
raise TypeError('on_off必须为bool。')
|
||||||
|
b = 'smooth' if on_off else 'auto'
|
||||||
|
self._scroll._driver.run_js(f'document.documentElement.style.setProperty("scroll-behavior","{b}");')
|
||||||
|
self._scroll._wait_complete = on_off
|
||||||
|
|
||||||
|
|
||||||
|
class WindowSetter(object):
|
||||||
|
"""用于设置窗口大小的类"""
|
||||||
|
|
||||||
|
def __init__(self, page):
|
||||||
|
"""
|
||||||
|
:param page: 页面对象
|
||||||
|
"""
|
||||||
|
self._page = page
|
||||||
|
self._window_id = self._get_info()['windowId']
|
||||||
|
|
||||||
|
def maximized(self):
|
||||||
|
"""窗口最大化"""
|
||||||
|
s = self._get_info()['bounds']['windowState']
|
||||||
|
if s in ('fullscreen', 'minimized'):
|
||||||
|
self._perform({'windowState': 'normal'})
|
||||||
|
self._perform({'windowState': 'maximized'})
|
||||||
|
|
||||||
|
def minimized(self):
|
||||||
|
"""窗口最小化"""
|
||||||
|
s = self._get_info()['bounds']['windowState']
|
||||||
|
if s == 'fullscreen':
|
||||||
|
self._perform({'windowState': 'normal'})
|
||||||
|
self._perform({'windowState': 'minimized'})
|
||||||
|
|
||||||
|
def fullscreen(self):
|
||||||
|
"""设置窗口为全屏"""
|
||||||
|
s = self._get_info()['bounds']['windowState']
|
||||||
|
if s == 'minimized':
|
||||||
|
self._perform({'windowState': 'normal'})
|
||||||
|
self._perform({'windowState': 'fullscreen'})
|
||||||
|
|
||||||
|
def normal(self):
|
||||||
|
"""设置窗口为常规模式"""
|
||||||
|
s = self._get_info()['bounds']['windowState']
|
||||||
|
if s == 'fullscreen':
|
||||||
|
self._perform({'windowState': 'normal'})
|
||||||
|
self._perform({'windowState': 'normal'})
|
||||||
|
|
||||||
|
def size(self, width=None, height=None):
|
||||||
|
"""设置窗口大小
|
||||||
|
:param width: 窗口宽度
|
||||||
|
:param height: 窗口高度
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if width or height:
|
||||||
|
s = self._get_info()['bounds']['windowState']
|
||||||
|
if s != 'normal':
|
||||||
|
self._perform({'windowState': 'normal'})
|
||||||
|
info = self._get_info()['bounds']
|
||||||
|
width = width - 16 if width else info['width']
|
||||||
|
height = height + 7 if height else info['height']
|
||||||
|
self._perform({'width': width, 'height': height})
|
||||||
|
|
||||||
|
def location(self, x=None, y=None):
|
||||||
|
"""设置窗口在屏幕中的位置,相对左上角坐标
|
||||||
|
:param x: 距离顶部距离
|
||||||
|
:param y: 距离左边距离
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if x is not None or y is not None:
|
||||||
|
self.normal()
|
||||||
|
info = self._get_info()['bounds']
|
||||||
|
x = x if x is not None else info['left']
|
||||||
|
y = y if y is not None else info['top']
|
||||||
|
self._perform({'left': x - 8, 'top': y})
|
||||||
|
|
||||||
|
def hide(self):
|
||||||
|
"""隐藏浏览器窗口,只在Windows系统可用"""
|
||||||
|
show_or_hide_browser(self._page, hide=True)
|
||||||
|
|
||||||
|
def show(self):
|
||||||
|
"""显示浏览器窗口,只在Windows系统可用"""
|
||||||
|
show_or_hide_browser(self._page, hide=False)
|
||||||
|
|
||||||
|
def _get_info(self):
|
||||||
|
"""获取窗口位置及大小信息"""
|
||||||
|
return self._page.run_cdp('Browser.getWindowForTarget')
|
||||||
|
|
||||||
|
def _perform(self, bounds):
|
||||||
|
"""执行改变窗口大小操作
|
||||||
|
:param bounds: 控制数据
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
self._page.run_cdp('Browser.setWindowBounds', windowId=self._window_id, bounds=bounds)
|
192
DrissionPage/setter.pyi
Normal file
192
DrissionPage/setter.pyi
Normal file
@ -0,0 +1,192 @@
|
|||||||
|
# -*- coding:utf-8 -*-
|
||||||
|
"""
|
||||||
|
@Author : g1879
|
||||||
|
@Contact : g1879@qq.com
|
||||||
|
"""
|
||||||
|
from http.cookiejar import Cookie
|
||||||
|
from typing import Union, Tuple
|
||||||
|
|
||||||
|
from requests.adapters import HTTPAdapter
|
||||||
|
from requests.auth import HTTPBasicAuth
|
||||||
|
from requests.cookies import RequestsCookieJar
|
||||||
|
|
||||||
|
from .chromium_base import ChromiumBase, ChromiumPageScroll
|
||||||
|
from .chromium_element import ChromiumElement
|
||||||
|
from .chromium_frame import ChromiumFrame
|
||||||
|
from .chromium_page import ChromiumPage
|
||||||
|
from .chromium_tab import ChromiumTab
|
||||||
|
from .session_page import SessionPage
|
||||||
|
from .web_page import WebPage
|
||||||
|
|
||||||
|
|
||||||
|
class ChromiumBaseSetter(object):
|
||||||
|
def __init__(self, page):
|
||||||
|
self._page: ChromiumBase = ...
|
||||||
|
|
||||||
|
@property
|
||||||
|
def load_strategy(self) -> PageLoadStrategy: ...
|
||||||
|
|
||||||
|
@property
|
||||||
|
def scroll(self) -> PageScrollSetter: ...
|
||||||
|
|
||||||
|
def retry_times(self, times: int) -> None: ...
|
||||||
|
|
||||||
|
def retry_interval(self, interval: float) -> None: ...
|
||||||
|
|
||||||
|
def timeouts(self, implicit: float = None, page_load: float = None, script: float = None) -> None: ...
|
||||||
|
|
||||||
|
def user_agent(self, ua: str, platform: str = None) -> None: ...
|
||||||
|
|
||||||
|
def session_storage(self, item: str, value: Union[str, bool]) -> None: ...
|
||||||
|
|
||||||
|
def local_storage(self, item: str, value: Union[str, bool]) -> None: ...
|
||||||
|
|
||||||
|
def cookie(self, cookies: Union[RequestsCookieJar, str, dict]) -> None: ...
|
||||||
|
|
||||||
|
def cookies(self, cookies: Union[RequestsCookieJar, list, tuple, str, dict]) -> None: ...
|
||||||
|
|
||||||
|
def headers(self, headers: dict) -> None: ...
|
||||||
|
|
||||||
|
def upload_files(self, files: Union[str, list, tuple]) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
|
class ChromiumPageSetter(ChromiumBaseSetter):
|
||||||
|
_page: ChromiumPage = ...
|
||||||
|
|
||||||
|
def main_tab(self, tab_id: str = None) -> None: ...
|
||||||
|
|
||||||
|
@property
|
||||||
|
def window(self) -> WindowSetter: ...
|
||||||
|
|
||||||
|
def tab_to_front(self, tab_or_id: Union[str, ChromiumTab] = None) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
|
class SessionPageSetter(object):
|
||||||
|
def __init__(self, page: SessionPage):
|
||||||
|
self._page: SessionPage = ...
|
||||||
|
|
||||||
|
def retry_times(self, times: int) -> None: ...
|
||||||
|
|
||||||
|
def retry_interval(self, interval: float) -> None: ...
|
||||||
|
|
||||||
|
def timeout(self, second: float) -> None: ...
|
||||||
|
|
||||||
|
def cookie(self, cookie: Union[Cookie, str, dict]) -> None: ...
|
||||||
|
|
||||||
|
def cookies(self, cookies: Union[RequestsCookieJar, list, tuple, str, dict]) -> None: ...
|
||||||
|
|
||||||
|
def headers(self, headers: dict) -> None: ...
|
||||||
|
|
||||||
|
def header(self, attr: str, value: str) -> None: ...
|
||||||
|
|
||||||
|
def user_agent(self, ua: str) -> None: ...
|
||||||
|
|
||||||
|
def proxies(self, http: str = None, https: str = None) -> None: ...
|
||||||
|
|
||||||
|
def auth(self, auth: Union[Tuple[str, str], HTTPBasicAuth, None]) -> None: ...
|
||||||
|
|
||||||
|
def hooks(self, hooks: Union[dict, None]) -> None: ...
|
||||||
|
|
||||||
|
def params(self, params: Union[dict, None]) -> None: ...
|
||||||
|
|
||||||
|
def verify(self, on_off: Union[bool, None]) -> None: ...
|
||||||
|
|
||||||
|
def cert(self, cert: Union[str, Tuple[str, str], None]) -> None: ...
|
||||||
|
|
||||||
|
def stream(self, on_off: Union[bool, None]) -> None: ...
|
||||||
|
|
||||||
|
def trust_env(self, on_off: Union[bool, None]) -> None: ...
|
||||||
|
|
||||||
|
def max_redirects(self, times: Union[int, None]) -> None: ...
|
||||||
|
|
||||||
|
def add_adapter(self, url: str, adapter: HTTPAdapter) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
|
class WebPageSetter(ChromiumPageSetter):
|
||||||
|
_page: WebPage = ...
|
||||||
|
_session_setter: SessionPageSetter = ...
|
||||||
|
_chromium_setter: ChromiumPageSetter = ...
|
||||||
|
|
||||||
|
def user_agent(self, ua: str, platform: str = None) -> None: ...
|
||||||
|
|
||||||
|
def headers(self, headers: dict) -> None: ...
|
||||||
|
|
||||||
|
def cookies(self, cookies) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
|
class WebPageTabSetter(ChromiumBaseSetter):
|
||||||
|
_page: WebPage = ...
|
||||||
|
_session_setter: SessionPageSetter = ...
|
||||||
|
_chromium_setter: ChromiumBaseSetter = ...
|
||||||
|
|
||||||
|
def user_agent(self, ua: str, platform: str = None) -> None: ...
|
||||||
|
|
||||||
|
def headers(self, headers: dict) -> None: ...
|
||||||
|
|
||||||
|
def cookies(self, cookies) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
|
class ChromiumElementSetter(object):
|
||||||
|
def __init__(self, ele: ChromiumElement):
|
||||||
|
self._ele: ChromiumElement = ...
|
||||||
|
|
||||||
|
def attr(self, attr: str, value: str) -> None: ...
|
||||||
|
|
||||||
|
def prop(self, prop: str, value: str) -> None: ...
|
||||||
|
|
||||||
|
def innerHTML(self, html: str) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
|
class ChromiumFrameSetter(ChromiumBaseSetter):
|
||||||
|
_page: ChromiumFrame = ...
|
||||||
|
|
||||||
|
def attr(self, attr: str, value: str) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
|
class PageLoadStrategy(object):
|
||||||
|
def __init__(self, page: ChromiumBase):
|
||||||
|
self._page: ChromiumBase = ...
|
||||||
|
|
||||||
|
def __call__(self, value: str) -> None: ...
|
||||||
|
|
||||||
|
def normal(self) -> None: ...
|
||||||
|
|
||||||
|
def eager(self) -> None: ...
|
||||||
|
|
||||||
|
def none(self) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
|
class PageScrollSetter(object):
|
||||||
|
def __init__(self, scroll: ChromiumPageScroll):
|
||||||
|
self._scroll: ChromiumPageScroll = ...
|
||||||
|
|
||||||
|
def wait_complete(self, on_off: bool = True): ...
|
||||||
|
|
||||||
|
def smooth(self, on_off: bool = True): ...
|
||||||
|
|
||||||
|
|
||||||
|
class WindowSetter(object):
|
||||||
|
|
||||||
|
def __init__(self, page: ChromiumPage):
|
||||||
|
self._page: ChromiumPage = ...
|
||||||
|
self._window_id: str = ...
|
||||||
|
|
||||||
|
def maximized(self) -> None: ...
|
||||||
|
|
||||||
|
def minimized(self) -> None: ...
|
||||||
|
|
||||||
|
def fullscreen(self) -> None: ...
|
||||||
|
|
||||||
|
def normal(self) -> None: ...
|
||||||
|
|
||||||
|
def size(self, width: int = None, height: int = None) -> None: ...
|
||||||
|
|
||||||
|
def location(self, x: int = None, y: int = None) -> None: ...
|
||||||
|
|
||||||
|
def hide(self) -> None: ...
|
||||||
|
|
||||||
|
def show(self) -> None: ...
|
||||||
|
|
||||||
|
def _get_info(self) -> dict: ...
|
||||||
|
|
||||||
|
def _perform(self, bounds: dict) -> None: ...
|
@ -8,12 +8,13 @@ from requests import Session
|
|||||||
from .base import BasePage
|
from .base import BasePage
|
||||||
from .chromium_base import ChromiumBase, Timeout
|
from .chromium_base import ChromiumBase, Timeout
|
||||||
from .chromium_driver import ChromiumDriver
|
from .chromium_driver import ChromiumDriver
|
||||||
from .chromium_page import ChromiumPage, ChromiumPageSetter
|
from .chromium_page import ChromiumPage
|
||||||
from .chromium_tab import WebPageTab
|
from .chromium_tab import WebPageTab
|
||||||
from .commons.web import set_session_cookies, set_browser_cookies
|
from .commons.web import set_session_cookies, set_browser_cookies
|
||||||
from .configs.chromium_options import ChromiumOptions
|
from .configs.chromium_options import ChromiumOptions
|
||||||
from .configs.session_options import SessionOptions
|
from .configs.session_options import SessionOptions
|
||||||
from .session_page import SessionPage, SessionPageSetter
|
from .session_page import SessionPage
|
||||||
|
from .setter import WebPageSetter
|
||||||
|
|
||||||
|
|
||||||
class WebPage(SessionPage, ChromiumPage, BasePage):
|
class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||||
@ -426,35 +427,3 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
|||||||
self._has_driver = None
|
self._has_driver = None
|
||||||
|
|
||||||
|
|
||||||
class WebPageSetter(ChromiumPageSetter):
|
|
||||||
def __init__(self, page):
|
|
||||||
super().__init__(page)
|
|
||||||
self._session_setter = SessionPageSetter(self._page)
|
|
||||||
self._chromium_setter = ChromiumPageSetter(self._page)
|
|
||||||
|
|
||||||
def cookies(self, cookies):
|
|
||||||
"""添加cookies信息到浏览器或session对象
|
|
||||||
:param cookies: 可以接收`CookieJar`、`list`、`tuple`、`str`、`dict`格式的`cookies`
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if self._page.mode == 'd' and self._page._has_driver:
|
|
||||||
self._chromium_setter.cookies(cookies)
|
|
||||||
elif self._page.mode == 's' and self._page._has_session:
|
|
||||||
self._session_setter.cookies(cookies)
|
|
||||||
|
|
||||||
def headers(self, headers) -> None:
|
|
||||||
"""设置固定发送的headers
|
|
||||||
:param headers: dict格式的headers数据
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if self._page.mode == 's':
|
|
||||||
self._session_setter.headers(headers)
|
|
||||||
else:
|
|
||||||
self._chromium_setter.headers(headers)
|
|
||||||
|
|
||||||
def user_agent(self, ua, platform=None):
|
|
||||||
"""设置user agent,d模式下只有当前tab有效"""
|
|
||||||
if self._page.mode == 's':
|
|
||||||
self._session_setter.user_agent(ua)
|
|
||||||
else:
|
|
||||||
self._chromium_setter.user_agent(ua, platform)
|
|
||||||
|
@ -12,12 +12,13 @@ from .base import BasePage
|
|||||||
from .chromium_driver import ChromiumDriver
|
from .chromium_driver import ChromiumDriver
|
||||||
from .chromium_element import ChromiumElement
|
from .chromium_element import ChromiumElement
|
||||||
from .chromium_frame import ChromiumFrame
|
from .chromium_frame import ChromiumFrame
|
||||||
from .chromium_page import ChromiumPage, ChromiumPageSetter
|
from .chromium_page import ChromiumPage
|
||||||
from .chromium_tab import WebPageTab
|
from .chromium_tab import WebPageTab
|
||||||
from .configs.chromium_options import ChromiumOptions
|
from .configs.chromium_options import ChromiumOptions
|
||||||
from .configs.session_options import SessionOptions
|
from .configs.session_options import SessionOptions
|
||||||
from .session_element import SessionElement
|
from .session_element import SessionElement
|
||||||
from .session_page import SessionPage, SessionPageSetter
|
from .session_page import SessionPage
|
||||||
|
from .setter import WebPageSetter
|
||||||
|
|
||||||
|
|
||||||
class WebPage(SessionPage, ChromiumPage, BasePage):
|
class WebPage(SessionPage, ChromiumPage, BasePage):
|
||||||
@ -166,13 +167,3 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
|
|||||||
def _on_download_begin(self, **kwargs): ...
|
def _on_download_begin(self, **kwargs): ...
|
||||||
|
|
||||||
|
|
||||||
class WebPageSetter(ChromiumPageSetter):
|
|
||||||
_page: WebPage = ...
|
|
||||||
_session_setter: SessionPageSetter = ...
|
|
||||||
_chromium_setter: ChromiumPageSetter = ...
|
|
||||||
|
|
||||||
def user_agent(self, ua: str, platform: str = None) -> None: ...
|
|
||||||
|
|
||||||
def headers(self, headers: dict) -> None: ...
|
|
||||||
|
|
||||||
def cookies(self, cookies) -> None: ...
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user