自动等待填写上传路径;增加remove_ele()方法;新加载的iframe自动等待完成;wait增加input_upload_paths()

This commit is contained in:
g1879 2023-02-19 15:56:36 +08:00
parent 76920e05d8
commit 0a73226c4f
7 changed files with 57 additions and 24 deletions

View File

@ -16,7 +16,7 @@ from .chromium_element import ChromiumWaiter, ChromiumScroll, ChromiumElement, r
ChromiumElementWaiter
from .common.constants import HANDLE_ALERT_METHOD, ERROR, NoneElement
from .common.errors import ContextLossError, ElementLossError, AlertExistsError, CallMethodError, TabClosedError, \
NoRectError
NoRectError, ElementNotFoundError
from .common.locator import get_loc
from .common.tools import get_usable_path
from .common.web import offset_scroll, cookies_to_tuple
@ -212,7 +212,10 @@ class ChromiumBase(BasePage):
if self._upload_list:
files = self._upload_list if kwargs['mode'] == 'selectMultiple' else self._upload_list[:1]
self.run_cdp('DOM.setFileInputFiles', files=files, backendNodeId=kwargs['backendNodeId'])
self._upload_list = []
self.driver.Page.fileChooserOpened = None
self.run_cdp('Page.setInterceptFileChooserDialog', enabled=False)
self._upload_list = None
def __call__(self, loc_or_str, timeout=None):
"""在内部查找元素
@ -233,8 +236,7 @@ class ChromiumBase(BasePage):
@property
def _wait_driver(self):
"""返回用于控制浏览器的ChromiumDriver对象会先等待页面加载完毕"""
while self._is_loading:
sleep(.1)
self.wait.load_complete()
return self.driver
@property
@ -461,6 +463,8 @@ class ChromiumBase(BasePage):
loc = get_loc(loc_or_ele)[1]
elif isinstance(loc_or_ele, ChromiumElement) or str(type(loc_or_ele)).endswith(".ChromiumFrame'>"):
return loc_or_ele
elif not loc_or_ele:
raise ElementNotFoundError
else:
raise ValueError('loc_or_str参数只能是tuple、str、ChromiumElement类型。')
@ -557,6 +561,17 @@ class ChromiumBase(BasePage):
while self.ready_state != 'complete':
sleep(.1)
def remove_ele(self, loc_or_ele):
"""从页面上删除一个元素
:param loc_or_ele: 元素对象或定位符
:return: None
"""
if not loc_or_ele:
return
ele = self._ele(loc_or_ele)
if ele:
self.run_cdp('DOM.removeNode', nodeId=ele.ids.node_id)
def get_session_storage(self, item=None):
"""获取sessionStorage信息不设置item则获取全部
:param item: 要获取的项不设置则返回全部
@ -850,7 +865,7 @@ class ChromiumBaseSetter(object):
:param files: 文件路径列表或字符串字符串时多个文件用回车分隔
:return: None
"""
if self._page._upload_list is None:
if not self._page._upload_list:
self._page.driver.Page.fileChooserOpened = self._page._onFileChooserOpened
self._page.run_cdp('Page.setInterceptFileChooserDialog', enabled=True)
@ -885,7 +900,7 @@ class ChromiumPageWaiter(ChromiumWaiter):
while perf_counter() < end_time:
if self._driver.is_loading == start:
return True
sleep(.005)
sleep(.01)
return False
def load_start(self, timeout=None):
@ -902,6 +917,11 @@ class ChromiumPageWaiter(ChromiumWaiter):
"""
return self._loading(timeout=timeout, start=False)
def input_upload_paths(self):
"""等待自动填写上传文件路径"""
while self._driver._upload_list:
sleep(.01)
class ChromiumPageScroll(ChromiumScroll):
def __init__(self, page):

View File

@ -13,7 +13,7 @@ from requests.cookies import RequestsCookieJar
from .common.constants import NoneElement
from .base import BasePage
from .chromium_driver import ChromiumDriver
from .chromium_element import ChromiumElement, ChromiumScroll, ChromiumElementWaiter, ChromiumWaiter
from .chromium_element import ChromiumElement, ChromiumScroll, ChromiumWaiter
from .chromium_frame import ChromiumFrame
from .session_element import SessionElement
@ -68,7 +68,7 @@ class ChromiumBase(BasePage):
def _set_runtime_settings(self) -> None: ...
def __call__(self, loc_or_str: Union[Tuple[str, str], str, ChromiumElement],
timeout: float = None) -> Union[ChromiumElement, ChromiumFrame, None]: ...
timeout: float = None) -> Union[ChromiumElement, ChromiumFrame, NoneElement]: ...
@property
def title(self) -> str: ...
@ -164,6 +164,8 @@ class ChromiumBase(BasePage):
def stop_loading(self) -> None: ...
def remove_ele(self, loc_or_ele: Union[ChromiumElement, ChromiumFrame, str, Tuple[str, str]]) -> None: ...
def run_cdp(self, cmd: str, **cmd_args) -> dict: ...
def run_cdp_loaded(self, cmd: str, **cmd_args) -> dict: ...
@ -202,6 +204,8 @@ class ChromiumPageWaiter(ChromiumWaiter):
def load_complete(self, timeout: Union[int, float] = None) -> bool: ...
def input_upload_paths(self) -> None: ...
class ChromiumPageScroll(ChromiumScroll):
def __init__(self, page: ChromiumBase): ...

