diff --git a/DrissionPage/_elements/chromium_element.py b/DrissionPage/_elements/chromium_element.py index 8754c2d..01fa730 100644 --- a/DrissionPage/_elements/chromium_element.py +++ b/DrissionPage/_elements/chromium_element.py @@ -981,7 +981,7 @@ def find_by_xpath(ele, xpath, index, timeout, relative=True): res = ele.owner._run_cdp('Runtime.getProperties', objectId=res['result']['objectId'], ownProperties=True)['result'][:-1] if index is None: - r = ChromiumElementsList(page=ele.owner) + r = ChromiumElementsList(owner=ele.owner) for i in res: if i['value']['type'] == 'object': r.append(make_chromium_eles(ele.owner, _ids=i['value']['objectId'], is_obj_id=True)) @@ -1010,7 +1010,7 @@ def find_by_xpath(ele, xpath, index, timeout, relative=True): if result: return result - return NoneElement(ele.owner) if index is not None else ChromiumElementsList(page=ele.owner) + return NoneElement(ele.owner) if index is not None else ChromiumElementsList(owner=ele.owner) def find_by_css(ele, selector, index, timeout): @@ -1049,7 +1049,7 @@ def find_by_css(ele, selector, index, timeout): if result: return result - return NoneElement(ele.owner) if index is not None else ChromiumElementsList(page=ele.owner) + return NoneElement(ele.owner) if index is not None else ChromiumElementsList(owner=ele.owner) def make_chromium_eles(page, _ids, index=1, is_obj_id=True, ele_only=False): @@ -1073,7 +1073,7 @@ def make_chromium_eles(page, _ids, index=1, is_obj_id=True, ele_only=False): return get_node_func(page, obj_id, ele_only) else: # 获取全部 - nodes = ChromiumElementsList(page=page) + nodes = ChromiumElementsList(owner=page) for obj_id in _ids: # if obj_id == 0: # continue diff --git a/DrissionPage/_elements/session_element.py b/DrissionPage/_elements/session_element.py index 7502c47..7ed8bdb 100644 --- a/DrissionPage/_elements/session_element.py +++ b/DrissionPage/_elements/session_element.py @@ -266,7 +266,7 @@ def make_session_ele(html_or_ele, loc=None, index=1, method=None): # 把lxml元素对象包装成SessionElement对象并按需要返回一个或全部 if index is None: - r = SessionElementsList(page=page) + r = SessionElementsList(owner=page) for e in eles: if e != '\n': r.append(SessionElement(e, page) if isinstance(e, HtmlElement) else e) diff --git a/DrissionPage/_pages/chromium_base.py b/DrissionPage/_pages/chromium_base.py index f33c311..a41658a 100644 --- a/DrissionPage/_pages/chromium_base.py +++ b/DrissionPage/_pages/chromium_base.py @@ -42,10 +42,6 @@ class ChromiumBase(BasePage): """标签页、Frame、Page基类""" def __init__(self, browser, target_id=None): - """ - :param browser: Chromium - :param target_id: 要控制的target id，不指定默认为激活的标签页 - """ super().__init__() self._browser = browser self._is_loading = None @@ -72,14 +68,13 @@ class ChromiumBase(BasePage): self._d_set_runtime_settings() self._connect_browser(target_id) + def __call__(self, locator, index=1, timeout=None): + return self.ele(locator, index, timeout) + def _d_set_runtime_settings(self): pass def _connect_browser(self, target_id=None): - """连接浏览器，在第一次时运行 - :param target_id: 要控制的target id，不指定默认为激活的标签页 - :return: None - """ self._is_reading = False if not target_id: @@ -109,10 +104,6 @@ class ChromiumBase(BasePage): self._ready_state = 'complete' def _driver_init(self, target_id): - """新建页面、页面刷新后要进行的cdp参数初始化 - :param target_id: 要跳转到的target id - :return: None - """ self._is_loading = True self._driver = self.browser._get_driver(target_id, self) @@ -139,10 +130,6 @@ class ChromiumBase(BasePage): self._driver.set_callback('Page.frameDetached', self._onFrameDetached) def _get_document(self, timeout=10): - """获取页面文档 - :param timeout: 超时时间（秒） - :return: 是否获取成功 - """ if self._is_reading: return self._is_reading = True @@ -185,7 +172,6 @@ class ChromiumBase(BasePage): self.browser._frames[kwargs['frameId']] = self.tab_id def _onFrameStartedLoading(self, **kwargs): - """页面开始加载时执行""" self.browser._frames[kwargs['frameId']] = self.tab_id if kwargs['frameId'] == self._frame_id: self._doc_got = False @@ -198,14 +184,12 @@ class ChromiumBase(BasePage): t.start() def _onFrameNavigated(self, **kwargs): - """页面跳转时执行""" if kwargs['frame']['id'] == self._frame_id: self._doc_got = False self._ready_state = 'loading' self._is_loading = True def _onDomContentEventFired(self, **kwargs): - """在页面刷新、变化后重新读取页面内容""" if self._load_mode == 'eager': self._run_cdp('Page.stopLoading') if self._get_document(self._load_end_time - perf_counter() - .1): @@ -213,13 +197,11 @@ class ChromiumBase(BasePage): self._ready_state = 'interactive' def _onLoadEventFired(self, **kwargs): - """在页面刷新、变化后重新读取页面内容""" if self._doc_got is False and self._get_document(self._load_end_time - perf_counter() - .1): self._doc_got = True self._ready_state = 'complete' def _onFrameStoppedLoading(self, **kwargs): - """页面加载完成后执行""" self.browser._frames[kwargs['frameId']] = self.tab_id if kwargs['frameId'] == self._frame_id: if self._doc_got is False: @@ -227,7 +209,6 @@ class ChromiumBase(BasePage): self._ready_state = 'complete' def _onFileChooserOpened(self, **kwargs): - """文件选择框打开时执行""" if self._upload_list: if 'backendNodeId' not in kwargs: raise TypeError('该输入框无法接管，请改用对元素输入路径的方法设置。') @@ -238,18 +219,7 @@ class ChromiumBase(BasePage): self._run_cdp('Page.setInterceptFileChooserDialog', enabled=False) self._upload_list = None - def __call__(self, locator, index=1, timeout=None): - """在内部查找元素 - 例：ele = page('@id=ele_id') - :param locator: 元素的定位信息，可以是loc元组，或查询字符串 - :param index: 获取第几个元素，从1开始，可传入负数获取倒数第几个 - :param timeout: 超时时间（秒） - :return: ChromiumElement对象 - """ - return self.ele(locator, index, timeout) - def _wait_to_stop(self): - """eager策略超时时使页面停止加载""" end_time = perf_counter() + self.timeouts.page_load while perf_counter() < end_time: sleep(.1) @@ -259,28 +229,24 @@ class ChromiumBase(BasePage): # ----------挂件---------- @property def wait(self): - """返回用于等待的对象""" if self._wait is None: self._wait = BaseWaiter(self) return self._wait @property def set(self): - """返回用于设置的对象""" if self._set is None: self._set = ChromiumBaseSetter(self) return self._set @property def screencast(self): - """返回用于录屏的对象""" if self._screencast is None: self._screencast = Screencast(self) return self._screencast @property def actions(self): - """返回用于执行动作链的对象""" if self._actions is None: self._actions = Actions(self) self.wait.doc_loaded() @@ -288,21 +254,18 @@ class ChromiumBase(BasePage): @property def listen(self): - """返回用于聆听数据包的对象""" if self._listener is None: self._listener = Listener(self) return self._listener @property def states(self): - """返回用于获取状态信息的对象""" if self._states is None: self._states = PageStates(self) return self._states @property def scroll(self): - """返回用于滚动滚动条的对象""" self.wait.doc_loaded() if self._scroll is None: self._scroll = PageScroller(self) @@ -310,7 +273,6 @@ class ChromiumBase(BasePage): @property def rect(self): - """返回获取窗口坐标和大小的对象""" # self.wait.doc_loaded() if self._rect is None: self._rect = TabRect(self) @@ -318,19 +280,16 @@ class ChromiumBase(BasePage): @property def console(self): - """返回获取控制台信息的对象""" if self._console is None: self._console = Console(self) return self._console @property def timeout(self): - """返回timeout设置""" return self._timeouts.base @property def timeouts(self): - """返回timeouts设置""" return self._timeouts # ----------挂件结束---------- @@ -341,35 +300,29 @@ class ChromiumBase(BasePage): @property def driver(self): - """返回用于控制浏览器的Driver对象""" if self._driver is None: raise RuntimeError('浏览器已关闭或链接已断开。') return self._driver @property def title(self): - """返回当前页面title""" return self._run_cdp_loaded('Target.getTargetInfo', targetId=self._target_id)['targetInfo']['title'] @property def url(self): - """返回当前页面url""" return self._run_cdp_loaded('Target.getTargetInfo', targetId=self._target_id)['targetInfo']['url'] @property def _browser_url(self): - """用于被MixTab覆盖""" return self.url @property def html(self): - """返回当前页面html文本""" self.wait.doc_loaded() return self._run_cdp('DOM.getOuterHTML', objectId=self._root_id)['outerHTML'] @property def json(self): - """当返回内容是json格式时，返回对应的字典，非json格式时返回None""" try: return loads(self('t:pre', timeout=.5).text) except JSONDecodeError: @@ -377,37 +330,30 @@ class ChromiumBase(BasePage): @property def tab_id(self): - """返回当前标签页id""" return self._target_id @property def _target_id(self): - """返回当前标签页id""" return self.driver.id if self.driver.is_running else '' @property def active_ele(self): - """返回当前焦点所在元素""" return self._run_js_loaded('return document.activeElement;') @property def load_mode(self): - """返回页面加载策略，有3种：'none'、'normal'、'eager'""" return self._load_mode @property def user_agent(self): - """返回user agent""" return self._run_cdp('Runtime.evaluate', expression='navigator.userAgent;')['result']['value'] @property def upload_list(self): - """返回等待上传文件列表""" return self._upload_list @property def _js_ready_state(self): - """返回js获取的ready state信息""" try: return self._run_cdp('Runtime.evaluate', expression='document.readyState;', _timeout=3)['result']['value'] except ContextLostError: @@ -416,114 +362,47 @@ class ChromiumBase(BasePage): return 'timeout' def run_cdp(self, cmd, **cmd_args): - """执行Chrome DevTools Protocol语句 - :param cmd: 协议项目 - :param cmd_args: 参数 - :return: 执行的结果 - """ r = self.driver.run(cmd, **cmd_args) return r if __ERROR__ not in r else raise_error(r, user=True) def run_cdp_loaded(self, cmd, **cmd_args): - """执行Chrome DevTools Protocol语句，执行前等待页面加载完毕 - :param cmd: 协议项目 - :param cmd_args: 参数 - :return: 执行的结果 - """ self.wait.doc_loaded() r = self.driver.run(cmd, **cmd_args) return r if __ERROR__ not in r else raise_error(r, user=True) def _run_cdp(self, cmd, **cmd_args): - """执行Chrome DevTools Protocol语句 - :param cmd: 协议项目 - :param cmd_args: 参数 - :return: 执行的结果 - """ ignore = cmd_args.pop('_ignore', None) r = self.driver.run(cmd, **cmd_args) return r if __ERROR__ not in r else raise_error(r, ignore) def _run_cdp_loaded(self, cmd, **cmd_args): - """执行Chrome DevTools Protocol语句，执行前等待页面加载完毕 - :param cmd: 协议项目 - :param cmd_args: 参数 - :return: 执行的结果 - """ self.wait.doc_loaded() return self._run_cdp(cmd, **cmd_args) def run_js(self, script, *args, as_expr=False, timeout=None): - """运行javascript代码 - :param script: js文本或js文件路径 - :param args: 参数，按顺序在js文本中对应arguments[0]、arguments[1]... - :param as_expr: 是否作为表达式运行，为True时args无效 - :param timeout: js超时时间（秒），为None则使用页面timeouts.script设置 - :return: 运行的结果 - """ return self._run_js(script, *args, as_expr=as_expr, timeout=timeout) def run_js_loaded(self, script, *args, as_expr=False, timeout=None): - """运行javascript代码，执行前等待页面加载完毕 - :param script: js文本或js文件路径 - :param args: 参数，按顺序在js文本中对应arguments[0]、arguments[1]... - :param as_expr: 是否作为表达式运行，为True时args无效 - :param timeout: js超时时间（秒），为None则使用页面timeouts.script属性值 - :return: 运行的结果 - """ self.wait.doc_loaded() return self._run_js(script, *args, as_expr=as_expr, timeout=timeout) def _run_js(self, script, *args, as_expr=False, timeout=None): - """运行javascript代码 - :param script: js文本或js文件路径 - :param args: 参数，按顺序在js文本中对应arguments[0]、arguments[1]... - :param as_expr: 是否作为表达式运行，为True时args无效 - :param timeout: js超时时间（秒），为None则使用页面timeouts.script设置 - :return: 运行的结果 - """ return run_js(self, script, as_expr, self.timeouts.script if timeout is None else timeout, args) def _run_js_loaded(self, script, *args, as_expr=False, timeout=None): - """运行javascript代码，执行前等待页面加载完毕 - :param script: js文本或js文件路径 - :param args: 参数，按顺序在js文本中对应arguments[0]、arguments[1]... - :param as_expr: 是否作为表达式运行，为True时args无效 - :param timeout: js超时时间（秒），为None则使用页面timeouts.script属性值 - :return: 运行的结果 - """ self.wait.doc_loaded() return run_js(self, script, as_expr, self.timeouts.script if timeout is None else timeout, args) def run_async_js(self, script, *args, as_expr=False): - """以异步方式执行js代码或js文件路径 - :param script: js文本 - :param args: 参数，按顺序在js文本中对应arguments[0]、arguments[1]... - :param as_expr: 是否作为表达式运行，为True时args无效 - :return: None - """ run_js(self, script, as_expr, 0, args) def get(self, url, show_errmsg=False, retry=None, interval=None, timeout=None): - """访问url - :param url: 目标url - :param show_errmsg: 是否显示和抛出异常 - :param retry: 重试次数，为None时使用页面对象retry_times属性值 - :param interval: 重试间隔（秒），为None时使用页面对象retry_interval属性值 - :param timeout: 连接超时时间（秒），为None时使用页面对象timeouts.page_load属性值 - :return: 目标url是否可用 - """ retry, interval, is_file = self._before_connect(url, retry, interval) self._url_available = self._d_connect(self._url, times=retry, interval=interval, show_errmsg=show_errmsg, timeout=timeout) return self._url_available def cookies(self, all_domains=False, all_info=False): - """返回cookies信息 - :param all_domains: 是否返回所有域的cookies - :param all_info: 是否返回所有信息，为False时只返回name、value、domain - :return: cookies信息 - """ txt = 'Storage' if all_domains else 'Network' cookies = self._run_cdp_loaded(f'{txt}.getCookies')['cookies'] @@ -535,51 +414,21 @@ class ChromiumBase(BasePage): return CookiesList(r) def ele(self, locator, index=1, timeout=None): - """获取一个符合条件的元素对象 - :param locator: 定位符或元素对象 - :param index: 获取第几个元素，从1开始，可传入负数获取倒数第几个 - :param timeout: 查找超时时间（秒），默认与页面等待时间一致 - :return: ChromiumElement对象 - """ return self._ele(locator, timeout=timeout, index=index, method='ele()') def eles(self, locator, timeout=None): - """获取所有符合条件的元素对象 - :param locator: 定位符或元素对象 - :param timeout: 查找超时时间（秒），默认与页面等待时间一致 - :return: ChromiumElement对象组成的列表 - """ return self._ele(locator, timeout=timeout, index=None) def s_ele(self, locator=None, index=1, timeout=None): - """查找一个符合条件的元素以SessionElement形式返回，处理复杂页面时效率很高 - :param locator: 元素的定位信息，可以是loc元组，或查询字符串 - :param index: 获取第几个，从1开始，可传入负数获取倒数第几个 - :param timeout: 查找元素超时时间（秒），默认与页面等待时间一致 - :return: SessionElement对象或属性、文本 - """ return (NoneElement(self, method='s_ele()', args={'locator': locator, 'index': index}) if locator and not self.wait.eles_loaded(locator, timeout=timeout) else make_session_ele(self, locator, index=index, method='s_ele()')) def s_eles(self, locator, timeout=None): - """查找所有符合条件的元素以SessionElement列表形式返回 - :param locator: 元素的定位信息，可以是loc元组，或查询字符串 - :param timeout: 查找元素超时时间（秒），默认与页面等待时间一致 - :return: SessionElement对象组成的列表 - """ return (make_session_ele(self, locator, index=None) if self.wait.eles_loaded(locator, timeout=timeout) else SessionElementsList()) def _find_elements(self, locator, timeout=None, index=1, relative=False, raise_err=None): - """执行元素查找 - :param locator: 定位符或元素对象 - :param timeout: 查找超时时间（秒） - :param index: 第几个结果，从1开始，可传入负数获取倒数第几个，为None返回所有 - :param relative: MixTab用的表示是否相对定位的参数 - :param raise_err: 找不到元素是是否抛出异常，为None时根据全局设置 - :return: ChromiumElement对象或元素对象组成的列表 - """ if isinstance(locator, (str, tuple)): loc = get_loc(locator)[1] elif locator._type in ('ChromiumElement', 'ChromiumFrame'): @@ -645,33 +494,17 @@ class ChromiumBase(BasePage): return r def refresh(self, ignore_cache=False): - """刷新当前页面 - :param ignore_cache: 是否忽略缓存 - :return: None - """ self._is_loading = True self._run_cdp('Page.reload', ignoreCache=ignore_cache) self.wait.load_start() def forward(self, steps=1): - """在浏览历史中前进若干步 - :param steps: 前进步数 - :return: None - """ self._forward_or_back(steps) def back(self, steps=1): - """在浏览历史中后退若干步 - :param steps: 后退步数 - :return: None - """ self._forward_or_back(-steps) def _forward_or_back(self, steps): - """执行浏览器前进或后退，会跳过url相同的历史记录 - :param steps: 步数 - :return: None - """ if steps == 0: return @@ -694,7 +527,6 @@ class ChromiumBase(BasePage): self._run_cdp('Page.navigateToHistoryEntry', entryId=nid) def stop_loading(self): - """页面停止加载""" try: self._run_cdp('Page.stopLoading') end_time = perf_counter() + 5 @@ -706,10 +538,6 @@ class ChromiumBase(BasePage): self._ready_state = 'complete' 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, raise_err=False) @@ -717,12 +545,6 @@ class ChromiumBase(BasePage): self._run_cdp('DOM.removeNode', nodeId=ele._node_id, _ignore=ElementLostError) def add_ele(self, html_or_info, insert_to=None, before=None): - """新建一个元素 - :param html_or_info: 新元素的html文本或信息。信息格式为：(tag, {attr1: value, ...}) - :param insert_to: 插入到哪个元素中，可接收元素对象和定位符，为None且为html添加到body，不为html不插入 - :param before: 在哪个子节点前面插入，可接收对象和定位符，为None插入到父元素末尾 - :return: 元素对象 - """ if isinstance(html_or_info, str): insert_to = self.ele(insert_to) if insert_to else self.ele('t:body') args = [html_or_info, insert_to] @@ -774,69 +596,33 @@ class ChromiumBase(BasePage): return ele def get_frame(self, loc_ind_ele, timeout=None): - """获取页面中一个frame对象 - :param loc_ind_ele: 定位符、iframe序号、ChromiumFrame对象，序号从1开始，可传入负数获取倒数第几个 - :param timeout: 查找元素超时时间（秒） - :return: ChromiumFrame对象 - """ return get_frame(self, loc_ind_ele=loc_ind_ele, timeout=timeout) def get_frames(self, locator=None, timeout=None): - """获取所有符合条件的frame对象 - :param locator: 定位符，为None时返回所有 - :param timeout: 查找超时时间（秒） - :return: ChromiumFrame对象组成的列表 - """ locator = locator or 'xpath://*[name()="iframe" or name()="frame"]' frames = self._ele(locator, timeout=timeout, index=None, raise_err=False) return [i for i in frames if i._type == 'ChromiumFrame'] def session_storage(self, item=None): - """返回sessionStorage信息，不设置item则获取全部 - :param item: 要获取的项，不设置则返回全部 - :return: sessionStorage一个或所有项内容 - """ js = f'sessionStorage.getItem("{item}")' if item else 'sessionStorage' return self._run_js_loaded(js, as_expr=True) def local_storage(self, item=None): - """返回localStorage信息，不设置item则获取全部 - :param item: 要获取的项目，不设置则返回全部 - :return: localStorage一个或所有项内容 - """ js = f'localStorage.getItem("{item}")' if item else 'localStorage' return self._run_js_loaded(js, as_expr=True) def get_screenshot(self, path=None, name=None, as_bytes=None, as_base64=None, full_page=False, left_top=None, right_bottom=None): - """对页面进行截图，可对整个网页、可见网页、指定范围截图。对可视范围外截图需要90以上版本浏览器支持 - :param path: 保存路径 - :param name: 完整文件名，后缀可选 'jpg','jpeg','png','webp' - :param as_bytes: 是否以字节形式返回图片，可选 'jpg','jpeg','png','webp'，生效时path参数和as_base64参数无效 - :param as_base64: 是否以base64字符串形式返回图片，可选 'jpg','jpeg','png','webp'，生效时path参数无效 - :param full_page: 是否整页截图，为True截取整个网页，为False截取可视窗口 - :param left_top: 截取范围左上角坐标 - :param right_bottom: 截取范围右下角角坐标 - :return: 图片完整路径或字节文本 - """ return self._get_screenshot(path=path, name=name, as_bytes=as_bytes, as_base64=as_base64, full_page=full_page, left_top=left_top, right_bottom=right_bottom) def add_init_js(self, script): - """添加初始化脚本，在页面加载任何脚本前执行 - :param script: js文本 - :return: 添加的脚本的id - """ js_id = self._run_cdp('Page.addScriptToEvaluateOnNewDocument', source=script, includeCommandLineAPI=True)['identifier'] self._init_jss.append(js_id) return js_id def remove_init_js(self, script_id=None): - """删除初始化脚本，js_id传入None时删除所有 - :param script_id: 脚本的id - :return: None - """ if script_id is None: for js_id in self._init_jss: self._run_cdp('Page.removeScriptToEvaluateOnNewDocument', identifier=js_id) @@ -847,13 +633,6 @@ class ChromiumBase(BasePage): self._init_jss.remove(script_id) def clear_cache(self, session_storage=True, local_storage=True, cache=True, cookies=True): - """清除缓存，可选要清除的项 - :param session_storage: 是否清除sessionStorage - :param local_storage: 是否清除localStorage - :param cache: 是否清除cache - :param cookies: 是否清除cookies - :return: None - """ if session_storage or local_storage: self._run_cdp_loaded('DOMStorage.enable') i = self._run_cdp('Storage.getStorageKeyForFrame', frameId=self._frame_id)['storageKey'] @@ -870,16 +649,11 @@ class ChromiumBase(BasePage): self._run_cdp_loaded('Network.clearBrowserCookies') def disconnect(self): - """断开与页面的连接，不关闭页面""" if self._driver: self._driver.stop() self.browser._all_drivers.get(self._driver.id, set()).discard(self._driver) def reconnect(self, wait=0): - """断开与页面原来的页面，重新建立连接 - :param wait: 断开后等待若干秒再连接 - :return: None - """ t_id = self._target_id self.disconnect() sleep(wait) @@ -889,13 +663,6 @@ class ChromiumBase(BasePage): self._get_document() def handle_alert(self, accept=True, send=None, timeout=None, next_one=False): - """处理提示框，可以自动等待提示框出现 - :param accept: True表示确认，False表示取消，为None不会按按钮但依然返回文本值 - :param send: 处理prompt提示框时可输入文本 - :param timeout: 等待提示框出现的超时时间（秒），为None则使用self.timeout属性的值 - :param next_one: 是否处理下一个出现的提示框，为True时timeout参数无效 - :return: 提示框内容文本，未等到提示框则返回False - """ r = self._handle_alert(accept=accept, send=send, timeout=timeout, next_one=next_one) if not isinstance(accept, bool): return r @@ -904,13 +671,6 @@ class ChromiumBase(BasePage): return r def _handle_alert(self, accept=True, send=None, timeout=None, next_one=False): - """处理提示框，可以自动等待提示框出现 - :param accept: True表示确认，False表示取消，其它值不会按按钮但依然返回文本值 - :param send: 处理prompt提示框时可输入文本 - :param timeout: 等待提示框出现的超时时间（秒），为None则使用self.timeout属性的值 - :param next_one: 是否处理下一个出现的提示框，为True时timeout参数无效 - :return: 提示框内容文本，未等到提示框则返回False - """ if next_one: self._alert.handle_next = accept self._alert.next_text = send @@ -934,7 +694,6 @@ class ChromiumBase(BasePage): return res_text def _on_alert_open(self, **kwargs): - """alert出现时触发的方法""" self._alert.activated = True self._alert.text = kwargs['message'] self._alert.type = kwargs['type'] @@ -952,7 +711,6 @@ class ChromiumBase(BasePage): self._alert.handle_next = None def _on_alert_close(self, **kwargs): - """alert关闭时触发的方法""" self._alert.activated = False self._alert.text = None self._alert.type = None @@ -962,10 +720,6 @@ class ChromiumBase(BasePage): self._has_alert = False def _wait_loaded(self, timeout=None): - """等待页面加载完成，超时触发停止加载 - :param timeout: 超时时间（秒） - :return: 是否成功，超时返回False - """ timeout = timeout if timeout is not None else self.timeouts.page_load end_time = perf_counter() + timeout while perf_counter() < end_time: @@ -984,14 +738,6 @@ class ChromiumBase(BasePage): return False def _d_connect(self, to_url, times=0, interval=1, show_errmsg=False, timeout=None): - """尝试连接，重试若干次 - :param to_url: 要访问的url - :param times: 重试次数 - :param interval: 重试间隔（秒） - :param show_errmsg: 是否抛出异常 - :param timeout: 连接超时时间（秒） - :return: 是否成功，返回None表示不确定 - """ err = None self._is_loading = True timeout = timeout if timeout is not None else self.timeouts.page_load @@ -1041,17 +787,6 @@ class ChromiumBase(BasePage): def _get_screenshot(self, path=None, name=None, as_bytes=None, as_base64=None, full_page=False, left_top=None, right_bottom=None, ele=None): - """对页面进行截图，可对整个网页、可见网页、指定范围截图。对可视范围外截图需要90以上版本浏览器支持 - :param path: 保存路径 - :param name: 完整文件名，后缀可选 'jpg','jpeg','png','webp' - :param as_bytes: 是否以字节形式返回图片，可选 'jpg','jpeg','png','webp'，生效时path参数和as_base64参数无效 - :param as_base64: 是否以base64字符串形式返回图片，可选 'jpg','jpeg','png','webp'，生效时path参数无效 - :param full_page: 是否整页截图，为True截取整个网页，为False截取可视窗口 - :param left_top: 截取范围左上角坐标 - :param right_bottom: 截取范围右下角角坐标 - :param ele: 为异域iframe内元素截图设置 - :return: 图片完整路径或字节文本 - """ if as_bytes: if as_bytes is True: pic_type = 'png' @@ -1132,11 +867,6 @@ class Timeout(object): """用于保存d模式timeout信息的类""" def __init__(self, base=None, page_load=None, script=None): - """ - :param base: 默认超时时间 - :param page_load: 页面加载超时时间 - :param script: js超时时间 - """ self.base = 10 if base is None else base self.page_load = 30 if page_load is None else page_load self.script = 30 if script is None else script diff --git a/DrissionPage/_pages/chromium_base.pyi b/DrissionPage/_pages/chromium_base.pyi index aa2cf48..7965bf1 100644 --- a/DrissionPage/_pages/chromium_base.pyi +++ b/DrissionPage/_pages/chromium_base.pyi @@ -17,7 +17,6 @@ from .._elements.session_element import SessionElement from .._functions.cookies import CookiesList from .._functions.elements import SessionElementsList, ChromiumElementsList from .._pages.chromium_frame import ChromiumFrame -from .._pages.chromium_page import ChromiumPage from .._units.actions import Actions from .._units.console import Console from .._units.listener import Listener @@ -32,287 +31,624 @@ PIC_TYPE = Literal['jpg', 'jpeg', 'png', 'webp', True] class ChromiumBase(BasePage): + _tab: Union[ChromiumTab, MixTab, ChromiumFrame] = ... + _browser: Chromium = ... + _driver: Optional[Driver] = ... + _frame_id: str = ... + _is_reading: bool = ... + _is_timeout: bool = ... + _timeouts: Timeout = ... + _first_run: bool = ... + _is_loading: Optional[bool] = ... + _load_mode: str = ... + _scroll: Optional[Scroller] = ... + _url: str = ... + _root_id: Optional[str] = ... + _upload_list: Optional[list] = ... + _wait: Optional[BaseWaiter] = ... + _set: Optional[ChromiumBaseSetter] = ... + _screencast: Optional[Screencast] = ... + _actions: Optional[Actions] = ... + _listener: Optional[Listener] = ... + _states: Optional[PageStates] = ... + _alert: Alert = ... + _has_alert: bool = ... + _doc_got: bool = ... + _load_end_time: float = ... + _init_jss: list = ... + _ready_state: Optional[str] = ... + _rect: Optional[TabRect] = ... + _console: Optional[Console] = ... + _type: str = ... def __init__(self, browser: Chromium, tab_id: str = None): - self._tab: Union[ChromiumTab, MixTab, ChromiumFrame] = ... - self._browser: Chromium = ... - self._driver: Driver = ... - self._frame_id: str = ... - self._is_reading: bool = ... - self._is_timeout: bool = ... - self._timeouts: Timeout = ... - self._first_run: bool = ... - self._is_loading: bool = ... - self._load_mode: str = ... - self._scroll: Scroller = ... - self._url: str = ... - self._root_id: str = ... - self._upload_list: list = ... - self._wait: BaseWaiter = ... - self._set: ChromiumBaseSetter = ... - self._screencast: Screencast = ... - self._actions: Actions = ... - self._listener: Listener = ... - self._states: PageStates = ... - self._alert: Alert = ... - self._has_alert: bool = ... - self._doc_got: bool = ... - self._load_end_time: float = ... - self._init_jss: list = ... - self._ready_state: Optional[str] = ... - self._rect: TabRect = ... - self._console: Console = ... - self._type: str = ... + """ + :param browser: Chromium + :param target_id: 要控制的target id，不指定默认为激活的标签页 + """ + ... - def _connect_browser(self, target_id: str = None) -> None: ... + def __call__(self, + locator: Union[Tuple[str, str], str, ChromiumElement], + index: int = 1, + timeout: float = None) -> ChromiumElement: + """在内部查找元素 + 例：ele = page('@id=ele_id') + :param locator: 元素的定位信息，可以是loc元组，或查询字符串 + :param index: 获取第几个元素，从1开始，可传入负数获取倒数第几个 + :param timeout: 超时时间（秒） + :return: ChromiumElement对象 + """ + ... - def _driver_init(self, target_id: str) -> None: ... + def _d_set_runtime_settings(self) -> None: ... - def _get_document(self, timeout: float = 10) -> bool: ... + def _connect_browser(self, target_id: str = None) -> None: + """连接浏览器，在第一次时运行 + :param target_id: 要控制的target id，不指定默认为激活的标签页 + :return: None + """ + ... - def _wait_loaded(self, timeout: float = None) -> bool: ... + def _driver_init(self, target_id: str) -> None: + """新建页面、页面刷新后要进行的cdp参数初始化 + :param target_id: 要跳转到的target id + :return: None + """ + ... + + def _get_document(self, timeout: float = 10) -> bool: + """获取页面文档 + :param timeout: 超时时间（秒） + :return: 是否获取成功 + """ + ... def _onFrameDetached(self, **kwargs) -> None: ... def _onFrameAttached(self, **kwargs) -> None: ... - def _onFrameStartedLoading(self, **kwargs): ... + def _onFrameStartedLoading(self, **kwargs): + """页面开始加载时执行""" + ... - def _onFrameNavigated(self, **kwargs): ... + def _onFrameNavigated(self, **kwargs): + """页面跳转时执行""" + ... - def _onDomContentEventFired(self, **kwargs): ... + def _onDomContentEventFired(self, **kwargs): + """在页面刷新、变化后重新读取页面内容""" + ... - def _onLoadEventFired(self, **kwargs): ... + def _onLoadEventFired(self, **kwargs): + """在页面刷新、变化后重新读取页面内容""" + ... - def _onFrameStoppedLoading(self, **kwargs): ... + def _onFrameStoppedLoading(self, **kwargs): + """页面加载完成后执行""" + ... - def _onFileChooserOpened(self, **kwargs): ... + def _onFileChooserOpened(self, **kwargs): + """文件选择框打开时执行""" + ... - def _wait_to_stop(self): ... - - # def _d_set_start_options(self, address) -> None: ... - - def _d_set_runtime_settings(self) -> None: ... - - def __call__(self, - locator: Union[Tuple[str, str], str, ChromiumElement], - index: int = 1, - timeout: float = None) -> ChromiumElement: ... + def _wait_to_stop(self): + """eager策略超时时使页面停止加载""" + ... @property - def _js_ready_state(self) -> str: ... + def wait(self) -> BaseWaiter: + """返回用于等待的对象""" + ... @property - def browser(self) -> Chromium: ... + def set(self) -> ChromiumBaseSetter: + """返回用于设置的对象""" + ... @property - def title(self) -> str: ... + def screencast(self) -> Screencast: + """返回用于录屏的对象""" + ... @property - def driver(self) -> Driver: ... + def actions(self) -> Actions: + """返回用于执行动作链的对象""" + ... @property - def url(self) -> str: ... + def listen(self) -> Listener: + """返回用于聆听数据包的对象""" + ... @property - def _browser_url(self) -> str: ... + def states(self) -> PageStates: + """返回用于获取状态信息的对象""" + ... @property - def html(self) -> str: ... + def scroll(self) -> PageScroller: + """返回用于滚动滚动条的对象""" + ... @property - def json(self) -> Union[dict, None]: ... + def rect(self) -> TabRect: + """返回获取窗口坐标和大小的对象""" + ... @property - def _target_id(self) -> str: ... + def console(self) -> Console: + """返回获取控制台信息的对象""" + ... @property - def tab_id(self) -> str: ... + def timeout(self) -> float: + """返回timeout设置""" + ... @property - def active_ele(self) -> ChromiumElement: ... + def timeouts(self) -> Timeout: + """返回timeouts设置""" + ... @property - def load_mode(self) -> str: ... + def browser(self) -> Chromium: + """返回浏览器对象""" + ... @property - def user_agent(self) -> str: ... + def driver(self) -> Driver: + """返回用于控制浏览器的Driver对象""" + ... @property - def scroll(self) -> PageScroller: ... + def title(self) -> str: + """返回当前页面title""" + ... @property - def rect(self) -> TabRect: ... + def url(self) -> str: + """返回当前页面url""" + ... @property - def console(self) -> Console: ... + def _browser_url(self) -> str: + """用于被MixTab覆盖""" + ... @property - def timeout(self) -> float: ... + def html(self) -> str: + """返回当前页面html文本""" + ... @property - def timeouts(self) -> Timeout: ... + def json(self) -> Union[dict, None]: + """当返回内容是json格式时，返回对应的字典，非json格式时返回None""" + ... @property - def upload_list(self) -> list: ... + def tab_id(self) -> str: + """返回当前标签页id""" + ... @property - def wait(self) -> BaseWaiter: ... + def _target_id(self) -> str: + """返回当前标签页id""" + ... @property - def set(self) -> ChromiumBaseSetter: ... + def active_ele(self) -> ChromiumElement: + """返回当前焦点所在元素""" + ... @property - def screencast(self) -> Screencast: ... + def load_mode(self) -> Literal['none', 'normal', 'eager']: + """返回页面加载策略，有3种：'none'、'normal'、'eager'""" + ... @property - def actions(self) -> Actions: ... + def user_agent(self) -> str: + """返回user agent""" + ... @property - def listen(self) -> Listener: ... + def upload_list(self) -> list: + """返回等待上传文件列表""" + ... @property - def states(self) -> PageStates: ... + def _js_ready_state(self) -> str: + """返回js获取的ready state信息""" + ... - def run_js(self, script: Union[str, Path], *args, as_expr: bool = False, timeout: float = None) -> Any: ... + def run_cdp(self, cmd: str, **cmd_args) -> dict: + """执行Chrome DevTools Protocol语句 + :param cmd: 协议项目 + :param cmd_args: 参数 + :return: 执行的结果 + """ + ... - def run_js_loaded(self, script: Union[str, Path], *args, as_expr: bool = False, timeout: float = None) -> Any: ... + def run_cdp_loaded(self, cmd: str, **cmd_args) -> dict: + """执行Chrome DevTools Protocol语句，执行前等待页面加载完毕 + :param cmd: 协议项目 + :param cmd_args: 参数 + :return: 执行的结果 + """ + ... - def _run_js(self, script: Union[str, Path], *args, as_expr: bool = False, timeout: float = None) -> Any: ... + def _run_cdp(self, cmd: str, **cmd_args) -> dict: + """执行Chrome DevTools Protocol语句 + :param cmd: 协议项目 + :param cmd_args: 参数 + :return: 执行的结果 + """ + ... - def _run_js_loaded(self, script: Union[str, Path], *args, as_expr: bool = False, timeout: float = None) -> Any: ... + def _run_cdp_loaded(self, cmd: str, **cmd_args) -> dict: + """执行Chrome DevTools Protocol语句，执行前等待页面加载完毕 + :param cmd: 协议项目 + :param cmd_args: 参数 + :return: 执行的结果 + """ + ... - def run_async_js(self, script: Union[str, Path], *args, as_expr: bool = False) -> None: ... + def run_js(self, script: Union[str, Path], *args, as_expr: bool = False, timeout: float = None) -> Any: + """运行javascript代码 + :param script: js文本或js文件路径 + :param args: 参数，按顺序在js文本中对应arguments[0]、arguments[1]... + :param as_expr: 是否作为表达式运行，为True时args无效 + :param timeout: js超时时间（秒），为None则使用页面timeouts.script设置 + :return: 运行的结果 + """ + ... + + def run_js_loaded(self, script: Union[str, Path], *args, as_expr: bool = False, timeout: float = None) -> Any: + """运行javascript代码，执行前等待页面加载完毕 + :param script: js文本或js文件路径 + :param args: 参数，按顺序在js文本中对应arguments[0]、arguments[1]... + :param as_expr: 是否作为表达式运行，为True时args无效 + :param timeout: js超时时间（秒），为None则使用页面timeouts.script属性值 + :return: 运行的结果 + """ + ... + + def _run_js(self, script: Union[str, Path], *args, as_expr: bool = False, timeout: float = None) -> Any: + """运行javascript代码 + :param script: js文本或js文件路径 + :param args: 参数，按顺序在js文本中对应arguments[0]、arguments[1]... + :param as_expr: 是否作为表达式运行，为True时args无效 + :param timeout: js超时时间（秒），为None则使用页面timeouts.script设置 + :return: 运行的结果 + """ + ... + + def _run_js_loaded(self, script: Union[str, Path], *args, as_expr: bool = False, timeout: float = None) -> Any: + """运行javascript代码，执行前等待页面加载完毕 + :param script: js文本或js文件路径 + :param args: 参数，按顺序在js文本中对应arguments[0]、arguments[1]... + :param as_expr: 是否作为表达式运行，为True时args无效 + :param timeout: js超时时间（秒），为None则使用页面timeouts.script属性值 + :return: 运行的结果 + """ + ... + + def run_async_js(self, script: Union[str, Path], *args, as_expr: bool = False) -> None: + """以异步方式执行js代码或js文件路径 + :param script: js文本 + :param args: 参数，按顺序在js文本中对应arguments[0]、arguments[1]... + :param as_expr: 是否作为表达式运行，为True时args无效 + :return: None + """ + ... def get(self, url: str, show_errmsg: bool = False, retry: int = None, - interval: float = None, timeout: float = None) -> Union[None, bool]: ... + interval: float = None, timeout: float = None) -> Union[None, bool]: + """访问url + :param url: 目标url + :param show_errmsg: 是否显示和抛出异常 + :param retry: 重试次数，为None时使用页面对象retry_times属性值 + :param interval: 重试间隔（秒），为None时使用页面对象retry_interval属性值 + :param timeout: 连接超时时间（秒），为None时使用页面对象timeouts.page_load属性值 + :return: 目标url是否可用 + """ + ... - def cookies(self, all_domains: bool = False, all_info: bool = False) -> CookiesList: ... + def cookies(self, all_domains: bool = False, all_info: bool = False) -> CookiesList: + """返回cookies信息 + :param all_domains: 是否返回所有域的cookies + :param all_info: 是否返回所有信息，为False时只返回name、value、domain + :return: cookies信息 + """ + ... def ele(self, locator: Union[Tuple[str, str], str, ChromiumElement, ChromiumFrame], index: int = 1, - timeout: float = None) -> ChromiumElement: ... + timeout: float = None) -> ChromiumElement: + """获取一个符合条件的元素对象 + :param locator: 定位符或元素对象 + :param index: 获取第几个元素，从1开始，可传入负数获取倒数第几个 + :param timeout: 查找超时时间（秒），默认与页面等待时间一致 + :return: ChromiumElement对象 + """ + ... def eles(self, locator: Union[Tuple[str, str], str], - timeout: float = None) -> ChromiumElementsList: ... + timeout: float = None) -> ChromiumElementsList: + """获取所有符合条件的元素对象 + :param locator: 定位符或元素对象 + :param timeout: 查找超时时间（秒），默认与页面等待时间一致 + :return: ChromiumElement对象组成的列表 + """ + ... def s_ele(self, locator: Union[Tuple[str, str], str] = None, index: int = 1, - timeout: float = None) -> SessionElement: ... + timeout: float = None) -> SessionElement: + """查找一个符合条件的元素以SessionElement形式返回，处理复杂页面时效率很高 + :param locator: 元素的定位信息，可以是loc元组，或查询字符串 + :param index: 获取第几个，从1开始，可传入负数获取倒数第几个 + :param timeout: 查找元素超时时间（秒），默认与页面等待时间一致 + :return: SessionElement对象或属性、文本 + """ + ... def s_eles(self, locator: Union[Tuple[str, str], str], - timeout: float = None) -> SessionElementsList: ... + timeout: float = None) -> SessionElementsList: + """查找所有符合条件的元素以SessionElement列表形式返回 + :param locator: 元素的定位信息，可以是loc元组，或查询字符串 + :param timeout: 查找元素超时时间（秒），默认与页面等待时间一致 + :return: SessionElement对象组成的列表 + """ + ... def _find_elements(self, locator: Union[Tuple[str, str], str, ChromiumElement, ChromiumFrame], timeout: float = None, index: Optional[int] = 1, relative: bool = False, - raise_err: bool = None) -> Union[ChromiumElement, ChromiumFrame, ChromiumElementsList]: ... + raise_err: bool = None) -> Union[ChromiumElement, ChromiumFrame, ChromiumElementsList]: + """执行元素查找 + :param locator: 定位符或元素对象 + :param timeout: 查找超时时间（秒） + :param index: 第几个结果，从1开始，可传入负数获取倒数第几个，为None返回所有 + :param relative: MixTab用的表示是否相对定位的参数 + :param raise_err: 找不到元素是是否抛出异常，为None时根据全局设置 + :return: ChromiumElement对象或元素对象组成的列表 + """ + ... - def refresh(self, ignore_cache: bool = False) -> None: ... + def refresh(self, ignore_cache: bool = False) -> None: + """刷新当前页面 + :param ignore_cache: 是否忽略缓存 + :return: None + """ + ... - def forward(self, steps: int = 1) -> None: ... + def forward(self, steps: int = 1) -> None: + """在浏览历史中前进若干步 + :param steps: 前进步数 + :return: None + """ + ... - def back(self, steps: int = 1) -> None: ... + def back(self, steps: int = 1) -> None: + """在浏览历史中后退若干步 + :param steps: 后退步数 + :return: None + """ + ... - def _forward_or_back(self, steps: int) -> None: ... + def _forward_or_back(self, steps: int) -> None: + """执行浏览器前进或后退，会跳过url相同的历史记录 + :param steps: 步数 + :return: None + """ + ... - def stop_loading(self) -> None: ... + def stop_loading(self) -> None: + """页面停止加载""" + ... - def remove_ele(self, loc_or_ele: Union[ChromiumElement, ChromiumFrame, str, Tuple[str, str]]) -> None: ... + def remove_ele(self, loc_or_ele: Union[ChromiumElement, ChromiumFrame, str, Tuple[str, str]]) -> None: + """从页面上删除一个元素 + :param loc_or_ele: 元素对象或定位符 + :return: None + """ + ... def add_ele(self, html_or_info: Union[str, Tuple[str, dict]], insert_to: Union[ChromiumElement, str, Tuple[str, str], None] = None, - before: Union[ChromiumElement, str, Tuple[str, str], None] = None) -> ChromiumElement: ... + before: Union[ChromiumElement, str, Tuple[str, str], None] = None) -> ChromiumElement: + """新建一个元素 + :param html_or_info: 新元素的html文本或信息。信息格式为：(tag, {attr1: value, ...}) + :param insert_to: 插入到哪个元素中，可接收元素对象和定位符，为None且为html添加到body，不为html不插入 + :param before: 在哪个子节点前面插入，可接收对象和定位符，为None插入到父元素末尾 + :return: 元素对象 + """ + ... def get_frame(self, loc_ind_ele: Union[str, int, tuple, ChromiumFrame, ChromiumElement], - timeout: float = None) -> ChromiumFrame: ... + timeout: float = None) -> ChromiumFrame: + """获取页面中一个frame对象 + :param loc_ind_ele: 定位符、iframe序号、ChromiumFrame对象，序号从1开始，可传入负数获取倒数第几个 + :param timeout: 查找元素超时时间（秒） + :return: ChromiumFrame对象 + """ + ... - def get_frames(self, locator: Union[str, tuple] = None, timeout: float = None) -> List[ChromiumFrame]: ... + def get_frames(self, locator: Union[str, tuple] = None, timeout: float = None) -> List[ChromiumFrame]: + """获取所有符合条件的frame对象 + :param locator: 定位符，为None时返回所有 + :param timeout: 查找超时时间（秒） + :return: ChromiumFrame对象组成的列表 + """ + ... - def run_cdp(self, cmd: str, **cmd_args) -> dict: ... + def session_storage(self, item: str = None) -> Union[str, dict, None]: + """返回sessionStorage信息，不设置item则获取全部 + :param item: 要获取的项，不设置则返回全部 + :return: sessionStorage一个或所有项内容 + """ + ... - def run_cdp_loaded(self, cmd: str, **cmd_args) -> dict: ... - - def _run_cdp(self, cmd: str, **cmd_args) -> dict: ... - - def _run_cdp_loaded(self, cmd: str, **cmd_args) -> dict: ... - - def session_storage(self, item: str = None) -> Union[str, dict, None]: ... - - def local_storage(self, item: str = None) -> Union[str, dict, None]: ... - - def add_init_js(self, script: str) -> str: ... - - def remove_init_js(self, script_id: str = None) -> None: ... + def local_storage(self, item: str = None) -> Union[str, dict, None]: + """返回localStorage信息，不设置item则获取全部 + :param item: 要获取的项目，不设置则返回全部 + :return: localStorage一个或所有项内容 + """ + ... def get_screenshot(self, path: [str, Path] = None, name: str = None, as_bytes: PIC_TYPE = None, as_base64: PIC_TYPE = None, full_page: bool = False, left_top: Tuple[int, int] = None, - right_bottom: Tuple[int, int] = None) -> Union[str, bytes]: ... + right_bottom: Tuple[int, int] = None) -> Union[str, bytes]: + """对页面进行截图，可对整个网页、可见网页、指定范围截图。对可视范围外截图需要90以上版本浏览器支持 + :param path: 保存路径 + :param name: 完整文件名，后缀可选 'jpg','jpeg','png','webp' + :param as_bytes: 是否以字节形式返回图片，可选 'jpg','jpeg','png','webp'，生效时path参数和as_base64参数无效 + :param as_base64: 是否以base64字符串形式返回图片，可选 'jpg','jpeg','png','webp'，生效时path参数无效 + :param full_page: 是否整页截图，为True截取整个网页，为False截取可视窗口 + :param left_top: 截取范围左上角坐标 + :param right_bottom: 截取范围右下角角坐标 + :return: 图片完整路径或字节文本 + """ + ... + + def add_init_js(self, script: str) -> str: + """添加初始化脚本，在页面加载任何脚本前执行 + :param script: js文本 + :return: 添加的脚本的id + """ + ... + + def remove_init_js(self, script_id: str = None) -> None: + """删除初始化脚本，js_id传入None时删除所有 + :param script_id: 脚本的id + :return: None + """ + ... + + def clear_cache(self, session_storage: bool = True, local_storage: bool = True, cache: bool = True, + cookies: bool = True) -> None: + """清除缓存，可选要清除的项 + :param session_storage: 是否清除sessionStorage + :param local_storage: 是否清除localStorage + :param cache: 是否清除cache + :param cookies: 是否清除cookies + :return: None + """ + ... + + def disconnect(self) -> None: + """断开与页面的连接，不关闭页面""" + ... + + def reconnect(self, wait: float = 0) -> None: + """断开与页面原来的页面，重新建立连接 + :param wait: 断开后等待若干秒再连接 + :return: None + """ + ... + + def handle_alert(self, accept: Optional[bool] = True, send: str = None, timeout: float = None, + next_one: bool = False) -> Union[str, False]: + """处理提示框，可以自动等待提示框出现 + :param accept: True表示确认，False表示取消，为None不会按按钮但依然返回文本值 + :param send: 处理prompt提示框时可输入文本 + :param timeout: 等待提示框出现的超时时间（秒），为None则使用self.timeout属性的值 + :param next_one: 是否处理下一个出现的提示框，为True时timeout参数无效 + :return: 提示框内容文本，未等到提示框则返回False + """ + ... + + def _handle_alert(self, accept: bool = True, send: str = None, timeout: float = None, + next_one: bool = False) -> Union[str, False]: + """处理提示框，可以自动等待提示框出现 + :param accept: True表示确认，False表示取消，其它值不会按按钮但依然返回文本值 + :param send: 处理prompt提示框时可输入文本 + :param timeout: 等待提示框出现的超时时间（秒），为None则使用self.timeout属性的值 + :param next_one: 是否处理下一个出现的提示框，为True时timeout参数无效 + :return: 提示框内容文本，未等到提示框则返回False + """ + ... + + def _on_alert_open(self, **kwargs): + """alert出现时触发的方法""" + ... + + def _on_alert_close(self, **kwargs): + """alert关闭时触发的方法""" + ... + + def _wait_loaded(self, timeout: float = None) -> bool: + """等待页面加载完成，超时触发停止加载 + :param timeout: 超时时间（秒） + :return: 是否成功，超时返回False + """ + ... + + def _d_connect(self, to_url: str, times: int = 0, interval: float = 1, show_errmsg: bool = False, + timeout: float = None) -> Union[bool, None]: + """尝试连接，重试若干次 + :param to_url: 要访问的url + :param times: 重试次数 + :param interval: 重试间隔（秒） + :param show_errmsg: 是否抛出异常 + :param timeout: 连接超时时间（秒） + :return: 是否成功，返回None表示不确定 + """ + ... def _get_screenshot(self, path: [str, Path] = None, name: str = None, as_bytes: PIC_TYPE = None, as_base64: PIC_TYPE = None, full_page: bool = False, left_top: Tuple[float, float] = None, - right_bottom: Tuple[float, float] = None, ele: ChromiumElement = None) -> Union[str, bytes]: ... - - def clear_cache(self, session_storage: bool = True, local_storage: bool = True, cache: bool = True, - cookies: bool = True) -> None: ... - - def disconnect(self) -> None: ... - - def reconnect(self, wait: float = 0) -> None: ... - - def handle_alert(self, accept: Optional[bool] = True, send: str = None, timeout: float = None, - next_one: bool = False) -> Union[str, False]: ... - - def _handle_alert(self, accept: bool = True, send: str = None, timeout: float = None, - next_one: bool = False) -> Union[str, False]: ... - - def _on_alert_close(self, **kwargs): ... - - def _on_alert_open(self, **kwargs): ... - - def _d_connect(self, to_url: str, times: int = 0, interval: float = 1, show_errmsg: bool = False, - timeout: float = None) -> Union[bool, None]: ... + right_bottom: Tuple[float, float] = None, ele: ChromiumElement = None) -> Union[str, bytes]: + """对页面进行截图，可对整个网页、可见网页、指定范围截图。对可视范围外截图需要90以上版本浏览器支持 + :param path: 保存路径 + :param name: 完整文件名，后缀可选 'jpg','jpeg','png','webp' + :param as_bytes: 是否以字节形式返回图片，可选 'jpg','jpeg','png','webp'，生效时path参数和as_base64参数无效 + :param as_base64: 是否以base64字符串形式返回图片，可选 'jpg','jpeg','png','webp'，生效时path参数无效 + :param full_page: 是否整页截图，为True截取整个网页，为False截取可视窗口 + :param left_top: 截取范围左上角坐标 + :param right_bottom: 截取范围右下角角坐标 + :param ele: 为异域iframe内元素截图设置 + :return: 图片完整路径或字节文本 + """ + ... class Timeout(object): + base: float = ... + page_load: float = ... + script: float = ... def __init__(self, base=None, page_load=None, script=None): - self.base: float = ... - self.page_load: float = ... - self.script: float = ... + """ + :param base: 默认超时时间 + :param page_load: 页面加载超时时间 + :param script: js超时时间 + """ + ... @property - def as_dict(self) -> dict: ... + def as_dict(self) -> dict: + """以dict格式返回timeout设置""" + ... class Alert(object): - - def __init__(self): - self.activated: bool = ... - self.text: str = ... - self.type: str = ... - self.defaultPrompt: str = ... - self.response_accept: str = ... - self.response_text: str = ... - self.handle_next: Optional[bool] = ... - self.next_text: str = ... - self.auto: Optional[bool] = ... - - -def get_mhtml(page: Union[ChromiumPage, ChromiumTab], - path: Union[str, Path] = None, - name: str = None) -> str: ... - - -def get_pdf(page: Union[ChromiumPage, ChromiumTab], - path: Union[str, Path] = None, - name: str = None, kwargs: dict = None) -> bytes: ... + activated: Optional[bool] = ... + text: Optional[str] = ... + type: Optional[str] = ... + defaultPrompt: Optional[str] = ... + response_accept: Optional[str] = ... + response_text: Optional[str] = ... + handle_next: Optional[bool] = ... + next_text: Optional[str] = ... + auto: Optional[bool] = ... diff --git a/DrissionPage/_pages/mix_page.pyi b/DrissionPage/_pages/mix_page.pyi index 727a155..9979141 100644 --- a/DrissionPage/_pages/mix_page.pyi +++ b/DrissionPage/_pages/mix_page.pyi @@ -24,18 +24,19 @@ from .._units.setter import MixPageSetter class MixPage(SessionPage, ChromiumPage, BasePage): + _mode: str = ... + _set: MixPageSetter = ... + _has_driver: Optional[bool] = ... + _has_session: Optional[bool] = ... + _session_options: Union[SessionOptions, None] = ... + _chromium_options: Union[ChromiumOptions, None] = ... def __init__(self, mode: str = 'd', timeout: float = None, chromium_options: Union[ChromiumOptions, bool] = None, session_or_options: Union[Session, SessionOptions, bool] = None) -> None: - self._mode: str = ... - self._set: MixPageSetter = ... - self._has_driver: bool = ... - self._has_session: bool = ... - self._session_options: Union[SessionOptions, None] = ... - self._chromium_options: Union[ChromiumOptions, None] = ... + ... def __call__(self, locator: Union[Tuple[str, str], str, ChromiumElement, SessionElement], diff --git a/DrissionPage/_units/clicker.py b/DrissionPage/_units/clicker.py index e020278..40cddc1 100644 --- a/DrissionPage/_units/clicker.py +++ b/DrissionPage/_units/clicker.py @@ -14,28 +14,12 @@ from ..errors import CanNotClickError, CDPError, NoRectError, AlertExistsError class Clicker(object): def __init__(self, ele): - """ - :param ele: ChromiumElement - """ self._ele = ele def __call__(self, by_js=False, timeout=1.5, wait_stop=True): - """点击元素 - 如果遇到遮挡，可选择是否用js点击 - :param by_js: 是否用js点击，为None时先用模拟点击，遇到遮挡改用js，为True时直接用js点击，为False时只用模拟点击 - :param timeout: 模拟点击的超时时间（秒），等待元素可见、可用、进入视口 - :param wait_stop: 是否等待元素运动结束再执行点击 - :return: 是否点击成功 - """ return self.left(by_js, timeout, wait_stop) def left(self, by_js=False, timeout=1.5, wait_stop=True): - """点击元素，可选择是否用js点击 - :param by_js: 是否用js点击，为None时先用模拟点击，遇到遮挡改用js，为True时直接用js点击，为False时只用模拟点击 - :param timeout: 模拟点击的超时时间（秒），等待元素可见、可用、进入视口 - :param wait_stop: 是否等待元素运动结束再执行点击 - :return: 是否点击成功 - """ if self._ele.tag == 'option': if not self._ele.states.is_selected: self._ele.parent('t:select').select.by_option(self._ele) @@ -108,15 +92,10 @@ class Clicker(object): return False def right(self): - """右键单击""" self._ele.owner.scroll.to_see(self._ele) return self._click(*self._ele.rect.viewport_click_point, button='right') def middle(self, get_tab=True): - """中键单击，默认返回新出现的tab对象 - :param get_tab: 是否返回新tab对象，为False则返回None - :return: Tab对象或None - """ self._ele.owner.scroll.to_see(self._ele) curr_tid = self._ele.tab.browser.tab_ids[0] self._click(*self._ele.rect.viewport_click_point, button='middle') @@ -128,13 +107,6 @@ class Clicker(object): else self._ele.tab.browser.get_tab(tid)) def at(self, offset_x=None, offset_y=None, button='left', count=1): - """带偏移量点击本元素，相对于左上角坐标。不传入x或y值时点击元素中间点 - :param offset_x: 相对元素左上角坐标的x轴偏移量 - :param offset_y: 相对元素左上角坐标的y轴偏移量 - :param button: 点击哪个键，可选 left, middle, right, back, forward - :param count: 点击次数 - :return: None - """ self._ele.owner.scroll.to_see(self._ele) if offset_x is None and offset_y is None: w, h = self._ele.rect.size @@ -143,22 +115,9 @@ class Clicker(object): return self._click(*offset_scroll(self._ele, offset_x, offset_y), button=button, count=count) def multi(self, times=2): - """多次点击 - :param times: 默认双击 - :return: None - """ return self.at(count=times) def to_download(self, save_path=None, rename=None, suffix=None, new_tab=False, by_js=False, timeout=None): - """点击触发下载 - :param save_path: 保存路径，为None保存在原来设置的，如未设置保存到当前路径 - :param rename: 重命名文件名 - :param suffix: 指定文件后缀 - :param new_tab: 该下载是否在新tab中触发 - :param by_js: 是否用js方式点击，逻辑与click()一致 - :param timeout: 等待下载触发的超时时间，为None则使用页面对象设置 - :return: DownloadMission对象 - """ if save_path: self._ele.tab.set.download_path(save_path) elif not self._ele.tab._browser._dl_mgr._running: @@ -172,21 +131,11 @@ class Clicker(object): return obj.wait.download_begin(timeout=timeout) def to_upload(self, file_paths, by_js=False): - """触发上传文件选择框并自动填入指定路径 - :param file_paths: 文件路径，如果上传框支持多文件，可传入列表或字符串，字符串时多个文件用回车分隔 - :param by_js: 是否用js方式点击，逻辑与click()一致 - :return: None - """ self._ele.owner.set.upload_files(file_paths) self.left(by_js=by_js) self._ele.owner.wait.upload_paths_inputted() def for_new_tab(self, by_js=False, timeout=3): - """点击后等待新tab出现并返回其对象 - :param by_js: 是否使用js点击，逻辑与click()一致 - :param timeout: 等待超时时间 - :return: 新标签页对象，如果没有等到新标签页出现则抛出异常 - """ curr_tid = self._ele.tab.browser.tab_ids[0] self.left(by_js=by_js) tid = self._ele.tab.browser.wait.new_tab(timeout=timeout, curr_tab=curr_tid) @@ -196,13 +145,6 @@ class Clicker(object): else self._ele.tab.browser.get_tab(tid)) def for_url_change(self, text=None, exclude=False, by_js=False, timeout=None): - """点击并等待tab的url变成包含或不包含指定文本 - :param text: 用于识别的文本，为None等待当前url变化 - :param exclude: 是否排除，为True时当url不包含text指定文本时返回True，text为None时自动设为True - :param by_js: 是否用js点击 - :param timeout: 超时时间（秒），为None使用页面设置 - :return: 是否等待成功 - """ if text is None: exclude = True text = self._ele.tab.url @@ -210,13 +152,6 @@ class Clicker(object): return True if self._ele.tab.wait.url_change(text=text, exclude=exclude, timeout=timeout) else False def for_title_change(self, text=None, exclude=False, by_js=False, timeout=None): - """点击并等待tab的title变成包含或不包含指定文本 - :param text: 用于识别的文本，为None等待当前title变化 - :param exclude: 是否排除，为True时当title不包含text指定文本时返回True，text为None时自动设为True - :param by_js: 是否用js点击 - :param timeout: 超时时间（秒），为None使用页面设置 - :return: 是否等待成功 - """ if text is None: exclude = True text = self._ele.tab.title @@ -224,13 +159,6 @@ class Clicker(object): return True if self._ele.tab.wait.title_change(text=text, exclude=exclude, timeout=timeout) else False def _click(self, view_x, view_y, button='left', count=1): - """实施点击 - :param view_x: 视口x坐标 - :param view_y: 视口y坐标 - :param button: 'left' 'right' 'middle' 'back' 'forward' - :param count: 点击次数 - :return: None - """ self._ele.owner._run_cdp('Input.dispatchMouseEvent', type='mousePressed', x=view_x, y=view_y, button=button, clickCount=count, _ignore=AlertExistsError) self._ele.owner._run_cdp('Input.dispatchMouseEvent', type='mouseReleased', x=view_x, diff --git a/DrissionPage/_units/clicker.pyi b/DrissionPage/_units/clicker.pyi index 325b2ac..b3c181e 100644 --- a/DrissionPage/_units/clicker.pyi +++ b/DrissionPage/_units/clicker.pyi @@ -14,26 +14,66 @@ from .._pages.tabs import MixTab, ChromiumTab class Clicker(object): + _ele: ChromiumElement = ... + def __init__(self, ele: ChromiumElement): - self._ele: ChromiumElement = ... + """ + :param ele: ChromiumElement + """ + ... def __call__(self, by_js: Union[bool, str, None] = False, - timeout: float = 1.5, wait_stop: bool = True) -> Union[ChromiumElement, False]: ... + timeout: float = 1.5, wait_stop: bool = True) -> Union[ChromiumElement, False]: + """点击元素 + 如果遇到遮挡，可选择是否用js点击 + :param by_js: 是否用js点击，为None时先用模拟点击，遇到遮挡改用js，为True时直接用js点击，为False时只用模拟点击 + :param timeout: 模拟点击的超时时间（秒），等待元素可见、可用、进入视口 + :param wait_stop: 是否等待元素运动结束再执行点击 + :return: 是否点击成功 + """ + ... def left(self, by_js: Union[bool, str, None] = False, - timeout: float = 1.5, wait_stop: bool = True) -> Union[ChromiumElement, False]: ... + timeout: float = 1.5, wait_stop: bool = True) -> Union[ChromiumElement, False]: + """点击元素，可选择是否用js点击 + :param by_js: 是否用js点击，为None时先用模拟点击，遇到遮挡改用js，为True时直接用js点击，为False时只用模拟点击 + :param timeout: 模拟点击的超时时间（秒），等待元素可见、可用、进入视口 + :param wait_stop: 是否等待元素运动结束再执行点击 + :return: 是否点击成功 + """ + ... - def right(self) -> ChromiumElement: ... + def right(self) -> ChromiumElement: + """右键单击""" + ... - def middle(self, get_tab: bool = True) -> Union[ChromiumTab, MixTab, None]: ... + def middle(self, get_tab: bool = True) -> Union[ChromiumTab, MixTab, None]: + """中键单击，默认返回新出现的tab对象 + :param get_tab: 是否返回新tab对象，为False则返回None + :return: Tab对象或None + """ + ... def at(self, offset_x: float = None, offset_y: float = None, button: str = 'left', - count: int = 1) -> ChromiumElement: ... + count: int = 1) -> ChromiumElement: + """带偏移量点击本元素，相对于左上角坐标。不传入x或y值时点击元素中间点 + :param offset_x: 相对元素左上角坐标的x轴偏移量 + :param offset_y: 相对元素左上角坐标的y轴偏移量 + :param button: 点击哪个键，可选 left, middle, right, back, forward + :param count: 点击次数 + :return: None + """ + ... - def multi(self, times: int = 2) -> ChromiumElement: ... + def multi(self, times: int = 2) -> ChromiumElement: + """多次点击 + :param times: 默认双击 + :return: None + """ + ... def to_download(self, save_path: Union[str, Path] = None, @@ -41,20 +81,66 @@ class Clicker(object): suffix: str = None, new_tab: bool = False, by_js: bool = False, - timeout: float = None) -> DownloadMission: ... + timeout: float = None) -> DownloadMission: + """点击触发下载 + :param save_path: 保存路径，为None保存在原来设置的，如未设置保存到当前路径 + :param rename: 重命名文件名 + :param suffix: 指定文件后缀 + :param new_tab: 该下载是否在新tab中触发 + :param by_js: 是否用js方式点击，逻辑与click()一致 + :param timeout: 等待下载触发的超时时间，为None则使用页面对象设置 + :return: DownloadMission对象 + """ + ... - def to_upload(self, file_paths: Union[str, Path, list, tuple], by_js: bool = False) -> None: ... + def to_upload(self, file_paths: Union[str, Path, list, tuple], by_js: bool = False) -> None: + """触发上传文件选择框并自动填入指定路径 + :param file_paths: 文件路径，如果上传框支持多文件，可传入列表或字符串，字符串时多个文件用回车分隔 + :param by_js: 是否用js方式点击，逻辑与click()一致 + :return: None + """ + ... - def for_new_tab(self, by_js: bool = False, timeout: float = 3) -> Union[ChromiumTab, MixTab]: ... + def for_new_tab(self, by_js: bool = False, timeout: float = 3) -> Union[ChromiumTab, MixTab]: + """点击后等待新tab出现并返回其对象 + :param by_js: 是否使用js点击，逻辑与click()一致 + :param timeout: 等待超时时间 + :return: 新标签页对象，如果没有等到新标签页出现则抛出异常 + """ + ... def for_url_change(self, text: str = None, exclude: bool = False, - by_js: bool = False, timeout: float = None) -> bool: ... + by_js: bool = False, timeout: float = None) -> bool: + """点击并等待tab的url变成包含或不包含指定文本 + :param text: 用于识别的文本，为None等待当前url变化 + :param exclude: 是否排除，为True时当url不包含text指定文本时返回True，text为None时自动设为True + :param by_js: 是否用js点击 + :param timeout: 超时时间（秒），为None使用页面设置 + :return: 是否等待成功 + """ + ... def for_title_change(self, text: str = None, exclude: bool = False, - by_js: bool = False, timeout: float = None) -> bool: ... + by_js: bool = False, timeout: float = None) -> bool: + """点击并等待tab的title变成包含或不包含指定文本 + :param text: 用于识别的文本，为None等待当前title变化 + :param exclude: 是否排除，为True时当title不包含text指定文本时返回True，text为None时自动设为True + :param by_js: 是否用js点击 + :param timeout: 超时时间（秒），为None使用页面设置 + :return: 是否等待成功 + """ + ... def _click(self, view_x: float, view_y: float, button: str = 'left', - count: int = 1) -> ChromiumElement: ... + count: int = 1) -> ChromiumElement: + """实施点击 + :param view_x: 视口x坐标 + :param view_y: 视口y坐标 + :param button: 'left' 'right' 'middle' 'back' 'forward' + :param count: 点击次数 + :return: None + """ + ... diff --git a/DrissionPage/_units/console.py b/DrissionPage/_units/console.py index 5dba0a5..0a269b2 100644 --- a/DrissionPage/_units/console.py +++ b/DrissionPage/_units/console.py @@ -53,7 +53,6 @@ class ConsoleData(object): self._data = data def __getattr__(self, item): - """获取属性""" return self._data.get(item, None) def __repr__(self): diff --git a/DrissionPage/_units/console.pyi b/DrissionPage/_units/console.pyi index 56a82bb..39966de 100644 --- a/DrissionPage/_units/console.pyi +++ b/DrissionPage/_units/console.pyi @@ -10,10 +10,16 @@ class Console(object): owner: ChromiumBase = ... _caught: Optional[Queue] = ... - def __init__(self, owner: ChromiumBase) -> None: ... + def __init__(self, owner: ChromiumBase) -> None: + """ + :param owner: 页面对象 + """ + ... @property - def messages(self) -> List[ConsoleData]: ... + def messages(self) -> List[ConsoleData]: + """以list方式返回获取到的信息，返回后会清空列表""" + ... def start(self) -> None: """开启console监听""" diff --git a/DrissionPage/_units/cookies_setter.py b/DrissionPage/_units/cookies_setter.py index dd9764c..38f64c1 100644 --- a/DrissionPage/_units/cookies_setter.py +++ b/DrissionPage/_units/cookies_setter.py @@ -10,44 +10,21 @@ from .._functions.cookies import set_tab_cookies, set_session_cookies, set_brows class BrowserCookiesSetter(object): def __init__(self, owner): - """ - :param owner: Chromium对象 - """ self._owner = owner def __call__(self, cookies): - """设置一个或多个cookie - :param cookies: cookies信息 - :return: None - """ set_browser_cookies(self._owner, cookies) def clear(self): - """清除cookies""" self._owner._run_cdp('Storage.clearCookies') class CookiesSetter(BrowserCookiesSetter): def __call__(self, cookies): - """设置一个或多个cookie - :param cookies: cookies信息 - :return: None - """ set_tab_cookies(self._owner, cookies) - def clear(self): - """清除cookies""" - self._owner._run_cdp('Network.clearBrowserCookies') - def remove(self, name, url=None, domain=None, path=None): - """删除一个cookie - :param name: cookie的name字段 - :param url: cookie的url字段，可选 - :param domain: cookie的domain字段，可选 - :param path: cookie的path字段，可选 - :return: None - """ d = {'name': name} if url is not None: d['url'] = url @@ -61,50 +38,33 @@ class CookiesSetter(BrowserCookiesSetter): d['path'] = path self._owner._run_cdp('Network.deleteCookies', **d) + def clear(self): + self._owner._run_cdp('Network.clearBrowserCookies') + class SessionCookiesSetter(object): def __init__(self, owner): self._owner = owner def __call__(self, cookies): - """设置多个cookie，注意不要传入单个 - :param cookies: cookies信息 - :return: None - """ set_session_cookies(self._owner.session, cookies) def remove(self, name): - """删除一个cookie - :param name: cookie的name字段 - :return: None - """ self._owner.session.cookies.set(name, None) def clear(self): - """清除cookies""" self._owner.session.cookies.clear() class MixPageCookiesSetter(CookiesSetter, SessionCookiesSetter): def __call__(self, cookies): - """设置多个cookie，注意不要传入单个 - :param cookies: cookies信息 - :return: None - """ if self._owner.mode == 'd' and self._owner._has_driver: super().__call__(cookies) elif self._owner.mode == 's' and self._owner._has_session: super(CookiesSetter, self).__call__(cookies) def remove(self, name, url=None, domain=None, path=None): - """删除一个cookie - :param name: cookie的name字段 - :param url: cookie的url字段，可选，d模式时才有效 - :param domain: cookie的domain字段，可选，d模式时才有效 - :param path: cookie的path字段，可选，d模式时才有效 - :return: None - """ if self._owner.mode == 'd' and self._owner._has_driver: super().remove(name, url, domain, path) elif self._owner.mode == 's' and self._owner._has_session: @@ -113,7 +73,6 @@ class MixPageCookiesSetter(CookiesSetter, SessionCookiesSetter): super(CookiesSetter, self).remove(name) def clear(self): - """清除cookies""" if self._owner.mode == 'd' and self._owner._has_driver: super().clear() elif self._owner.mode == 's' and self._owner._has_session: diff --git a/DrissionPage/_units/cookies_setter.pyi b/DrissionPage/_units/cookies_setter.pyi index 8ce3d58..a3b8560 100644 --- a/DrissionPage/_units/cookies_setter.pyi +++ b/DrissionPage/_units/cookies_setter.pyi @@ -18,42 +18,117 @@ from .._pages.mix_page import MixPage class BrowserCookiesSetter(object): _owner: Chromium = ... - def __init__(self, page: Chromium): ... + def __init__(self, owner: Chromium): + """ + :param owner: Chromium对象 + """ + ... - def __call__(self, cookies: Union[CookieJar, Cookie, list, tuple, str, dict]) -> None: ... + def __call__(self, cookies: Union[CookieJar, Cookie, list, tuple, str, dict]) -> None: + """设置一个或多个cookie + :param cookies: cookies信息 + :return: None + """ + ... - def clear(self) -> None: ... + def clear(self) -> None: + """清除cookies""" + ... class CookiesSetter(BrowserCookiesSetter): _owner: ChromiumBase = ... - def __init__(self, page: ChromiumBase): ... + def __init__(self, owner: ChromiumBase): + """ + :param owner: 页面对象 + """ + ... - def remove(self, name: str, url: str = None, domain: str = None, path: str = None) -> None: ... + def __call__(self, cookies: Union[CookieJar, Cookie, list, tuple, str, dict]) -> None: + """设置一个或多个cookie + :param cookies: cookies信息 + :return: None + """ + ... - def clear(self) -> None: ... + def remove(self, + name: str, + url: str = None, + domain: str = None, + path: str = None) -> None: + """删除一个cookie + :param name: cookie的name字段 + :param url: cookie的url字段，可选 + :param domain: cookie的domain字段，可选 + :param path: cookie的path字段，可选 + :return: None + """ + ... + + def clear(self) -> None: + """清除cookies""" + ... class SessionCookiesSetter(object): _owner: SessionPage = ... - def __init__(self, page: SessionPage): ... + def __init__(self, owner: SessionPage): + """ + :param owner: SessionPage对象 + """ + ... - def __call__(self, cookies: Union[CookieJar, Cookie, list, tuple, str, dict]) -> None: ... + def __call__(self, cookies: Union[CookieJar, Cookie, list, tuple, str, dict]) -> None: + """设置一个或多个cookie + :param cookies: cookies信息 + :return: None + """ + ... - def remove(self, name: str) -> None: ... + def remove(self, name: str) -> None: + """删除一个cookie + :param name: cookie的name字段 + :return: None + """ + ... - def clear(self) -> None: ... + def clear(self) -> None: + """清除cookies""" + ... class MixPageCookiesSetter(CookiesSetter, SessionCookiesSetter): _owner: Union[MixPage, MixTab] = ... - def __init__(self, page: SessionPage): ... + def __init__(self, owner: Union[MixPage, MixTab]): + """ + :param owner: MixPage, MixTab对象 + """ + ... - def __call__(self, cookies: Union[CookieJar, Cookie, list, tuple, str, dict]) -> None: ... + def __call__(self, cookies: Union[CookieJar, Cookie, list, tuple, str, dict]) -> None: + """设置一个或多个cookie + :param cookies: cookies信息 + :return: None + """ + ... - def remove(self, name: str, url: str = None, domain: str = None, path: str = None) -> None: ... + def remove(self, + name: str, + url: str = None, + domain: str = None, + path: str = None) -> None: + """删除一个cookie + :param name: cookie的name字段 + :param url: cookie的url字段，可选，d模式时才有效 + :param domain: cookie的domain字段，可选，d模式时才有效 + :param path: cookie的path字段，可选，d模式时才有效 + :return: None + """ + ... - def clear(self) -> None: ... + def clear(self) -> None: + """清除cookies""" + ... diff --git a/DrissionPage/_units/downloader.py b/DrissionPage/_units/downloader.py index d220f3c..6e9db72 100644 --- a/DrissionPage/_units/downloader.py +++ b/DrissionPage/_units/downloader.py @@ -16,9 +16,6 @@ from DataRecorder.tools import get_usable_path class DownloadManager(object): def __init__(self, browser): - """ - :param browser: Browser对象 - """ self._browser = browser t = TabDownloadSettings('browser') @@ -35,15 +32,9 @@ class DownloadManager(object): @property def missions(self): - """返回所有未完成的下载任务""" return self._missions def set_path(self, tab, path): - """设置某个tab的下载路径 - :param tab: 页面对象 - :param path: 下载路径（绝对路径str） - :return: None - """ tid = tab if isinstance(tab, str) else tab.tab_id TabDownloadSettings(tid).path = path if not self._running or tid == 'browser': @@ -56,53 +47,23 @@ class DownloadManager(object): self._running = True def set_rename(self, tab_id, rename=None, suffix=None): - """设置某个tab的重命名文件名 - :param tab_id: tab id - :param rename: 文件名，可不含后缀，会自动使用远程文件后缀 - :param suffix: 后缀名，显式设置后缀名，不使用远程文件后缀 - :return: None - """ ts = TabDownloadSettings(tab_id) ts.rename = rename ts.suffix = suffix def set_file_exists(self, tab_id, mode): - """设置某个tab下载文件重名时执行的策略 - :param tab_id: tab id - :param mode: 下载路径 - :return: None - """ TabDownloadSettings(tab_id).when_file_exists = mode def set_flag(self, tab_id, flag): - """设置某个tab的重命名文件名 - :param tab_id: tab id - :param flag: 等待标志 - :return: None - """ self._flags[tab_id] = flag def get_flag(self, tab_id): - """获取tab下载等待标记 - :param tab_id: tab id - :return: 任务对象或False - """ return self._flags.get(tab_id, None) def get_tab_missions(self, tab_id): - """获取某个tab正在下载的任务 - :param tab_id: - :return: 下载任务组成的列表 - """ return self._tab_missions.get(tab_id, []) def set_done(self, mission, state, final_path=None): - """设置任务结束 - :param mission: 任务对象 - :param state: 任务状态 - :param final_path: 最终路径 - :return: None - """ if mission.state not in ('canceled', 'skipped'): mission.state = state mission.final_path = final_path @@ -112,10 +73,6 @@ class DownloadManager(object): mission._is_done = True def cancel(self, mission): - """取消任务 - :param mission: 任务对象 - :return: None - """ mission.state = 'canceled' try: self._browser._run_cdp('Browser.cancelDownload', guid=mission.id) @@ -125,10 +82,6 @@ class DownloadManager(object): Path(mission.final_path).unlink(True) def skip(self, mission): - """跳过任务 - :param mission: 任务对象 - :return: None - """ mission.state = 'skipped' try: self._browser._run_cdp('Browser.cancelDownload', guid=mission.id) @@ -136,17 +89,11 @@ class DownloadManager(object): pass def clear_tab_info(self, tab_id): - """当tab关闭时清除有关信息 - :param tab_id: 标签页id - :return: None - """ self._tab_missions.pop(tab_id, None) self._flags.pop(tab_id, None) TabDownloadSettings.TABS.pop(tab_id, None) def _onDownloadWillBegin(self, **kwargs): - """用于获取弹出新标签页触发的下载任务""" - # print(kwargs) guid = kwargs['guid'] tab_id = self._browser._frames.get(kwargs['frameId'], 'browser') @@ -196,7 +143,6 @@ class DownloadManager(object): self._flags[tab_id] = m def _onDownloadProgress(self, **kwargs): - """下载状态变化时执行""" if kwargs['guid'] in self._missions: mission = self._missions[kwargs['guid']] if kwargs['state'] == 'inProgress': @@ -241,9 +187,6 @@ class TabDownloadSettings(object): return object.__new__(cls) def __init__(self, tab_id): - """ - :param tab_id: tab id - """ if hasattr(self, '_created'): return self._created = True @@ -258,15 +201,6 @@ class TabDownloadSettings(object): class DownloadMission(object): def __init__(self, mgr, tab_id, _id, path, name, url, save_path): - """ - :param mgr: BrowserDownloadManager对象 - :param tab_id: 标签页id - :param _id: 任务id - :param path: 保存路径 - :param name: 文件名 - :param url: url - :param save_path: 下载路径 - """ self._mgr = mgr self.url = url self.tab_id = tab_id @@ -285,25 +219,16 @@ class DownloadMission(object): @property def rate(self): - """以百分比形式返回下载进度""" return round((self.received_bytes / self.total_bytes) * 100, 2) if self.total_bytes else None @property def is_done(self): - """返回任务是否在运行中""" return self._is_done def cancel(self): - """取消该任务，如任务已完成，删除已下载的文件""" self._mgr.cancel(self) def wait(self, show=True, timeout=None, cancel_if_timeout=True): - """等待任务结束 - :param show: 是否显示下载信息 - :param timeout: 超时时间（秒），为None则无限等待 - :param cancel_if_timeout: 超时时是否取消任务 - :return: 等待成功返回完整路径，否则返回False - """ if show: print(f'url：{self.url}') end_time = perf_counter() diff --git a/DrissionPage/_units/downloader.pyi b/DrissionPage/_units/downloader.pyi index 955611c..b719cbf 100644 --- a/DrissionPage/_units/downloader.pyi +++ b/DrissionPage/_units/downloader.pyi @@ -9,9 +9,10 @@ from typing import Dict, Optional, Union, Literal from .._base.browser import Chromium from .._pages.chromium_base import ChromiumBase -# from .._pages.chromium_page import ChromiumPage + FILE_EXISTS = Literal['skip', 'rename', 'overwrite', 's', 'r', 'o'] + class DownloadManager(object): _browser: Chromium = ... _missions: Dict[str, DownloadMission] = ... @@ -19,34 +20,107 @@ class DownloadManager(object): _flags: dict = ... _running: bool = ... - def __init__(self, browser: Chromium): ... + def __init__(self, browser: Chromium): + """ + :param browser: Browser对象 + """ + ... @property - def missions(self) -> Dict[str, DownloadMission]: ... + def missions(self) -> Dict[str, DownloadMission]: + """返回所有未完成的下载任务""" + ... - def set_path(self, tab: Union[str,ChromiumBase], path: str) -> None: ... + def set_path(self, tab: Union[str, ChromiumBase], path: str) -> None: + """设置某个tab的下载路径 + :param tab: 页面对象 + :param path: 下载路径（绝对路径str） + :return: None + """ + ... - def set_rename(self, tab_id: str, rename: str = None, suffix: str = None) -> None: ... + def set_rename(self, + tab_id: str, + rename: str = None, + suffix: str = None) -> None: + """设置某个tab的重命名文件名 + :param tab_id: tab id + :param rename: 文件名，可不含后缀，会自动使用远程文件后缀 + :param suffix: 后缀名，显式设置后缀名，不使用远程文件后缀 + :return: None + """ + ... - def set_file_exists(self, tab_id: str, mode: FILE_EXISTS) -> None: ... + def set_file_exists(self, tab_id: str, mode: FILE_EXISTS) -> None: + """设置某个tab下载文件重名时执行的策略 + :param tab_id: tab id + :param mode: 下载路径 + :return: None + """ + ... - def set_flag(self, tab_id: str, flag: Union[bool, DownloadMission, None]) -> None: ... + def set_flag(self, tab_id: str, flag: Union[bool, DownloadMission, None]) -> None: + """设置某个tab的重命名文件名 + :param tab_id: tab id + :param flag: 等待标志 + :return: None + """ + ... - def get_flag(self, tab_id: str) -> Union[bool, DownloadMission, None]: ... + def get_flag(self, tab_id: str) -> Union[bool, DownloadMission, None]: + """获取tab下载等待标记 + :param tab_id: tab id + :return: 任务对象或False + """ + ... - def get_tab_missions(self, tab_id: str) -> list: ... + def get_tab_missions(self, tab_id: str) -> list: + """获取某个tab正在下载的任务 + :param tab_id: + :return: 下载任务组成的列表 + """ + ... - def set_done(self, mission: DownloadMission, state: str, final_path: str = None) -> None: ... + def set_done(self, + mission: DownloadMission, + state: str, + final_path: str = None) -> None: + """设置任务结束 + :param mission: 任务对象 + :param state: 任务状态 + :param final_path: 最终路径 + :return: None + """ + ... - def cancel(self, mission: DownloadMission) -> None: ... + def cancel(self, mission: DownloadMission) -> None: + """取消任务 + :param mission: 任务对象 + :return: None + """ + ... - def skip(self, mission: DownloadMission) -> None: ... + def skip(self, mission: DownloadMission) -> None: + """跳过任务 + :param mission: 任务对象 + :return: None + """ + ... - def clear_tab_info(self, tab_id: str) -> None: ... + def clear_tab_info(self, tab_id: str) -> None: + """当tab关闭时清除有关信息 + :param tab_id: 标签页id + :return: None + """ + ... - def _onDownloadWillBegin(self, **kwargs) -> None: ... + def _onDownloadWillBegin(self, **kwargs) -> None: + """用于获取弹出新标签页触发的下载任务""" + ... - def _onDownloadProgress(self, **kwargs) -> None: ... + def _onDownloadProgress(self, **kwargs) -> None: + """下载状态变化时执行""" + ... class TabDownloadSettings(object): @@ -58,7 +132,11 @@ class TabDownloadSettings(object): path: Optional[str] = ... when_file_exists: str = ... - def __init__(self, tab_id: str): ... + def __init__(self, tab_id: str): + """ + :param tab_id: tab id + """ + ... class DownloadMission(object): @@ -75,15 +153,47 @@ class DownloadMission(object): save_path: str = ... _is_done: bool = ... - def __init__(self, mgr: DownloadManager, tab_id: str, _id: str, path: str, name: str, url: str, - save_path: str): ... + def __init__(self, + mgr: DownloadManager, + tab_id: str, + _id: str, + path: str, + name: str, + url: str, + save_path: str): + """ + :param mgr: BrowserDownloadManager对象 + :param tab_id: 标签页id + :param _id: 任务id + :param path: 保存路径 + :param name: 文件名 + :param url: url + :param save_path: 下载路径 + """ + ... @property - def rate(self) -> float: ... + def rate(self) -> float: + """以百分比形式返回下载进度""" + ... @property - def is_done(self) -> bool: ... + def is_done(self) -> bool: + """返回任务是否在运行中""" + ... - def cancel(self) -> None: ... + def cancel(self) -> None: + """取消该任务，如任务已完成，删除已下载的文件""" + ... - def wait(self, show: bool = True, timeout=None, cancel_if_timeout=True) -> Union[bool, str]: ... + def wait(self, + show: bool = True, + timeout=None, + cancel_if_timeout=True) -> Union[bool, str]: + """等待任务结束 + :param show: 是否显示下载信息 + :param timeout: 超时时间（秒），为None则无限等待 + :param cancel_if_timeout: 超时时是否取消任务 + :return: 等待成功返回完整路径，否则返回False + """ + ... diff --git a/DrissionPage/_units/listener.py b/DrissionPage/_units/listener.py index a12f144..9d875cd 100644 --- a/DrissionPage/_units/listener.py +++ b/DrissionPage/_units/listener.py @@ -22,9 +22,6 @@ class Listener(object): """监听器基类""" def __init__(self, owner): - """ - :param owner: ChromiumBase对象 - """ self._owner = owner self._address = owner.browser.address self._target_id = owner._target_id @@ -41,7 +38,7 @@ class Listener(object): self._targets = True self._is_regex = False - self._method = ('GET', 'POST') + self._method = {'GET', 'POST'} self._res_type = True @property @@ -50,15 +47,6 @@ class Listener(object): return self._targets def set_targets(self, targets=True, is_regex=False, method=('GET', 'POST'), res_type=True): - """指定要等待的数据包 - :param targets: 要匹配的数据包url特征，可用list等传入多个，为True时获取所有 - :param is_regex: 设置的target是否正则表达式 - :param method: 设置监听的请求类型，可指定多个，为True时监听全部 - :param res_type: 设置监听的资源类型，可指定多个，为True时监听全部，可指定的值有： - Document, Stylesheet, Image, Media, Font, Script, TextTrack, XHR, Fetch, Prefetch, EventSource, WebSocket, - Manifest, SignedExchange, Ping, CSPViolationReport, Preflight, Other - :return: None - """ if targets is not None: if not isinstance(targets, (str, list, tuple, set)) and targets is not True: raise TypeError('targets只能是str、list、tuple、set、True。') @@ -91,15 +79,6 @@ class Listener(object): raise TypeError('res_type参数只能是str、list、tuple、set、True类型。') def start(self, targets=None, is_regex=None, method=None, res_type=None): - """拦截目标请求，每次拦截前清空结果 - :param targets: 要匹配的数据包url特征，可用list等传入多个，为True时获取所有 - :param is_regex: 设置的target是否正则表达式，为None时保持原来设置 - :param method: 设置监听的请求类型，可指定多个，默认('GET', 'POST')，为True时监听全部，为None时保持原来设置 - :param res_type: 设置监听的资源类型，可指定多个，默认为True时监听全部，为None时保持原来设置，可指定的值有： - Document, Stylesheet, Image, Media, Font, Script, TextTrack, XHR, Fetch, Prefetch, EventSource, WebSocket, - Manifest, SignedExchange, Ping, CSPViolationReport, Preflight, Other - :return: None - """ if targets is not None: if is_regex is None: is_regex = False @@ -117,13 +96,6 @@ class Listener(object): self.listening = True def wait(self, count=1, timeout=None, fit_count=True, raise_err=None): - """等待符合要求的数据包到达指定数量 - :param count: 需要捕捉的数据包数量 - :param timeout: 超时时间（秒），为None无限等待 - :param fit_count: 是否必须满足总数要求，发生超时，为True返回False，为False返回已捕捉到的数据包 - :param raise_err: 超时时是否抛出错误，为None时根据Settings设置 - :return: count为1时返回数据包对象，大于1时返回列表，超时且fit_count为True时返回False - """ if not self.listening: raise RuntimeError('监听未启动或已暂停。') if not timeout: @@ -157,12 +129,6 @@ class Listener(object): return [self._caught.get_nowait() for _ in range(count)] def steps(self, count=None, timeout=None, gap=1): - """用于单步操作，可实现每收到若干个数据包执行一步操作（如翻页） - :param count: 需捕获的数据包总数，为None表示无限 - :param timeout: 每个数据包等待时间（秒），为None表示无限 - :param gap: 每接收到多少个数据包返回一次数据 - :return: 用于在接收到监听目标时触发动作的可迭代对象 - """ if not self.listening: raise RuntimeError('监听未启动或已暂停。') caught = 0 @@ -181,7 +147,6 @@ class Listener(object): sleep(.03) def stop(self): - """停止监听，清空已监听到的列表""" if self.listening: self.pause() self.clear() @@ -189,10 +154,6 @@ class Listener(object): self._driver = None def pause(self, clear=True): - """暂停监听 - :param clear: 是否清空已获取队列 - :return: None - """ if self.listening: self._driver.set_callback('Network.requestWillBeSent', None) self._driver.set_callback('Network.responseReceived', None) @@ -203,14 +164,12 @@ class Listener(object): self.clear() def resume(self): - """继续暂停的监听""" if self.listening: return self._set_callback() self.listening = True def clear(self): - """清空结果""" self._request_ids = {} self._extra_info_ids = {} self._caught = Queue(maxsize=0) @@ -218,12 +177,6 @@ class Listener(object): self._running_targets = 0 def wait_silent(self, timeout=None, targets_only=False, limit=0): - """等待所有请求结束 - :param timeout: 超时时间（秒），为None时无限等待 - :param targets_only: 是否只等待targets指定的请求结束 - :param limit: 剩下多少个连接时视为结束 - :return: 返回是否等待成功 - """ if not self.listening: raise RuntimeError('监听未启动或已暂停。') if timeout is None: @@ -242,12 +195,6 @@ class Listener(object): return False def _to_target(self, target_id, address, owner): - """切换监听的页面对象 - :param target_id: 新页面对象_target_id - :param address: 新页面对象address - :param owner: 新页面对象 - :return: None - """ self._target_id = target_id self._address = address self._owner = owner @@ -262,7 +209,6 @@ class Listener(object): self._set_callback() def _set_callback(self): - """设置监听请求的回调函数""" self._driver.set_callback('Network.requestWillBeSent', self._requestWillBeSent) self._driver.set_callback('Network.requestWillBeSentExtraInfo', self._requestWillBeSentExtraInfo) self._driver.set_callback('Network.responseReceived', self._response_received) @@ -271,7 +217,6 @@ class Listener(object): self._driver.set_callback('Network.loadingFailed', self._loading_failed) def _requestWillBeSent(self, **kwargs): - """接收到请求时的回调函数""" self._running_requests += 1 p = None if self._targets is True: @@ -300,19 +245,16 @@ class Listener(object): self._extra_info_ids.setdefault(kwargs['requestId'], {})['obj'] = p if p else False def _requestWillBeSentExtraInfo(self, **kwargs): - """接收到请求额外信息时的回调函数""" self._running_requests += 1 self._extra_info_ids.setdefault(kwargs['requestId'], {})['request'] = kwargs def _response_received(self, **kwargs): - """接收到返回信息时处理方法""" request = self._request_ids.get(kwargs['requestId'], None) if request: request._raw_response = kwargs['response'] request._resource_type = kwargs['type'] def _responseReceivedExtraInfo(self, **kwargs): - """接收到返回额外信息时的回调函数""" self._running_requests -= 1 r = self._extra_info_ids.get(kwargs['requestId'], None) if r: @@ -327,7 +269,6 @@ class Listener(object): r['response'] = kwargs def _loading_finished(self, **kwargs): - """请求完成时处理方法""" self._running_requests -= 1 rid = kwargs['requestId'] packet = self._request_ids.get(rid) @@ -363,7 +304,6 @@ class Listener(object): self._running_targets -= 1 def _loading_failed(self, **kwargs): - """请求失败时的回调方法""" self._running_requests -= 1 r_id = kwargs['requestId'] data_packet = self._request_ids.get(r_id, None) @@ -393,26 +333,19 @@ class Listener(object): class FrameListener(Listener): def _requestWillBeSent(self, **kwargs): - """接收到请求时的回调函数""" if not self._owner._is_diff_domain and kwargs.get('frameId', None) != self._owner._frame_id: return super()._requestWillBeSent(**kwargs) def _response_received(self, **kwargs): - """接收到返回信息时处理方法""" if not self._owner._is_diff_domain and kwargs.get('frameId', None) != self._owner._frame_id: return super()._response_received(**kwargs) class DataPacket(object): - """返回的数据包管理类""" def __init__(self, tab_id, target): - """ - :param tab_id: 产生这个数据包的tab的id - :param target: 监听目标 - """ self.tab_id = tab_id self.target = target self.is_failed = False @@ -479,10 +412,6 @@ class DataPacket(object): return self._fail_info def wait_extra_info(self, timeout=None): - """等待额外的信息加载完成 - :param timeout: 超时时间（秒），None为无限等待 - :return: 是否等待成功 - """ if timeout is None: while self._responseExtraInfo is None: sleep(.1) @@ -511,14 +440,12 @@ class Request(object): @property def headers(self): - """以大小写不敏感字典返回headers数据""" if self._headers is None: self._headers = CaseInsensitiveDict(self._request['headers']) return self._headers @property def postData(self): - """返回postData数据""" if self._postData is None: if self._raw_post_data: postData = self._raw_post_data @@ -534,12 +461,10 @@ class Request(object): @property def cookies(self): - """以list形式返回发送的cookies""" return [c['cookie'] for c in self.extra_info.associatedCookies if not c['blockedReasons']] @property def extra_info(self): - """返回额外数据""" return RequestExtraInfo(self._data_packet._request_extra_info or {}) @@ -557,19 +482,16 @@ class Response(object): @property def headers(self): - """以大小写不敏感字典返回headers数据""" if self._headers is None: self._headers = CaseInsensitiveDict(self._response['headers']) return self._headers @property def raw_body(self): - """返回未被处理的body文本""" return self._raw_body @property def body(self): - """返回body内容，如果是json格式，自动进行转换，如果时图片格式，进行base64转换，其它格式直接返回文本""" if self._body is None: if self._is_base64_body: self._body = b64decode(self._raw_body) @@ -593,7 +515,6 @@ class ExtraInfo(object): @property def all_info(self): - """以dict形式返回所有额外信息""" return self._extra_info def __getattr__(self, item): diff --git a/DrissionPage/_units/listener.pyi b/DrissionPage/_units/listener.pyi index afa3e04..ee59ca6 100644 --- a/DrissionPage/_units/listener.pyi +++ b/DrissionPage/_units/listener.pyi @@ -19,21 +19,26 @@ __RES_TYPE__ = Literal['Document', 'Stylesheet', 'Image', 'Media', 'Font', 'Scri class Listener(object): + _owner: ChromiumBase = ... + _address: str = ... + _target_id: str = ... + _targets: Union[str, dict, True, None] = ... + _method: Union[set, True] = ... + _res_type: Union[set, True] = ... + _caught: Optional[Queue] = ... + _is_regex: bool = ... + _driver: Optional[Driver] = ... + _request_ids: Optional[dict] = ... + _extra_info_ids: Optional[dict] = ... + _running_requests: int = ... + _running_targets: int = ... + listening: bool = ... + def __init__(self, owner: ChromiumBase): - self._owner: ChromiumBase = ... - self._address: str = ... - self._target_id: str = ... - self._targets: Union[str, dict, None] = ... - self._method: set = ... - self._res_type: set = ... - self._caught: Queue = ... - self._is_regex: bool = ... - self._driver: Driver = ... - self._request_ids: dict = ... - self._extra_info_ids: dict = ... - self.listening: bool = ... - self._running_requests: int = ... - self._running_targets: int = ... + """ + :param owner: 页面对象 + """ + ... @property def targets(self) -> Optional[set]: ... @@ -42,42 +47,101 @@ class Listener(object): targets: Union[str, list, tuple, set, bool, None] = True, is_regex: Optional[bool] = False, method: Union[str, list, tuple, set, bool, None] = ('GET', 'POST'), - res_type: Union[__RES_TYPE__, list, tuple, set, bool, None] = True) -> None: ... + res_type: Union[__RES_TYPE__, list, tuple, set, bool, None] = True) -> None: + """指定要等待的数据包 + :param targets: 要匹配的数据包url特征，可用list等传入多个，为True时获取所有 + :param is_regex: 设置的target是否正则表达式 + :param method: 设置监听的请求类型，可指定多个，为True时监听全部 + :param res_type: 设置监听的资源类型，可指定多个，为True时监听全部，可指定的值有： + Document, Stylesheet, Image, Media, Font, Script, TextTrack, XHR, Fetch, Prefetch, EventSource, WebSocket, + Manifest, SignedExchange, Ping, CSPViolationReport, Preflight, Other + :return: None + """ + ... def start(self, targets: Union[str, list, tuple, set, bool, None] = None, is_regex: Optional[bool] = None, method: Union[str, list, tuple, set, bool, None] = None, - res_type: Union[__RES_TYPE__, list, tuple, set, bool, None] = None) -> None: ... - - def stop(self) -> None: ... - - def pause(self, clear: bool = True) -> None: ... - - def resume(self) -> None: ... + res_type: Union[__RES_TYPE__, list, tuple, set, bool, None] = None) -> None: + """拦截目标请求，每次拦截前清空结果 + :param targets: 要匹配的数据包url特征，可用list等传入多个，为True时获取所有 + :param is_regex: 设置的target是否正则表达式，为None时保持原来设置 + :param method: 设置监听的请求类型，可指定多个，默认('GET', 'POST')，为True时监听全部，为None时保持原来设置 + :param res_type: 设置监听的资源类型，可指定多个，默认为True时监听全部，为None时保持原来设置，可指定的值有： + Document, Stylesheet, Image, Media, Font, Script, TextTrack, XHR, Fetch, Prefetch, EventSource, WebSocket, + Manifest, SignedExchange, Ping, CSPViolationReport, Preflight, Other + :return: None + """ + ... def wait(self, count: int = 1, timeout: float = None, fit_count: bool = True, - raise_err: bool = None) -> Union[List[DataPacket], DataPacket, None]: ... + raise_err: bool = None) -> Union[List[DataPacket], DataPacket, None]: + """等待符合要求的数据包到达指定数量 + :param count: 需要捕捉的数据包数量 + :param timeout: 超时时间（秒），为None无限等待 + :param fit_count: 是否必须满足总数要求，发生超时，为True返回False，为False返回已捕捉到的数据包 + :param raise_err: 超时时是否抛出错误，为None时根据Settings设置 + :return: count为1时返回数据包对象，大于1时返回列表，超时且fit_count为True时返回False + """ + ... def steps(self, count: int = None, timeout: float = None, - gap=1) -> Iterable[Union[DataPacket, List[DataPacket]]]: ... + gap=1) -> Iterable[Union[DataPacket, List[DataPacket]]]: + """用于单步操作，可实现每收到若干个数据包执行一步操作（如翻页） + :param count: 需捕获的数据包总数，为None表示无限 + :param timeout: 每个数据包等待时间（秒），为None表示无限 + :param gap: 每接收到多少个数据包返回一次数据 + :return: 用于在接收到监听目标时触发动作的可迭代对象 + """ + ... - @property - def results(self) -> Union[DataPacket, Dict[str, List[DataPacket]], False]: ... + def stop(self) -> None: + """停止监听，清空已监听到的列表""" + ... - def clear(self) -> None: ... + def pause(self, clear: bool = True) -> None: + """暂停监听 + :param clear: 是否清空已获取队列 + :return: None + """ + ... + + def resume(self) -> None: + """继续暂停的监听""" + ... + + def clear(self) -> None: + """清空监听到但还没返回的结果""" + ... def wait_silent(self, timeout: float = None, targets_only: bool = False, - limit: int = 0) -> bool: ... + limit: int = 0) -> bool: + """等待所有请求结束 + :param timeout: 超时时间（秒），为None时无限等待 + :param targets_only: 是否只等待targets指定的请求结束 + :param limit: 剩下多少个连接时视为结束 + :return: 返回是否等待成功 + """ + ... - def _to_target(self, target_id: str, address: str, owner: ChromiumBase) -> None: ... + def _to_target(self, target_id: str, address: str, owner: ChromiumBase) -> None: + """切换监听的页面对象 + :param target_id: 新页面对象_target_id + :param address: 新页面对象address + :param owner: 新页面对象 + :return: None + """ + ... + + def _set_callback(self) -> None: ... def _requestWillBeSent(self, **kwargs) -> None: ... @@ -91,34 +155,85 @@ class Listener(object): def _loading_failed(self, **kwargs) -> None: ... - def _set_callback(self) -> None: ... - class FrameListener(Listener): + _owner: ChromiumFrame = ... + _is_diff: bool = ... + def __init__(self, owner: ChromiumFrame): - self._owner: ChromiumFrame = ... - self._is_diff: bool = ... + """ + :param owner: ChromiumFrame对象 + """ + ... class DataPacket(object): - """返回的数据包管理类""" + """数据包类""" + + tab_id: str = ... + target: str = ... + is_failed: bool = ... + _raw_request: Optional[dict] = ... + _raw_response: Optional[dict] = ... + _raw_post_data: Optional[str] = ... + _raw_body: Optional[str] = ... + _raw_fail_info: Optional[dict] = ... + _base64_body: bool = ... + _request: Optional[Request] = ... + _response: Optional[Response] = ... + _fail_info: Optional[FailInfo] = ... + _resource_type: Optional[str] = ... + _requestExtraInfo: Optional[dict] = ... + _responseExtraInfo: Optional[dict] = ... def __init__(self, tab_id: str, target: [str, bool]): - self.tab_id: str = ... - self.target: str = ... - self.is_failed: bool = ... - self._raw_request: Optional[dict] = ... - self._raw_response: Optional[dict] = ... - self._raw_post_data: str = ... - self._raw_body: str = ... - self._raw_fail_info: Optional[dict] = ... - self._base64_body: bool = ... - self._request: Request = ... - self._response: Response = ... - self._fail_info: Optional[FailInfo] = ... - self._resource_type: str = ... - self._requestExtraInfo: Optional[dict] = ... - self._responseExtraInfo: Optional[dict] = ... + """ + :param tab_id: 产生这个数据包的tab的id + :param target: 监听目标 + """ + ... + + @property + def url(self) -> str: + """请求网址""" + ... + + @property + def method(self) -> str: + """请求类型""" + ... + + @property + def frameId(self) -> str: + """发出请求的frame id""" + ... + + @property + def resourceType(self) -> str: + """资源类型""" + ... + + @property + def request(self) -> Request: + """数据""" + ... + + @property + def response(self) -> Response: + """Response数据""" + ... + + @property + def fail_info(self) -> Optional[FailInfo]: + """请求失败数据""" + ... + + def wait_extra_info(self, timeout: float = None) -> bool: + """等待额外的信息加载完成 + :param timeout: 超时时间（秒），None为无限等待 + :return: 是否等待成功 + """ + ... @property def _request_extra_info(self) -> Optional[dict]: ... @@ -126,116 +241,142 @@ class DataPacket(object): @property def _response_extra_info(self) -> Optional[dict]: ... - @property - def url(self) -> str: ... - - @property - def method(self) -> str: ... - - @property - def frameId(self) -> str: ... - - @property - def resourceType(self) -> str: ... - - @property - def request(self) -> Request: ... - - @property - def response(self) -> Response: ... - - @property - def fail_info(self) -> Optional[FailInfo]: ... - - def wait_extra_info(self, timeout: float = None) -> bool: ... - class Request(object): + _data_packet: DataPacket = ... + _request: dict = ... + _raw_post_data: str = ... + _postData: Optional[str] = ... + url: str = ... _headers: Union[CaseInsensitiveDict, None] = ... method: str = ... - urlFragment = ... - hasPostData = ... - postDataEntries = ... - mixedContentType = ... - initialPriority = ... - referrerPolicy = ... - isLinkPreload = ... - trustTokenParams = ... - isSameSite = ... + urlFragment: str = ... + hasPostData: bool = ... + postDataEntries: List[dict] = ... + mixedContentType: Literal['blockable', 'optionally-blockable', 'none'] = ... + initialPriority: Literal['VeryLow', 'Low', 'Medium', 'High', 'VeryHigh'] = ... + referrerPolicy: Literal['unsafe-url', 'no-referrer-when-downgrade', 'no-referrer', 'origin', + 'origin-when-cross-origin', 'same-origin', 'strict-origin', 'strict-origin-when-cross-origin'] = ... + isLinkPreload: bool = ... + trustTokenParams: dict = ... + isSameSite: bool = ... - def __init__(self, data_packet: DataPacket, raw_request: dict, post_data: str): - self._data_packet: DataPacket = ... - self._request: dict = ... - self._raw_post_data: str = ... - self._postData: str = ... + def __init__(self, + data_packet: DataPacket, + raw_request: dict, + post_data: str): + """ + :param data_packet: DataPacket对象 + :param raw_request: 未处理的请求数据 + :param post_data: post发送的数据 + """ + ... @property - def headers(self) -> dict: ... + def headers(self) -> dict: + """以大小写不敏感字典返回headers数据""" + ... @property - def postData(self) -> Any: ... + def postData(self) -> Any: + """返回postData数据""" + ... @property - def cookies(self) -> List[dict]: ... + def cookies(self) -> List[dict]: + """以list形式返回发送的cookies""" + ... @property - def extra_info(self) -> Optional[RequestExtraInfo]: ... + def extra_info(self) -> Optional[RequestExtraInfo]: + """返回额外数据""" + ... class Response(object): - url = ... - status = ... - statusText = ... - headersText = ... - mimeType = ... - requestHeaders = ... - requestHeadersText = ... - connectionReused = ... + _data_packet: DataPacket = ... + _response: dict = ... + _raw_body: str = ... + _is_base64_body: bool = ... + _body: Union[str, dict, bytes, None] = ... + _headers: Union[dict, CaseInsensitiveDict, None] = ... + + url: str = ... + status: int = ... + statusText: str = ... + headersText: str = ... + mimeType: str = ... + requestHeaders: dict = ... + requestHeadersText: str = ... + connectionReused: bool = ... connectionId = ... - remoteIPAddress = ... - remotePort = ... - fromDiskCache = ... - fromServiceWorker = ... - fromPrefetchCache = ... - encodedDataLength = ... - timing = ... - serviceWorkerResponseSource = ... - responseTime = ... - cacheStorageCacheName = ... - protocol = ... - alternateProtocolUsage = ... - securityState = ... - securityDetails = ... + remoteIPAddress: str = ... + remotePort: int = ... + fromDiskCache: bool = ... + fromServiceWorker: bool = ... + fromPrefetchCache: bool = ... + fromEarlyHints: bool = ... + serviceWorkerRouterInfo: dict = ... + encodedDataLength: int = ... + timing: dict = ... + serviceWorkerResponseSource: Literal['cache-storage', 'http-cache', 'fallback-code', 'network'] = ... + responseTime: float = ... + cacheStorageCacheName: str = ... + protocol: str = ... + alternateProtocolUsage: Literal['alternativeJobWonWithoutRace', 'alternativeJobWonRace', 'mainJobWonRace', + 'mappingMissing', 'broken', 'dnsAlpnH3JobWonWithoutRace', 'dnsAlpnH3JobWonRace', 'unspecifiedReason'] = ... + securityState: Literal['unknown', 'neutral', 'insecure', 'secure', 'info', 'insecure-broken'] = ... + securityDetails: dict = ... - def __init__(self, data_packet: DataPacket, raw_response: dict, raw_body: str, base64_body: bool): - self._data_packet: DataPacket = ... - self._response: dict = ... - self._raw_body: str = ... - self._is_base64_body: bool = ... - self._body: Union[str, dict, None] = ... - self._headers: dict = ... + def __init__(self, + data_packet: DataPacket, + raw_response: dict, + raw_body: str, + base64_body: bool): + """ + :param data_packet: DataPacket对象 + :param raw_response: 未处理的response信息 + :param raw_body: 未处理的body + :param base64_body: body是否base64格式 + """ + ... @property - def extra_info(self) -> Optional[ResponseExtraInfo]: ... + def headers(self) -> CaseInsensitiveDict: + """以大小写不敏感字典返回headers数据""" + ... @property - def headers(self) -> CaseInsensitiveDict: ... + def raw_body(self) -> str: + """返回未被处理的body文本""" + ... @property - def raw_body(self) -> str: ... + def body(self) -> Any: + """返回body内容，如果是json格式，自动进行转换，如果时图片格式，进行base64转换，其它格式直接返回文本""" + ... @property - def body(self) -> Any: ... + def extra_info(self) -> Optional[ResponseExtraInfo]: + """额外信息""" + ... class ExtraInfo(object): + _extra_info: dict = ... + def __init__(self, extra_info: dict): - self._extra_info: dict = ... + """ + :param extra_info: dict格式信息 + """ + ... @property - def all_info(self) -> dict: ... + def all_info(self) -> dict: + """以dict形式返回所有额外信息""" + ... class RequestExtraInfo(ExtraInfo): @@ -261,10 +402,14 @@ class ResponseExtraInfo(ExtraInfo): class FailInfo(object): _data_packet: DataPacket _fail_info: dict - _fail_info: float errorText: str canceled: bool blockedReason: Optional[str] corsErrorStatus: Optional[str] - def __init__(self, data_packet: DataPacket, fail_info: dict): ... + def __init__(self, data_packet: DataPacket, fail_info: dict): + """ + :param data_packet: DataPacket对象 + :param fail_info: 返回的失败数据 + """ + ... diff --git a/DrissionPage/_units/rect.py b/DrissionPage/_units/rect.py index f94a94b..ad64c46 100644 --- a/DrissionPage/_units/rect.py +++ b/DrissionPage/_units/rect.py @@ -9,14 +9,10 @@ class ElementRect(object): def __init__(self, ele): - """ - :param ele: ChromiumElement - """ self._ele = ele @property def corners(self): - """返回元素四个角坐标，顺序：左上、右上、右下、左下，没有大小的元素抛出NoRectError""" vr = self._get_viewport_rect('border') r = self._ele.owner._run_cdp_loaded('Page.getLayoutMetrics')['visualViewport'] sx = r['pageX'] @@ -25,53 +21,44 @@ class ElementRect(object): @property def viewport_corners(self): - """返回元素四个角视口坐标，顺序：左上、右上、右下、左下，没有大小的元素抛出NoRectError""" r = self._get_viewport_rect('border') return (r[0], r[1]), (r[2], r[3]), (r[4], r[5]), (r[6], r[7]) @property def size(self): - """返回元素大小，格式(宽, 高)""" border = self._ele.owner._run_cdp('DOM.getBoxModel', backendNodeId=self._ele._backend_id, nodeId=self._ele._node_id, objectId=self._ele._obj_id)['model']['border'] return border[2] - border[0], border[5] - border[1] @property def location(self): - """返回元素左上角的绝对坐标""" return self._get_page_coord(*self.viewport_location) @property def midpoint(self): - """返回元素中间点的绝对坐标""" return self._get_page_coord(*self.viewport_midpoint) @property def click_point(self): - """返回元素接受点击的点的绝对坐标""" return self._get_page_coord(*self.viewport_click_point) @property def viewport_location(self): - """返回元素左上角在视口中的坐标""" m = self._get_viewport_rect('border') return m[0], m[1] @property def viewport_midpoint(self): - """返回元素中间点在视口中的坐标""" m = self._get_viewport_rect('border') return m[0] + (m[2] - m[0]) // 2, m[3] + (m[5] - m[3]) // 2 @property def viewport_click_point(self): - """返回元素接受点击的点视口坐标""" m = self._get_viewport_rect('padding') return self.viewport_midpoint[0], m[1] + 3 @property def screen_location(self): - """返回元素左上角在屏幕上坐标，左上角为(0, 0)""" vx, vy = self._ele.owner.rect.viewport_location ex, ey = self.viewport_location pr = self._ele.owner._run_js('return window.devicePixelRatio;') @@ -79,7 +66,6 @@ class ElementRect(object): @property def screen_midpoint(self): - """返回元素中点在屏幕上坐标，左上角为(0, 0)""" vx, vy = self._ele.owner.rect.viewport_location ex, ey = self.viewport_midpoint pr = self._ele.owner._run_js('return window.devicePixelRatio;') @@ -87,7 +73,6 @@ class ElementRect(object): @property def screen_click_point(self): - """返回元素中点在屏幕上坐标，左上角为(0, 0)""" vx, vy = self._ele.owner.rect.viewport_location ex, ey = self.viewport_click_point pr = self._ele.owner._run_js('return window.devicePixelRatio;') @@ -95,20 +80,14 @@ class ElementRect(object): @property def scroll_position(self): - """返回滚动条位置，格式：(x, y)""" r = self._ele._run_js('return this.scrollLeft.toString() + " " + this.scrollTop.toString();') w, h = r.split(' ') return int(w), int(h) def _get_viewport_rect(self, quad): - """按照类型返回在可视窗口中的范围 - :param quad: 方框类型，margin border padding - :return: 四个角坐标 - """ return self._ele.owner._run_cdp('DOM.getBoxModel', backendNodeId=self._ele._backend_id)['model'][quad] def _get_page_coord(self, x, y): - """根据视口坐标获取绝对坐标""" r = self._ele.owner._run_cdp_loaded('Page.getLayoutMetrics')['visualViewport'] sx = r['pageX'] sy = r['pageY'] @@ -117,19 +96,14 @@ class ElementRect(object): class TabRect(object): def __init__(self, owner): - """ - :param owner: Page对象和Tab对象 - """ self._owner = owner @property def window_state(self): - """返回窗口状态：normal、fullscreen、maximized、minimized""" return self._get_window_rect()['windowState'] @property def window_location(self): - """返回窗口在屏幕上的坐标，左上角为(0, 0)""" r = self._get_window_rect() if r['windowState'] in ('maximized', 'fullscreen'): return 0, 0 @@ -137,7 +111,6 @@ class TabRect(object): @property def window_size(self): - """返回窗口大小""" r = self._get_window_rect() if r['windowState'] == 'fullscreen': return r['width'], r['height'] @@ -148,14 +121,12 @@ class TabRect(object): @property def page_location(self): - """返回页面左上角在屏幕中坐标，左上角为(0, 0)""" w, h = self.viewport_location r = self._get_page_rect()['layoutViewport'] return w - r['pageX'], h - r['pageY'] @property def viewport_location(self): - """返回视口在屏幕中坐标，左上角为(0, 0)""" w_bl, h_bl = self.window_location w_bs, h_bs = self.window_size w_vs, h_vs = self.viewport_size_with_scrollbar @@ -163,35 +134,29 @@ class TabRect(object): @property def size(self): - """返回页面总宽高，格式：(宽, 高)""" r = self._get_page_rect()['contentSize'] return r['width'], r['height'] @property def viewport_size(self): - """返回视口宽高，不包括滚动条，格式：(宽, 高)""" r = self._get_page_rect()['visualViewport'] return r['clientWidth'], r['clientHeight'] @property def viewport_size_with_scrollbar(self): - """返回视口宽高，包括滚动条，格式：(宽, 高)""" r = self._owner._run_js('return window.innerWidth.toString() + " " + window.innerHeight.toString();') w, h = r.split(' ') return int(w), int(h) @property def scroll_position(self): - """返回滚动条位置，格式：(x, y)""" r = self._get_page_rect()['visualViewport'] return r['pageX'], r['pageY'] def _get_page_rect(self): - """获取页面范围信息""" return self._owner._run_cdp_loaded('Page.getLayoutMetrics') def _get_window_rect(self): - """获取窗口范围信息""" return self._owner.browser._driver.run('Browser.getWindowForTarget', targetId=self._owner.tab_id)['bounds'] @@ -199,51 +164,40 @@ class FrameRect(object): """异域iframe使用""" def __init__(self, frame): - """ - :param frame: ChromiumFrame对象 - """ self._frame = frame @property def location(self): - """返回iframe元素左上角的绝对坐标""" return self._frame.frame_ele.rect.location @property def viewport_location(self): - """返回元素在视口中坐标，左上角为(0, 0)""" return self._frame.frame_ele.rect.viewport_location @property def screen_location(self): - """返回元素左上角在屏幕上坐标，左上角为(0, 0)""" return self._frame.frame_ele.rect.screen_location @property def size(self): - """返回frame内页面尺寸，格式：(宽, 高)""" w = self._frame.doc_ele._run_js('return this.body.scrollWidth') h = self._frame.doc_ele._run_js('return this.body.scrollHeight') return w, h @property def viewport_size(self): - """返回视口宽高，格式：(宽, 高)""" return self._frame.frame_ele.rect.size @property def corners(self): - """返回元素四个角坐标，顺序：左上、右上、右下、左下""" return self._frame.frame_ele.rect.corners @property def viewport_corners(self): - """返回元素四个角视口坐标，顺序：左上、右上、右下、左下""" return self._frame.frame_ele.rect.viewport_corners @property def scroll_position(self): - """返回滚动条位置，格式：(x, y)""" r = self._frame.doc_ele._run_js('return this.documentElement.scrollLeft.toString() + " " ' '+ this.documentElement.scrollTop.toString();') w, h = r.split(' ') diff --git a/DrissionPage/_units/rect.pyi b/DrissionPage/_units/rect.pyi index abd6047..b24eaba 100644 --- a/DrissionPage/_units/rect.pyi +++ b/DrissionPage/_units/rect.pyi @@ -17,113 +17,200 @@ from .._pages.mix_page import MixPage class ElementRect(object): + _ele: ChromiumElement = ... def __init__(self, ele: ChromiumElement): - self._ele: ChromiumElement = ... + """ + :param ele: ChromiumElement对象 + """ + ... @property - def size(self) -> Tuple[float, float]: ... + def corners(self) -> Tuple[Tuple[float, float], ...]: + """返回元素四个角坐标，顺序：左上、右上、右下、左下，没有大小的元素抛出NoRectError""" + ... @property - def location(self) -> Tuple[float, float]: ... + def viewport_corners(self) -> Tuple[Tuple[float, float], ...]: + """返回元素四个角视口坐标，顺序：左上、右上、右下、左下，没有大小的元素抛出NoRectError""" + ... @property - def midpoint(self) -> Tuple[float, float]: ... + def size(self) -> Tuple[float, float]: + """返回元素大小，格式(宽, 高)""" + ... @property - def click_point(self) -> Tuple[float, float]: ... + def location(self) -> Tuple[float, float]: + """返回元素左上角的绝对坐标""" + ... @property - def viewport_location(self) -> Tuple[float, float]: ... + def midpoint(self) -> Tuple[float, float]: + """返回元素中间点的绝对坐标""" + ... @property - def viewport_midpoint(self) -> Tuple[float, float]: ... + def click_point(self) -> Tuple[float, float]: + """返回元素接受点击的点的绝对坐标""" + ... @property - def viewport_click_point(self) -> Tuple[float, float]: ... + def viewport_location(self) -> Tuple[float, float]: + """返回元素左上角在视口中的坐标""" + ... @property - def screen_location(self) -> Tuple[float, float]: ... + def viewport_midpoint(self) -> Tuple[float, float]: + """返回元素中间点在视口中的坐标""" + ... @property - def screen_midpoint(self) -> Tuple[float, float]: ... + def viewport_click_point(self) -> Tuple[float, float]: + """返回元素接受点击的点视口坐标""" + ... @property - def screen_click_point(self) -> Tuple[float, float]: ... + def screen_location(self) -> Tuple[float, float]: + """返回元素左上角在屏幕上坐标，左上角为(0, 0)""" + ... @property - def corners(self) -> Tuple[Tuple[float, float], ...]: ... + def screen_midpoint(self) -> Tuple[float, float]: + """返回元素中点在屏幕上坐标，左上角为(0, 0)""" + ... @property - def viewport_corners(self) -> Tuple[Tuple[float, float], ...]: ... + def screen_click_point(self) -> Tuple[float, float]: + """返回元素中点在屏幕上坐标，左上角为(0, 0)""" + ... @property - def scroll_position(self) -> Tuple[float, float]: ... + def scroll_position(self) -> Tuple[float, float]: + """返回滚动条位置，格式：(x, y)""" + ... - def _get_viewport_rect(self, quad: str) -> Union[list, None]: ... + def _get_viewport_rect(self, quad: str) -> Union[list, None]: + """按照类型返回在可视窗口中的范围 + :param quad: 方框类型，margin border padding + :return: 四个角坐标 + """ + ... - def _get_page_coord(self, x: float, y: float) -> Tuple[float, float]: ... + def _get_page_coord(self, x: float, y: float) -> Tuple[float, float]: + """根据视口坐标获取绝对坐标 + :param x: 视口x坐标 + :param y: 视口y坐标 + :return: 坐标元组 + """ + ... class TabRect(object): def __init__(self, owner: ChromiumBase): + """ + :param owner: Page对象和Tab对象 + """ self._owner: Union[ChromiumPage, ChromiumTab, MixPage, MixTab] = ... @property - def window_state(self) -> str: ... + def window_state(self) -> str: + """返回窗口状态：normal、fullscreen、maximized、minimized""" + ... @property - def window_location(self) -> Tuple[int, int]: ... + def window_location(self) -> Tuple[int, int]: + """返回窗口在屏幕上的坐标，左上角为(0, 0)""" + ... @property - def page_location(self) -> Tuple[int, int]: ... + def window_size(self) -> Tuple[int, int]: + """返回窗口大小""" + ... @property - def viewport_location(self) -> Tuple[int, int]: ... + def page_location(self) -> Tuple[int, int]: + """返回页面左上角在屏幕中坐标，左上角为(0, 0)""" + ... @property - def window_size(self) -> Tuple[int, int]: ... + def viewport_location(self) -> Tuple[int, int]: + """返回视口在屏幕中坐标，左上角为(0, 0)""" + ... @property - def size(self) -> Tuple[int, int]: ... + def size(self) -> Tuple[int, int]: + """返回页面总宽高，格式：(宽, 高)""" + ... @property - def viewport_size(self) -> Tuple[int, int]: ... + def viewport_size(self) -> Tuple[int, int]: + """返回视口宽高，不包括滚动条，格式：(宽, 高)""" + ... @property - def viewport_size_with_scrollbar(self) -> Tuple[int, int]: ... + def viewport_size_with_scrollbar(self) -> Tuple[int, int]: + """返回视口宽高，包括滚动条，格式：(宽, 高)""" + ... @property - def scroll_position(self) -> Tuple[int, int]: ... + def scroll_position(self) -> Tuple[int, int]: + """返回滚动条位置，格式：(x, y)""" + ... - def _get_page_rect(self) -> dict: ... + def _get_page_rect(self) -> dict: + """获取页面范围信息""" + ... - def _get_window_rect(self) -> dict: ... + def _get_window_rect(self) -> dict: + """获取窗口范围信息""" + ... class FrameRect(object): + _frame: ChromiumFrame = ... + def __init__(self, frame: ChromiumFrame): - self._frame: ChromiumFrame = ... + """ + :param frame: ChromiumFrame对象 + """ + ... @property - def location(self) -> Tuple[float, float]: ... + def location(self) -> Tuple[float, float]: + """返回iframe元素左上角的绝对坐标""" + ... @property - def viewport_location(self) -> Tuple[float, float]: ... + def viewport_location(self) -> Tuple[float, float]: + """返回元素在视口中坐标，左上角为(0, 0)""" + ... @property - def screen_location(self) -> Tuple[float, float]: ... + def screen_location(self) -> Tuple[float, float]: + """返回元素左上角在屏幕上坐标，左上角为(0, 0)""" + ... @property - def size(self) -> Tuple[float, float]: ... + def size(self) -> Tuple[float, float]: + """返回frame内页面尺寸，格式：(宽, 高)""" + ... @property - def viewport_size(self) -> Tuple[float, float]: ... + def viewport_size(self) -> Tuple[float, float]: + """返回视口宽高，格式：(宽, 高)""" + ... @property - def corners(self) -> Tuple[Tuple[float, float], ...]: ... + def corners(self) -> Tuple[Tuple[float, float], ...]: + """返回元素四个角坐标，顺序：左上、右上、右下、左下""" + ... @property - def viewport_corners(self) -> Tuple[Tuple[float, float], ...]: ... + def viewport_corners(self) -> Tuple[Tuple[float, float], ...]: + """返回元素四个角视口坐标，顺序：左上、右上、右下、左下""" + ... @property - def scroll_position(self) -> Tuple[float, float]: ... + def scroll_position(self) -> Tuple[float, float]: + """返回滚动条位置，格式：(x, y)""" + ... diff --git a/DrissionPage/_units/scroller.py b/DrissionPage/_units/scroller.py index 84e6e04..9229bf9 100644 --- a/DrissionPage/_units/scroller.py +++ b/DrissionPage/_units/scroller.py @@ -25,75 +25,48 @@ class Scroller(object): self._wait_scrolled() def to_top(self): - """滚动到顶端，水平位置不变""" self._run_js('{}.scrollTo({}.scrollLeft, 0);') return self._owner def to_bottom(self): - """滚动到底端，水平位置不变""" self._run_js('{}.scrollTo({}.scrollLeft, {}.scrollHeight);') return self._owner def to_half(self): - """滚动到垂直中间位置，水平位置不变""" self._run_js('{}.scrollTo({}.scrollLeft, {}.scrollHeight/2);') return self._owner def to_rightmost(self): - """滚动到最右边，垂直位置不变""" self._run_js('{}.scrollTo({}.scrollWidth, {}.scrollTop);') return self._owner def to_leftmost(self): - """滚动到最左边，垂直位置不变""" self._run_js('{}.scrollTo(0, {}.scrollTop);') return self._owner def to_location(self, x, y): - """滚动到指定位置 - :param x: 水平距离 - :param y: 垂直距离 - :return: None - """ self._run_js(f'{{}}.scrollTo({x}, {y});') return self._owner def up(self, pixel=300): - """向上滚动若干像素，水平位置不变 - :param pixel: 滚动的像素 - :return: None - """ pixel = -pixel self._run_js(f'{{}}.scrollBy(0, {pixel});') return self._owner def down(self, pixel=300): - """向下滚动若干像素，水平位置不变 - :param pixel: 滚动的像素 - :return: None - """ self._run_js(f'{{}}.scrollBy(0, {pixel});') return self._owner def left(self, pixel=300): - """向左滚动若干像素，垂直位置不变 - :param pixel: 滚动的像素 - :return: None - """ pixel = -pixel self._run_js(f'{{}}.scrollBy({pixel}, 0);') return self._owner def right(self, pixel=300): - """向右滚动若干像素，垂直位置不变 - :param pixel: 滚动的像素 - :return: None - """ self._run_js(f'{{}}.scrollBy({pixel}, 0);') return self._owner def _wait_scrolled(self): - """等待滚动结束""" if not self._wait_complete: return @@ -118,44 +91,26 @@ class Scroller(object): class ElementScroller(Scroller): def to_see(self, center=None): - """滚动页面直到元素可见 - :param center: 是否尽量滚动到页面正中，为None时如果被遮挡，则滚动到页面正中 - :return: None - """ self._owner.owner.scroll.to_see(self._owner, center=center) return self._owner def to_center(self): - """元素尽量滚动到视口中间""" self._owner.owner.scroll.to_see(self._owner, center=True) return self._owner class PageScroller(Scroller): def __init__(self, owner): - """ - :param owner: 页面对象 - """ super().__init__(owner) self._t1 = 'window' self._t2 = 'document.documentElement' def to_see(self, loc_or_ele, center=None): - """滚动页面直到元素可见 - :param loc_or_ele: 元素的定位信息，可以是loc元组，或查询字符串 - :param center: 是否尽量滚动到页面正中，为None时如果被遮挡，则滚动到页面正中 - :return: None - """ ele = self._owner._ele(loc_or_ele) self._to_see(ele, center) return self._owner def _to_see(self, ele, center): - """执行滚动页面直到元素可见 - :param ele: 元素对象 - :param center: 是否尽量滚动到页面正中，为None时如果被遮挡，则滚动到页面正中 - :return: None - """ txt = 'true' if center else 'false' ele._run_js(f'this.scrollIntoViewIfNeeded({txt});') if center or (center is not False and ele.states.is_covered): @@ -173,19 +128,14 @@ class PageScroller(Scroller): class FrameScroller(PageScroller): - def __init__(self, frame): + def __init__(self, owner): """ - :param frame: ChromiumFrame对象 + :param owner: ChromiumFrame对象 """ - super().__init__(frame.doc_ele) + super().__init__(owner.doc_ele) self._t1 = self._t2 = 'this.documentElement' def to_see(self, loc_or_ele, center=None): - """滚动页面直到元素可见 - :param loc_or_ele: 元素的定位信息，可以是loc元组，或查询字符串 - :param center: 是否尽量滚动到页面正中，为None时如果被遮挡，则滚动到页面正中 - :return: None - """ ele = self._owner._ele(loc_or_ele) self._to_see(ele, center) return self._owner diff --git a/DrissionPage/_units/scroller.pyi b/DrissionPage/_units/scroller.pyi index 4aab16b..a97c4fa 100644 --- a/DrissionPage/_units/scroller.pyi +++ b/DrissionPage/_units/scroller.pyi @@ -16,116 +16,304 @@ from .._pages.tabs import ChromiumTab, MixTab class Scroller(object): + _owner: Union[ChromiumBase, ChromiumElement] = ... _t1: str = ... _t2: str = ... - _owner: Union[ChromiumBase, ChromiumElement] = ... _wait_complete: bool = ... def __init__(self, owner: Union[ChromiumBase, ChromiumElement]): ... + def to_top(self) -> None: + """滚动到顶端，水平位置不变""" + ... + + def to_bottom(self) -> None: + """滚动到底端，水平位置不变""" + ... + + def to_half(self) -> None: + """滚动到垂直中间位置，水平位置不变""" + ... + + def to_rightmost(self) -> None: + """滚动到最右边，垂直位置不变""" + ... + + def to_leftmost(self) -> None: + """滚动到最左边，垂直位置不变""" + ... + + def to_location(self, x: int, y: int) -> None: + """滚动到指定位置 + :param x: 水平距离 + :param y: 垂直距离 + :return: None + """ + ... + + def up(self, pixel: int = 300) -> None: + """向上滚动若干像素，水平位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... + + def down(self, pixel: int = 300) -> None: + """向下滚动若干像素，水平位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... + + def left(self, pixel: int = 300) -> None: + """向左滚动若干像素，垂直位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... + + def right(self, pixel: int = 300) -> None: + """向右滚动若干像素，垂直位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... + def _run_js(self, js: str): ... - def to_top(self) -> None: ... - - def to_bottom(self) -> None: ... - - def to_half(self) -> None: ... - - def to_rightmost(self) -> None: ... - - def to_leftmost(self) -> None: ... - - def to_location(self, x: int, y: int) -> None: ... - - def up(self, pixel: int = 300) -> None: ... - - def down(self, pixel: int = 300) -> None: ... - - def left(self, pixel: int = 300) -> None: ... - - def right(self, pixel: int = 300) -> None: ... - - def _wait_scrolled(self) -> None: ... + def _wait_scrolled(self) -> None: + """等待滚动结束""" + ... class ElementScroller(Scroller): + _owner: ChromiumElement = ... - def to_see(self, center: Union[bool, None] = None) -> ChromiumElement: ... + def __init__(self, owner: ChromiumElement): + """ + :param owner: 元素对象 + """ + ... - def to_center(self) -> ChromiumElement: ... + def to_see(self, center: Union[bool, None] = None) -> ChromiumElement: + """滚动页面直到元素可见 + :param center: 是否尽量滚动到页面正中，为None时如果被遮挡，则滚动到页面正中 + :return: None + """ + ... - def to_top(self) -> ChromiumElement: ... + def to_center(self) -> ChromiumElement: + """元素尽量滚动到视口中间""" + ... - def to_bottom(self) -> ChromiumElement: ... + def to_top(self) -> ChromiumElement: + """滚动到顶端，水平位置不变""" + ... - def to_half(self) -> ChromiumElement: ... + def to_bottom(self) -> ChromiumElement: + """滚动到底端，水平位置不变""" + ... - def to_rightmost(self) -> ChromiumElement: ... + def to_half(self) -> ChromiumElement: + """滚动到垂直中间位置，水平位置不变""" + ... - def to_leftmost(self) -> ChromiumElement: ... + def to_rightmost(self) -> ChromiumElement: + """滚动到最右边，垂直位置不变""" + ... - def to_location(self, x: int, y: int) -> ChromiumElement: ... + def to_leftmost(self) -> ChromiumElement: + """滚动到最左边，垂直位置不变""" + ... - def up(self, pixel: int = 300) -> ChromiumElement: ... + def to_location(self, x: int, y: int) -> ChromiumElement: + """滚动到指定位置 + :param x: 水平距离 + :param y: 垂直距离 + :return: None + """ + ... - def down(self, pixel: int = 300) -> ChromiumElement: ... + def up(self, pixel: int = 300) -> ChromiumElement: + """向上滚动若干像素，水平位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... - def left(self, pixel: int = 300) -> ChromiumElement: ... + def down(self, pixel: int = 300) -> ChromiumElement: + """向下滚动若干像素，水平位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... - def right(self, pixel: int = 300) -> ChromiumElement: ... + def left(self, pixel: int = 300) -> ChromiumElement: + """向左滚动若干像素，垂直位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... + + def right(self, pixel: int = 300) -> ChromiumElement: + """向右滚动若干像素，垂直位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... class PageScroller(Scroller): - def __init__(self, owner: Union[ChromiumBase, ChromiumElement]): ... + _owner: Union[ChromiumBase, ChromiumElement] = ... + + def __init__(self, owner: Union[ChromiumBase, ChromiumElement]): + """ + :param owner: 页面对象 + """ + ... def to_see(self, loc_or_ele: Union[str, tuple, ChromiumElement], - center: Union[bool, None] = None) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: ... + center: Union[bool, None] = None) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: + """滚动页面直到元素可见 + :param loc_or_ele: 元素的定位信息，可以是loc元组，或查询字符串 + :param center: 是否尽量滚动到页面正中，为None时如果被遮挡，则滚动到页面正中 + :return: None + """ + ... - def to_top(self) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: ... + def to_top(self) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: + """滚动到顶端，水平位置不变""" + ... - def to_bottom(self) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: ... + def to_bottom(self) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: + """滚动到底端，水平位置不变""" + ... - def to_half(self) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: ... + def to_half(self) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: + """滚动到垂直中间位置，水平位置不变""" + ... - def to_rightmost(self) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: ... + def to_rightmost(self) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: + """滚动到最右边，垂直位置不变""" + ... - def to_leftmost(self) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: ... + def to_leftmost(self) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: + """滚动到最左边，垂直位置不变""" + ... - def to_location(self, x: int, y: int) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: ... + def to_location(self, x: int, y: int) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: + """滚动到指定位置 + :param x: 水平距离 + :param y: 垂直距离 + :return: None + """ + ... - def up(self, pixel: int = 300) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: ... + def up(self, pixel: int = 300) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: + """向上滚动若干像素，水平位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... - def down(self, pixel: int = 300) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: ... + def down(self, pixel: int = 300) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: + """向下滚动若干像素，水平位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... - def left(self, pixel: int = 300) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: ... + def left(self, pixel: int = 300) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: + """向左滚动若干像素，垂直位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... - def right(self, pixel: int = 300) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: ... + def right(self, pixel: int = 300) -> Union[ChromiumTab, MixTab, ChromiumPage, MixPage]: + """向右滚动若干像素，垂直位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... - def _to_see(self, ele: ChromiumElement, center: Union[bool, None]) -> None: ... + def _to_see(self, ele: ChromiumElement, center: Union[bool, None]) -> None: + """执行滚动页面直到元素可见 + :param ele: 元素对象 + :param center: 是否尽量滚动到页面正中，为None时如果被遮挡，则滚动到页面正中 + :return: None + """ + ... class FrameScroller(PageScroller): - def __init__(self, frame: ChromiumFrame): ... + _owner: ChromiumElement = ... - def to_top(self) -> ChromiumFrame: ... + def __init__(self, owner: ChromiumFrame): ... - def to_bottom(self) -> ChromiumFrame: ... + def to_see(self, + loc_or_ele: Union[str, tuple, ChromiumElement], + center: Union[bool, None] = None) -> ChromiumFrame: + """滚动页面直到元素可见 + :param loc_or_ele: 元素的定位信息，可以是loc元组，或查询字符串 + :param center: 是否尽量滚动到页面正中，为None时如果被遮挡，则滚动到页面正中 + :return: None + """ + ... - def to_half(self) -> ChromiumFrame: ... + def to_top(self) -> ChromiumFrame: + """滚动到顶端，水平位置不变""" + ... - def to_rightmost(self) -> ChromiumFrame: ... + def to_bottom(self) -> ChromiumFrame: + """滚动到底端，水平位置不变""" + ... - def to_leftmost(self) -> ChromiumFrame: ... + def to_half(self) -> ChromiumFrame: + """滚动到垂直中间位置，水平位置不变""" + ... - def to_location(self, x: int, y: int) -> ChromiumFrame: ... + def to_rightmost(self) -> ChromiumFrame: + """滚动到最右边，垂直位置不变""" + ... - def up(self, pixel: int = 300) -> ChromiumFrame: ... + def to_leftmost(self) -> ChromiumFrame: + """滚动到最左边，垂直位置不变""" + ... - def down(self, pixel: int = 300) -> ChromiumFrame: ... + def to_location(self, x: int, y: int) -> ChromiumFrame: + """滚动到指定位置 + :param x: 水平距离 + :param y: 垂直距离 + :return: None + """ + ... - def left(self, pixel: int = 300) -> ChromiumFrame: ... + def up(self, pixel: int = 300) -> ChromiumFrame: + """向上滚动若干像素，水平位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... - def right(self, pixel: int = 300) -> ChromiumFrame: ... + def down(self, pixel: int = 300) -> ChromiumFrame: + """向下滚动若干像素，水平位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... - def to_see(self, loc_or_ele, center=None) -> ChromiumFrame: ... + def left(self, pixel: int = 300) -> ChromiumFrame: + """向左滚动若干像素，垂直位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... + + def right(self, pixel: int = 300) -> ChromiumFrame: + """向右滚动若干像素，垂直位置不变 + :param pixel: 滚动的像素 + :return: None + """ + ... diff --git a/DrissionPage/_units/selector.py b/DrissionPage/_units/selector.py index 27301ad..52d8883 100644 --- a/DrissionPage/_units/selector.py +++ b/DrissionPage/_units/selector.py @@ -12,57 +12,38 @@ class SelectElement(object): """用于处理 select 标签""" def __init__(self, ele): - """ - :param ele: select 元素对象 - """ if ele.tag != 'select': raise TypeError("select方法只能在元素对象 + """ + ... - def __call__(self, text_or_index: Union[str, int, list, tuple], timeout: float = None) -> bool: ... + def __call__(self, + text_or_index: Union[str, int, list, tuple], + timeout: float = None) -> bool: + """选定下拉列表中子元素 + :param text_or_index: 根据文本、值选或序号择选项，若允许多选，传入list或tuple可多选 + :param timeout: 超时时间（秒），不输入默认实用页面超时时间 + :return: None + """ + ... @property - def is_multi(self) -> bool: ... + def is_multi(self) -> bool: + """返回是否多选表单""" + ... @property - def options(self) -> List[ChromiumElement]: ... + def options(self) -> List[ChromiumElement]: + """返回所有选项元素组成的列表""" + ... @property - def selected_option(self) -> Union[ChromiumElement, None]: ... + def selected_option(self) -> Optional[ChromiumElement]: + """返回第一个被选中的