diff --git a/DrissionPage/chromium_base.py b/DrissionPage/chromium_base.py index a046cc1..a77e9f1 100644 --- a/DrissionPage/chromium_base.py +++ b/DrissionPage/chromium_base.py @@ -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.locator import get_loc from .commons.tools import get_usable_path, clean_folder -from .commons.web import set_browser_cookies from .errors import ContextLossError, ElementLossError, AlertExistsError, CDPError, TabClosedError, \ NoRectError, BrowserConnectError, GetDocumentError from .network_listener import NetworkListener from .session_element import make_session_ele +from .setter import ChromiumBaseSetter from .waiter import ChromiumBaseWaiter @@ -890,112 +890,6 @@ class ChromiumBase(BasePage): 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): def __init__(self, page): """ @@ -1055,62 +949,6 @@ class Timeout(object): 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): def __init__(self, page): self._page = page diff --git a/DrissionPage/chromium_base.pyi b/DrissionPage/chromium_base.pyi index 160609d..ebbbd1b 100644 --- a/DrissionPage/chromium_base.pyi +++ b/DrissionPage/chromium_base.pyi @@ -8,9 +8,7 @@ from typing import Union, Tuple, List, Any from DataRecorder import Recorder from requests import Session -from requests.cookies import RequestsCookieJar -from .waiter import ChromiumBaseWaiter from .base import BasePage from .chromium_driver import ChromiumDriver from .chromium_element import ChromiumElement, ChromiumScroll @@ -18,6 +16,8 @@ from .chromium_frame import ChromiumFrame from .commons.constants import NoneElement from .network_listener import NetworkListener from .session_element import SessionElement +from .setter import ChromiumBaseSetter +from .waiter import ChromiumBaseWaiter class ChromiumBase(BasePage): @@ -222,37 +222,6 @@ class ChromiumPageScroll(ChromiumScroll): 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): def __init__(self, page: ChromiumBase, implicit=None, page_load=None, script=None): @@ -262,28 +231,6 @@ class Timeout(object): 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): def __init__(self, page: ChromiumBase): self._page: ChromiumBase = ... diff --git a/DrissionPage/chromium_element.py b/DrissionPage/chromium_element.py index 68b63fe..ad1930f 100644 --- a/DrissionPage/chromium_element.py +++ b/DrissionPage/chromium_element.py @@ -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, \ CDPError, NoResourceError, CanNotClickError from .session_element import make_session_ele +from .setter import ChromiumElementSetter from .waiter import ChromiumElementWaiter @@ -1467,38 +1468,6 @@ class ShadowRootStates(object): 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): def __init__(self, ele): """ diff --git a/DrissionPage/chromium_element.pyi b/DrissionPage/chromium_element.pyi index ae78c42..3b5dfe1 100644 --- a/DrissionPage/chromium_element.pyi +++ b/DrissionPage/chromium_element.pyi @@ -12,6 +12,7 @@ from .chromium_frame import ChromiumFrame from .chromium_page import ChromiumPage from .commons.constants import NoneElement from .session_element import SessionElement +from .setter import ChromiumElementSetter from .waiter import ChromiumElementWaiter 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: ... -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): def __init__(self, ele: ChromiumShadowRoot): """ diff --git a/DrissionPage/chromium_frame.py b/DrissionPage/chromium_frame.py index 65a76ae..34aa115 100644 --- a/DrissionPage/chromium_frame.py +++ b/DrissionPage/chromium_frame.py @@ -7,10 +7,11 @@ from re import search from threading import Thread 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 .commons.tools import get_usable_path from .errors import ContextLossError +from .setter import ChromiumFrameSetter 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) 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) diff --git a/DrissionPage/chromium_frame.pyi b/DrissionPage/chromium_frame.pyi index 0f56dab..a2bdce8 100644 --- a/DrissionPage/chromium_frame.pyi +++ b/DrissionPage/chromium_frame.pyi @@ -6,9 +6,10 @@ from pathlib import Path from typing import Union, Tuple, List, Any -from .waiter import FrameWaiter -from .chromium_base import ChromiumBase, ChromiumPageScroll, ChromiumBaseSetter +from .chromium_base import ChromiumBase, ChromiumPageScroll from .chromium_element import ChromiumElement, Locations, ChromiumElementStates +from .setter import ChromiumFrameSetter +from .waiter import FrameWaiter class ChromiumFrame(ChromiumBase): @@ -205,9 +206,3 @@ class ChromiumFrameScroll(ChromiumPageScroll): def __init__(self, frame: ChromiumFrame) -> 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: ... diff --git a/DrissionPage/chromium_page.py b/DrissionPage/chromium_page.py index 6dd7097..4903560 100644 --- a/DrissionPage/chromium_page.py +++ b/DrissionPage/chromium_page.py @@ -3,16 +3,16 @@ @Author : g1879 @Contact : g1879@qq.com """ -from platform import system 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_tab import ChromiumTab from .commons.browser import connect_browser from .commons.tools import port_is_using from .configs.chromium_options import ChromiumOptions from .errors import BrowserConnectError +from .setter import ChromiumPageSetter from .waiter import ChromiumPageWaiter @@ -689,191 +689,6 @@ class Alert(object): 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): if '.' in rename: return rename diff --git a/DrissionPage/chromium_page.pyi b/DrissionPage/chromium_page.pyi index 47115cb..916d85c 100644 --- a/DrissionPage/chromium_page.pyi +++ b/DrissionPage/chromium_page.pyi @@ -3,10 +3,10 @@ @Author : g1879 @Contact : g1879@qq.com """ -from os import popen 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_tab import ChromiumTab from .configs.chromium_options import ChromiumOptions @@ -21,7 +21,7 @@ class ChromiumPage(ChromiumBase): timeout: float = None): self._driver_options: ChromiumOptions = ... self._process_id: str = ... - self._window_setter: WindowSetter = ... + # self._window_setter: WindowSetter = ... self._main_tab: str = ... self._alert: Alert = ... self._browser_driver: ChromiumDriver = ... @@ -197,51 +197,4 @@ class Alert(object): 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: ... diff --git a/DrissionPage/chromium_tab.py b/DrissionPage/chromium_tab.py index 256569b..9279600 100644 --- a/DrissionPage/chromium_tab.py +++ b/DrissionPage/chromium_tab.py @@ -5,9 +5,10 @@ """ 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 .session_page import SessionPage, SessionPageSetter +from .session_page import SessionPage +from .setter import WebPageTabSetter class ChromiumTab(ChromiumBase): @@ -327,39 +328,3 @@ class WebPageTab(SessionPage, ChromiumTab): elif self._mode == 'd': return super(SessionPage, self)._find_elements(loc_or_ele, timeout=timeout, single=single, 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) - - diff --git a/DrissionPage/chromium_tab.pyi b/DrissionPage/chromium_tab.pyi index 6a99610..04f3ad6 100644 --- a/DrissionPage/chromium_tab.pyi +++ b/DrissionPage/chromium_tab.pyi @@ -7,12 +7,13 @@ from typing import Union, Tuple, Any, List from requests import Session, Response -from .chromium_base import ChromiumBase, ChromiumBaseSetter +from .chromium_base import ChromiumBase from .chromium_element import ChromiumElement from .chromium_frame import ChromiumFrame from .chromium_page import ChromiumPage, ChromiumTabRect 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 @@ -149,15 +150,3 @@ class WebPageTab(SessionPage, ChromiumTab): 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, 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: ... diff --git a/DrissionPage/commons/tools.py b/DrissionPage/commons/tools.py index a95dc7d..5adf7ca 100644 --- a/DrissionPage/commons/tools.py +++ b/DrissionPage/commons/tools.py @@ -3,6 +3,7 @@ @Author : g1879 @Contact : g1879@qq.com """ +from platform import system from pathlib import Path from re import search, sub from shutil import rmtree @@ -101,6 +102,80 @@ def clean_folder(folder_path, ignore=None): elif f.is_dir(): 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): # """获取端口号第一条进程的可执行文件路径 # :param port: 端口号 diff --git a/DrissionPage/commons/tools.pyi b/DrissionPage/commons/tools.pyi index f7b91e8..54b8197 100644 --- a/DrissionPage/commons/tools.pyi +++ b/DrissionPage/commons/tools.pyi @@ -3,9 +3,12 @@ @Author : g1879 @Contact : g1879@qq.com """ +from os import popen from pathlib import Path from typing import Union +from chromium_page import ChromiumPage + # 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 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: ... diff --git a/DrissionPage/session_page.py b/DrissionPage/session_page.py index cc67f6e..b6c1193 100644 --- a/DrissionPage/session_page.py +++ b/DrissionPage/session_page.py @@ -12,9 +12,10 @@ from requests.structures import CaseInsensitiveDict from tldextract import extract 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 .session_element import SessionElement, make_session_ele +from .setter import SessionPageSetter class SessionPage(BasePage): @@ -304,137 +305,6 @@ class SessionPage(BasePage): 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: """检查kwargs或headers中是否有arg所示属性""" return arg in kwargs['headers'] or arg in headers diff --git a/DrissionPage/session_page.pyi b/DrissionPage/session_page.pyi index bb803f0..5391a4a 100644 --- a/DrissionPage/session_page.pyi +++ b/DrissionPage/session_page.pyi @@ -3,20 +3,16 @@ @Author : g1879 @Contact : g1879@qq.com """ -from http.cookiejar import Cookie from typing import Any, Union, Tuple, List -# from DownloadKit import DownloadKit 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 .base import BasePage from .commons.constants import NoneElement from .configs.session_options import SessionOptions from .session_element import SessionElement +from .setter import SessionPageSetter class SessionPage(BasePage): @@ -160,47 +156,6 @@ class SessionPage(BasePage): **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], arg: str) -> bool: ... diff --git a/DrissionPage/setter.py b/DrissionPage/setter.py new file mode 100644 index 0000000..e14f23f --- /dev/null +++ b/DrissionPage/setter.py @@ -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) diff --git a/DrissionPage/setter.pyi b/DrissionPage/setter.pyi new file mode 100644 index 0000000..e750130 --- /dev/null +++ b/DrissionPage/setter.pyi @@ -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: ... diff --git a/DrissionPage/web_page.py b/DrissionPage/web_page.py index 059597b..e85bdca 100644 --- a/DrissionPage/web_page.py +++ b/DrissionPage/web_page.py @@ -8,12 +8,13 @@ from requests import Session from .base import BasePage from .chromium_base import ChromiumBase, Timeout from .chromium_driver import ChromiumDriver -from .chromium_page import ChromiumPage, ChromiumPageSetter +from .chromium_page import ChromiumPage 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 .session_page import SessionPage, SessionPageSetter +from .session_page import SessionPage +from .setter import WebPageSetter class WebPage(SessionPage, ChromiumPage, BasePage): @@ -426,35 +427,3 @@ class WebPage(SessionPage, ChromiumPage, BasePage): 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) diff --git a/DrissionPage/web_page.pyi b/DrissionPage/web_page.pyi index d153c30..1621fd9 100644 --- a/DrissionPage/web_page.pyi +++ b/DrissionPage/web_page.pyi @@ -12,12 +12,13 @@ from .base import BasePage from .chromium_driver import ChromiumDriver from .chromium_element import ChromiumElement from .chromium_frame import ChromiumFrame -from .chromium_page import ChromiumPage, ChromiumPageSetter +from .chromium_page import ChromiumPage from .chromium_tab import WebPageTab from .configs.chromium_options import ChromiumOptions from .configs.session_options import SessionOptions 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): @@ -166,13 +167,3 @@ class WebPage(SessionPage, ChromiumPage, BasePage): 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: ...