Pre Merge pull request !22 from donggoing/dev

DrissionPage/commons/tools.py

@@ -7,6 +7,7 @@ from platform import system
 from pathlib import Path
 from re import search, sub
 from shutil import rmtree
+from time import perf_counter, sleep
 
 
 def get_usable_path(path):
@@ -177,6 +178,42 @@ def get_chrome_hwnds_from_pid(pid, title):
     EnumWindows(callback, hwnds)
     return hwnds
 
+def wait_until(page, condition, timeout=10, poll=0.1, raise_err=True):
+    """等待返回值不为False或空，直到超时
+    :param page (DrissionPage): DrissionPage对象
+    :param condition (function | str | tuple): 等待条件，返回值不为False则停止等待
+    :param timeout (float, optional): 超时时间
+    :param poll (float, optional): 轮询间隔
+    :param message (str, optional): 超时时的报错信息
+    :param ignored_exceptions (bool, optional): 是否忽略异常
+    :return: DP Element or bool
+    """
+    end_time = perf_counter() + timeout
+    if isinstance(condition, str) or isinstance(condition, tuple):
+        if not callable(getattr(page, 's_ele', None)):
+            raise AttributeError('page对象缺少s_ele方法')
+        condition_method = lambda page: page.s_ele(condition)
+    elif callable(condition):
+        condition_method = condition
+    else:
+        raise ValueError('condition必须是函数或者字符串或者元组')
+    while perf_counter() < end_time:
+        try:
+            value = condition_method(page)
+            if value:
+                return value
+        except Exception as exc:
+            pass
+            
+        sleep(poll)
+        if perf_counter() > end_time:
+            break
+        
+    if raise_err:
+        raise TimeoutError('等待超时')
+    else:
+        return False
+
 # def get_exe_from_port(port):
 #     """获取端口号第一条进程的可执行文件路径
 #     :param port: 端口号

DrissionPage/commons/tools.pyi

@@ -6,6 +6,7 @@
 from os import popen
 from pathlib import Path
 from typing import Union
+from types import FunctionType
 
 from chromium_page import ChromiumPage
 
@@ -38,3 +39,5 @@ def get_browser_progress_id(progress: Union[popen, None], address: str) -> Union
 
 
 def get_chrome_hwnds_from_pid(pid: Union[str, int], title: str) -> list: ...
+
+def wait_until(page, condition: Union[FunctionType, str, tuple], timeout: float, poll: float, raise_err: bool): ...