View File

@ -4,11 +4,12 @@
@Contact : g1879@qq.com
"""
from re import search
from time import sleep
from time import sleep, perf_counter
from warnings import warn
from .chromium_base import ChromiumBase, ChromiumPageScroll, ChromiumBaseSetter
from .chromium_element import ChromiumElement
from .common.errors import ElementNotFoundError
class ChromiumFrame(ChromiumBase):
@ -32,6 +33,10 @@ class ChromiumFrame(ChromiumBase):
self.doc_ele = ChromiumElement(self, obj_id=obj_id)
self._ids = ChromiumFrameIds(self)
end_time = perf_counter() + 2
while perf_counter() < end_time and self.url == 'about:blank':
sleep(.1)
def __call__(self, loc_or_str, timeout=None):
"""在内部查找元素
ele2 = ele1('@id=ele_id')
@ -79,7 +84,7 @@ class ChromiumFrame(ChromiumBase):
self.doc_ele = ChromiumElement(self, obj_id=obj_id)
def _check_ok(self):
""""""
"""检查iframe元素是否还能使用不能使用则重新加载"""
if self._tab_obj._stopped.is_set():
self._reload()
@ -383,11 +388,13 @@ class ChromiumFrame(ChromiumBase):
:param timeout: 查找超时时间
:return: ChromiumElement对象
"""
if not loc_or_ele:
raise ElementNotFoundError
if isinstance(loc_or_ele, ChromiumElement):
return loc_or_ele
while self.is_loading:
sleep(.05)
self.wait.load_complete()
return self.doc_ele.ele(loc_or_ele, timeout) if single else self.doc_ele.eles(loc_or_ele, timeout)
@ -409,8 +416,7 @@ class ChromiumFrame(ChromiumBase):
result = self.driver.Page.navigate(url=to_url, frameId=self.frame_id)
is_timeout = not self._wait_loaded(timeout)
while self.is_loading:
sleep(.1)
self.wait.load_complete()
if is_timeout:
err = TimeoutError('页面连接超时。')

View File

@ -1,5 +1,5 @@
# -*- coding:utf-8 -*-
from .errors import NotElementFoundError
from .errors import ElementNotFoundError
HANDLE_ALERT_METHOD = 'Page.handleJavaScriptDialog'
FRAME_ELEMENT = ('iframe', 'frame')
@ -9,16 +9,16 @@ ERROR = 'error'
class NoneElement(object):
_instance = None
def __new__(cls, *args, **kwargs):
def __new__(cls):
if not cls._instance:
cls._instance = super(NoneElement, cls).__new__(cls, *args, **kwargs)
cls._instance = super(NoneElement, cls).__new__(cls)
return cls._instance
def __call__(self, *args, **kwargs):
raise NotElementFoundError
raise ElementNotFoundError
def __getattr__(self, item):
raise NotElementFoundError
raise ElementNotFoundError
def __eq__(self, other):
if other is None:

View File

@ -21,7 +21,7 @@ class ContextLossError(BaseError):
class ElementLossError(BaseError):
_info = '该元素对象已不在当前页面中'
_info = '页面内无此对象,可能因刷新已失效'
class CallMethodError(BaseError):
@ -32,7 +32,7 @@ class TabClosedError(BaseError):
_info = '标签页已关闭。'
class NotElementFoundError(BaseError):
class ElementNotFoundError(BaseError):
_info = '没有找到元素。'

View File

@ -10,11 +10,11 @@ from typing import Union
from selenium import webdriver
from DrissionPage.mixpage.drission import Drission
from .common.tools import unzip
from .configs.chromium_options import ChromiumOptions
from .configs.driver_options import DriverOptions
from .configs.options_manage import OptionsManager
from DrissionPage.mixpage.drission import Drission
from .common.tools import unzip
from .session_page import SessionPage

View File

@ -14,8 +14,9 @@ from requests.structures import CaseInsensitiveDict
from tldextract import extract
from .base import BasePage
from .configs.session_options import SessionOptions
from .common.errors import ElementNotFoundError
from .common.web import cookie_to_dict, set_session_cookies
from .configs.session_options import SessionOptions
from .session_element import SessionElement, make_session_ele
@ -199,6 +200,8 @@ class SessionPage(BasePage):
:param single: True则返回第一个False则返回全部
:return: SessionElement对象
"""
if not loc_or_ele:
raise ElementNotFoundError
return loc_or_ele if isinstance(loc_or_ele, SessionElement) else make_session_ele(self, loc_or_ele, single)
def get_cookies(self, as_dict=False, all_domains=False):