DrissionPage/DrissionPage/_pages/chromium_frame.py
g1879 71dfe3037c wait.ele_display()和wait.ele_hidden()会等待元素加载;
修复Page获取元素有时会返回str问题
2023-12-21 10:48:48 +08:00

623 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# -*- coding:utf-8 -*-
"""
@Author : g1879
@Contact : g1879@qq.com
"""
from copy import copy
from re import search, findall, DOTALL
from time import sleep, perf_counter
from .._elements.chromium_element import ChromiumElement
from .._pages.chromium_base import ChromiumBase
from .._units.listener import FrameListener
from .._units.rect import FrameRect
from .._units.scroller import FrameScroller
from .._units.setter import ChromiumFrameSetter
from .._units.states import FrameStates
from .._units.waiter import FrameWaiter
from ..errors import ContextLostError, ElementLostError, GetDocumentError, PageClosedError, JavaScriptError
class ChromiumFrame(ChromiumBase):
def __init__(self, page, ele):
"""
:param page: frame所在的页面对象
:param ele: frame所在元素
"""
page_type = str(type(page))
if 'ChromiumPage' in page_type or 'WebPage' in page_type:
self._page = self._target_page = self.tab = page
self._browser = page.browser
else: # Tab、Frame
self._page = page.page
self._browser = self._page.browser
self._target_page = page
self.tab = page.tab if 'ChromiumFrame' in page_type else page
self.address = page.address
self._tab_id = page.tab_id
self._backend_id = ele._backend_id
self._frame_ele = ele
self._states = None
self._reloading = False
node = page.run_cdp('DOM.describeNode', backendNodeId=ele._backend_id)['node']
self._frame_id = node['frameId']
if self._is_inner_frame():
self._is_diff_domain = False
self.doc_ele = ChromiumElement(self._target_page, backend_id=node['contentDocument']['backendNodeId'])
super().__init__(page.address, page.tab_id, page.timeout)
else:
self._is_diff_domain = True
delattr(self, '_frame_id')
super().__init__(page.address, node['frameId'], page.timeout)
obj_id = super().run_js('document;', as_expr=True)['objectId']
self.doc_ele = ChromiumElement(self, obj_id=obj_id)
self._rect = None
end_time = perf_counter() + 5
while perf_counter() < end_time:
if self.url not in (None, 'about:blank'):
break
sleep(.1)
def __call__(self, loc_or_str, timeout=None):
"""在内部查找元素
ele2 = ele1('@id=ele_id')
:param loc_or_str: 元素的定位信息可以是loc元组或查询字符串
:param timeout: 超时时间
:return: ChromiumElement对象或属性、文本
"""
return self.ele(loc_or_str, timeout)
def __repr__(self):
attrs = self._frame_ele.attrs
attrs = [f"{attr}='{attrs[attr]}'" for attr in attrs]
return f'<ChromiumFrame {self.frame_ele.tag} {" ".join(attrs)}>'
def _d_set_runtime_settings(self):
"""重写设置浏览器运行参数方法"""
if not hasattr(self, '_timeouts'):
self._timeouts = copy(self._target_page.timeouts)
self.retry_times = self._target_page.retry_times
self.retry_interval = self._target_page.retry_interval
self._download_path = self._target_page.download_path
self._load_mode = self._target_page._load_mode if not self._is_diff_domain else 'normal'
def _driver_init(self, tab_id, is_init=True):
"""避免出现服务器500错误
:param tab_id: 要跳转到的标签页id
:return: None
"""
try:
super()._driver_init(tab_id)
except:
self.browser.driver.get(f'http://{self.address}/json')
super()._driver_init(tab_id)
self._driver.set_callback('Inspector.detached', self._onInspectorDetached)
def _reload(self):
"""重新获取document"""
self._is_loading = True
debug = self._debug
d_debug = self.driver._debug
self._reloading = True
self._doc_got = False
if debug:
print(f'{self._frame_id} reload 开始')
self._driver.stop()
try:
self._frame_ele = ChromiumElement(self._target_page, backend_id=self._backend_id)
end_time = perf_counter() + 2
while perf_counter() < end_time:
node = self._target_page.run_cdp('DOM.describeNode', backendNodeId=self._frame_ele._backend_id)['node']
if 'frameId' in node:
break
else:
return
except (ElementLostError, PageClosedError):
return
if self._is_inner_frame():
self._is_diff_domain = False
self.doc_ele = ChromiumElement(self._target_page, backend_id=node['contentDocument']['backendNodeId'])
self._frame_id = node['frameId']
if self._listener:
self._listener._to_target(self._target_page.tab_id, self.address, self)
super().__init__(self.address, self._target_page.tab_id, self._target_page.timeout)
self._debug = debug
self.driver._debug = d_debug
else:
self._is_diff_domain = True
if self._listener:
self._listener._to_target(node['frameId'], self.address, self)
super().__init__(self.address, node['frameId'], self._target_page.timeout)
end_time = perf_counter() + self.timeouts.page_load
while perf_counter() < end_time:
try:
obj_id = super().run_js('document;', as_expr=True)['objectId']
self.doc_ele = ChromiumElement(self, obj_id=obj_id)
break
except Exception as e:
sleep(.1)
if self._debug:
print(f'获取doc失败重试 {e}')
else:
raise GetDocumentError
self._debug = debug
self.driver._debug = d_debug
self._is_loading = False
self._reloading = False
if self._debug:
print(f'{self._frame_id} reload 完毕')
def _get_document(self, timeout=10):
"""刷新cdp使用的document数据
:param timeout: 超时时间
:return: 是否获取成功
"""
if self._is_reading:
return
if self._debug:
print('获取文档开始')
self._is_reading = True
try:
if self._is_diff_domain is False:
node = self._target_page.run_cdp('DOM.describeNode', backendNodeId=self._backend_id)['node']
self.doc_ele = ChromiumElement(self._target_page, backend_id=node['contentDocument']['backendNodeId'])
else:
timeout = timeout if timeout >= .5 else .5
b_id = self.run_cdp('DOM.getDocument', _timeout=timeout)['root']['backendNodeId']
self.doc_ele = ChromiumElement(self, backend_id=b_id)
self._root_id = self.doc_ele._obj_id
r = self.run_cdp('Page.getFrameTree')
for i in findall(r"'id': '(.*?)'", str(r)):
self.browser._frames[i] = self.tab_id
if self._debug:
print('获取文档结束')
return True
except:
if self._debug:
print('获取文档失败')
return False
finally:
if not self._reloading: # 阻止reload时标识
self._is_loading = False
self._is_reading = False
def _onInspectorDetached(self, **kwargs):
"""异域转同域或退出"""
if self._debug:
print(f'{self._frame_id}触发InspectorDetached')
self._reload()
if self._debug:
print(f'{self._frame_id}执行InspectorDetached完毕')
def _onFrameDetached(self, **kwargs):
"""同域变异域"""
self.browser._frames.pop(kwargs['frameId'], None)
if kwargs['frameId'] == self._frame_id:
if self._debug:
print(f'{self._frame_id}触发FrameDetached')
self._reload()
if self._debug:
print(f'{self._frame_id}执行FrameDetached完毕')
# ----------挂件----------
@property
def scroll(self):
"""返回用于等待的对象"""
self.wait.load_complete()
if self._scroll is None:
self._scroll = FrameScroller(self)
return self._scroll
@property
def set(self):
"""返回用于等待的对象"""
if self._set is None:
self._set = ChromiumFrameSetter(self)
return self._set
@property
def states(self):
"""返回用于获取状态信息的对象"""
if self._states is None:
self._states = FrameStates(self)
return self._states
@property
def wait(self):
"""返回用于等待的对象"""
if self._wait is None:
self._wait = FrameWaiter(self)
return self._wait
@property
def rect(self):
"""返回获取坐标和大小的对象"""
if self._rect is None:
self._rect = FrameRect(self)
return self._rect
@property
def listen(self):
"""返回用于聆听数据包的对象"""
if self._listener is None:
self._listener = FrameListener(self)
return self._listener
# ----------挂件----------
@property
def _obj_id(self):
"""返回frame元素的object id"""
return self.frame_ele._obj_id
@property
def _node_id(self):
"""返回cdp中的node id"""
return self.frame_ele._node_id
@property
def page(self):
return self._page
@property
def frame_ele(self):
"""返回总页面上的frame元素"""
return self._frame_ele
@property
def tag(self):
"""返回元素tag"""
return self.frame_ele.tag
@property
def url(self):
"""返回frame当前访问的url"""
try:
return self.doc_ele.run_js('return this.location.href;')
except JavaScriptError:
return None
@property
def html(self):
"""返回元素outerHTML文本"""
tag = self.tag
out_html = self._target_page.run_cdp('DOM.getOuterHTML', backendNodeId=self.frame_ele._backend_id)['outerHTML']
sign = search(rf'<{tag}.*?>', out_html, DOTALL).group(0)
return f'{sign}{self.inner_html}</{tag}>'
@property
def inner_html(self):
"""返回元素innerHTML文本"""
return self.doc_ele.run_js('return this.documentElement.outerHTML;')
@property
def title(self):
"""返回页面title"""
r = self._ele('t:title', raise_err=False)
return r.text if r else None
@property
def cookies(self):
"""以dict格式返回cookies"""
return super().cookies if self._is_diff_domain else self.doc_ele.run_js('return this.cookie;')
@property
def attrs(self):
"""返回frame元素所有attribute属性"""
return self.frame_ele.attrs
@property
def active_ele(self):
"""返回当前焦点所在元素"""
return self.doc_ele.run_js('return this.activeElement;')
@property
def xpath(self):
"""返回frame的xpath绝对路径"""
return self.frame_ele.xpath
@property
def css_path(self):
"""返回frame的css selector绝对路径"""
return self.frame_ele.css_path
@property
def tab_id(self):
"""返回frame所在tab的id"""
return self._tab_id
@property
def download_path(self):
return self._download_path
@property
def _js_ready_state(self):
"""返回当前页面加载状态,'loading' 'interactive' 'complete'"""
if self._is_diff_domain:
return super()._js_ready_state
else:
try:
return self.doc_ele.run_js('return this.readyState;')
except ContextLostError:
try:
node = self.run_cdp('DOM.describeNode', backendNodeId=self.frame_ele._backend_id)['node']
doc = ChromiumElement(self._target_page, backend_id=node['contentDocument']['backendNodeId'])
return doc.run_js('return this.readyState;')
except:
return None
def refresh(self):
"""刷新frame页面"""
self.doc_ele.run_js('this.location.reload();')
def attr(self, attr):
"""返回frame元素attribute属性值
:param attr: 属性名
:return: 属性值文本没有该属性返回None
"""
return self.frame_ele.attr(attr)
def remove_attr(self, attr):
"""删除frame元素attribute属性
:param attr: 属性名
:return: None
"""
self.frame_ele.remove_attr(attr)
def run_js(self, script, *args, as_expr=False, timeout=None):
"""运行javascript代码
:param script: js文本
:param args: 参数按顺序在js文本中对应arguments[0]、arguments[1]...
:param as_expr: 是否作为表达式运行为True时args无效
:param timeout: js超时时间为None则使用页面timeouts.script设置
:return: 运行的结果
"""
if script.startswith('this.scrollIntoView'):
return self.frame_ele.run_js(script, *args, as_expr=as_expr, timeout=timeout)
else:
return self.doc_ele.run_js(script, *args, as_expr=as_expr, timeout=timeout)
def parent(self, level_or_loc=1, index=1):
"""返回上面某一级父元素,可指定层数或用查询语法定位
:param level_or_loc: 第几级父元素,或定位符
:param index: 当level_or_loc传入定位符使用此参数选择第几个结果
:return: 上级元素对象
"""
return self.frame_ele.parent(level_or_loc, index)
def prev(self, filter_loc='', index=1, timeout=0, ele_only=True):
"""返回当前元素前面一个符合条件的同级元素,可用查询语法筛选,可指定返回筛选结果的第几个
:param filter_loc: 用于筛选的查询语法
:param index: 前面第几个查询结果1开始
:param timeout: 查找节点的超时时间
:param ele_only: 是否只获取元素为False时把文本、注释节点也纳入
:return: 同级元素或节点
"""
return self.frame_ele.prev(filter_loc, index, timeout, ele_only=ele_only)
def next(self, filter_loc='', index=1, timeout=0, ele_only=True):
"""返回当前元素后面一个符合条件的同级元素,可用查询语法筛选,可指定返回筛选结果的第几个
:param filter_loc: 用于筛选的查询语法
:param index: 后面第几个查询结果1开始
:param timeout: 查找节点的超时时间
:param ele_only: 是否只获取元素为False时把文本、注释节点也纳入
:return: 同级元素或节点
"""
return self.frame_ele.next(filter_loc, index, timeout, ele_only=ele_only)
def before(self, filter_loc='', index=1, timeout=None, ele_only=True):
"""返回文档中当前元素前面符合条件的第一个元素,可用查询语法筛选,可指定返回筛选结果的第几个
查找范围不限同级元素而是整个DOM文档
:param filter_loc: 用于筛选的查询语法
:param index: 前面第几个查询结果1开始
:param timeout: 查找节点的超时时间
:param ele_only: 是否只获取元素为False时把文本、注释节点也纳入
:return: 本元素前面的某个元素或节点
"""
return self.frame_ele.before(filter_loc, index, timeout, ele_only=ele_only)
def after(self, filter_loc='', index=1, timeout=None, ele_only=True):
"""返回文档中此当前元素后面符合条件的第一个元素,可用查询语法筛选,可指定返回筛选结果的第几个
查找范围不限同级元素而是整个DOM文档
:param filter_loc: 用于筛选的查询语法
:param index: 后面第几个查询结果1开始
:param timeout: 查找节点的超时时间
:param ele_only: 是否只获取元素为False时把文本、注释节点也纳入
:return: 本元素后面的某个元素或节点
"""
return self.frame_ele.after(filter_loc, index, timeout, ele_only=ele_only)
def prevs(self, filter_loc='', timeout=0, ele_only=True):
"""返回当前元素前面符合条件的同级元素或节点组成的列表,可用查询语法筛选
:param filter_loc: 用于筛选的查询语法
:param timeout: 查找节点的超时时间
:param ele_only: 是否只获取元素为False时把文本、注释节点也纳入
:return: 同级元素或节点文本组成的列表
"""
return self.frame_ele.prevs(filter_loc, timeout, ele_only=ele_only)
def nexts(self, filter_loc='', timeout=0, ele_only=True):
"""返回当前元素后面符合条件的同级元素或节点组成的列表,可用查询语法筛选
:param filter_loc: 用于筛选的查询语法
:param timeout: 查找节点的超时时间
:param ele_only: 是否只获取元素为False时把文本、注释节点也纳入
:return: 同级元素或节点文本组成的列表
"""
return self.frame_ele.nexts(filter_loc, timeout, ele_only=ele_only)
def befores(self, filter_loc='', timeout=None, ele_only=True):
"""返回文档中当前元素前面符合条件的元素或节点组成的列表,可用查询语法筛选
查找范围不限同级元素而是整个DOM文档
:param filter_loc: 用于筛选的查询语法
:param timeout: 查找节点的超时时间
:param ele_only: 是否只获取元素为False时把文本、注释节点也纳入
:return: 本元素前面的元素或节点组成的列表
"""
return self.frame_ele.befores(filter_loc, timeout, ele_only=ele_only)
def afters(self, filter_loc='', timeout=None, ele_only=True):
"""返回文档中当前元素后面符合条件的元素或节点组成的列表,可用查询语法筛选
查找范围不限同级元素而是整个DOM文档
:param filter_loc: 用于筛选的查询语法
:param timeout: 查找节点的超时时间
:param ele_only: 是否只获取元素为False时把文本、注释节点也纳入
:return: 本元素前面的元素或节点组成的列表
"""
return self.frame_ele.afters(filter_loc, timeout, ele_only=ele_only)
def get_screenshot(self, path=None, name=None, as_bytes=None, as_base64=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参数无效
:return: 图片完整路径或字节文本
"""
return self.frame_ele.get_screenshot(path=path, name=name, as_bytes=as_bytes, as_base64=as_base64)
def _get_screenshot(self, path=None, name=None, as_bytes: [bool, str] = None, as_base64: [bool, str] = None,
full_page=False, left_top=None, right_bottom=None, ele=None):
"""实现截图
: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 not self._is_diff_domain:
return super().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)
if as_bytes:
if as_bytes is True:
pic_type = 'png'
else:
if as_bytes not in ('jpg', 'jpeg', 'png', 'webp'):
raise TypeError("只能接收 'jpg', 'jpeg', 'png', 'webp' 四种格式。")
pic_type = 'jpeg' if as_bytes == 'jpg' else as_bytes
elif as_base64:
if as_base64 is True:
pic_type = 'png'
else:
if as_base64 not in ('jpg', 'jpeg', 'png', 'webp'):
raise TypeError("只能接收 'jpg', 'jpeg', 'png', 'webp' 四种格式。")
pic_type = 'jpeg' if as_base64 == 'jpg' else as_base64
else:
path = str(path).rstrip('\\/') if path else '.'
if path and path.endswith(('.jpg', '.jpeg', '.png', '.webp')):
pic_type = path.rsplit('.', 1)[-1]
elif name and name.endswith(('.jpg', '.jpeg', '.png', '.webp')):
pic_type = name.rsplit('.', 1)[-1]
else:
pic_type = 'jpeg'
if pic_type == 'jpg':
pic_type = 'jpeg'
self.frame_ele.scroll.to_see(center=True)
self.scroll.to_see(ele, center=True)
cx, cy = ele.rect.viewport_location
w, h = ele.rect.size
img_data = f'data:image/{pic_type};base64,{self.frame_ele.get_screenshot(as_base64=True)}'
body = self.tab('t:body')
first_child = body('c::first-child')
if not isinstance(first_child, ChromiumElement):
first_child = first_child.frame_ele
js = f'''
img = document.createElement('img');
img.src = "{img_data}";
img.style.setProperty("z-index",9999999);
img.style.setProperty("position","fixed");
arguments[0].insertBefore(img, this);
return img;'''
new_ele = first_child.run_js(js, body)
new_ele.scroll.to_see(center=True)
top = int(self.frame_ele.style('border-top').split('px')[0])
left = int(self.frame_ele.style('border-left').split('px')[0])
r = self.tab.run_cdp('Page.getLayoutMetrics')['visualViewport']
sx = r['pageX']
sy = r['pageY']
r = self.tab.get_screenshot(path=path, name=name, as_bytes=as_bytes, as_base64=as_base64,
left_top=(cx + left + sx, cy + top + sy),
right_bottom=(cx + w + left + sx, cy + h + top + sy))
self.tab.remove_ele(new_ele)
return r
def _find_elements(self, loc_or_ele, timeout=None, single=True, relative=False, raise_err=None):
"""在frame内查找单个元素
:param loc_or_ele: 定位符或元素对象
:param timeout: 查找超时时间
:param single: True则返回第一个False则返回全部
:param relative: WebPage用的表示是否相对定位的参数
:param raise_err: 找不到元素是是否抛出异常为None时根据全局设置
:return: ChromiumElement对象
"""
if isinstance(loc_or_ele, ChromiumElement):
return loc_or_ele
self.wait.load_complete()
return self.doc_ele._ele(loc_or_ele, timeout,
raise_err=raise_err) if single else self.doc_ele.eles(loc_or_ele, timeout)
def _is_inner_frame(self):
"""返回当前frame是否同域"""
return self._frame_id in str(self._target_page.run_cdp('Page.getFrameTree')['frameTree'])
# ----------------即将废弃-----------------
@property
def is_alive(self):
"""返回是否仍可用"""
return self.states.is_alive
@property
def page_size(self):
"""返回frame内页面尺寸格式(宽,, 高)"""
return self.rect.size
@property
def size(self):
"""返回frame元素大小"""
return self.frame_ele.rect.size
@property
def location(self):
"""返回frame元素左上角的绝对坐标"""
return self.frame_ele.rect.location
@property
def locations(self):
"""返回用于获取元素位置的对象"""
return self.frame_ele.rect