blackteay/DrissionPage
1
0
Fork 0
mirror of https://gitee.com/g1879/DrissionPage.git synced 2024-12-10 04:00:23 +08:00
Code Issues Projects Releases Wiki Activity

ChromiumFrame增加_target_page

Browse Source
This commit is contained in:
g1879 2023-08-26 08:46:25 +08:00
parent 8d89e4f7ef
commit fd5e8aa89d
5 changed files with 49 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
DrissionPage/chromium_driver.py
View File

@ -25,7 +25,7 @@ class ChromiumDriver(object):
self.id = tab_id self.id = tab_id
self.address = address self.address = address
self.type = tab_type self.type = tab_type
self.debug = False self._debug = False
self.has_alert = False self.has_alert = False
self._websocket_url = f'ws://{address}/devtools/{tab_type}/{tab_id}' self._websocket_url = f'ws://{address}/devtools/{tab_type}/{tab_id}'
@ -57,11 +57,12 @@ class ChromiumDriver(object):
message_json = dumps(message) message_json = dumps(message)
if self.debug: if self._debug:
if self.debug is True or (isinstance(self.debug, str) and message.get('method', '').startswith(self.debug)): if self._debug is True or (
isinstance(self._debug, str) and message.get('method', '').startswith(self._debug)):
print(f'发> {message_json}') print(f'发> {message_json}')
elif isinstance(self.debug, (list, tuple, set)): elif isinstance(self._debug, (list, tuple, set)):
for m in self.debug: for m in self._debug:
if message.get('method', '').startswith(m): if message.get('method', '').startswith(m):
print(f'发> {message_json}') print(f'发> {message_json}')
break break
@ -112,12 +113,12 @@ class ChromiumDriver(object):
self.stop() self.stop()
return return
if self.debug: if self._debug:
if self.debug is True or 'id' in mes or (isinstance(self.debug, str) if self._debug is True or 'id' in mes or (isinstance(self._debug, str)
and mes.get('method', '').startswith(self.debug)): and mes.get('method', '').startswith(self._debug)):
print(f'<收 {message_json}') print(f'<收 {message_json}')
elif isinstance(self.debug, (list, tuple, set)): elif isinstance(self._debug, (list, tuple, set)):
for m in self.debug: for m in self._debug:
if mes.get('method', '').startswith(m): if mes.get('method', '').startswith(m):
print(f'<收 {message_json}') print(f'<收 {message_json}')
break break
@ -129,7 +130,7 @@ class ChromiumDriver(object):
if mes["id"] in self.method_results: if mes["id"] in self.method_results:
self.method_results[mes['id']].put(mes) self.method_results[mes['id']].put(mes)
elif self.debug: elif self._debug:
print(f'未知信息：{mes}') print(f'未知信息：{mes}')
def _handle_event_loop(self): def _handle_event_loop(self):

2
DrissionPage/chromium_driver.pyi
View File

@ -23,7 +23,7 @@ class ChromiumDriver(object):
id: str id: str
address: str address: str
type: str type: str
debug: bool _debug: bool
has_alert: bool has_alert: bool
_websocket_url: str _websocket_url: str
_cur_id: int _cur_id: int

45
DrissionPage/chromium_frame.py
View File

@ -21,7 +21,14 @@ class ChromiumFrame(ChromiumBase):
:param page: frame所在的页面对象 :param page: frame所在的页面对象
:param ele: frame所在元素 :param ele: frame所在元素
""" """
self.page = page page_type = str(type(page))
if 'ChromiumPage' in page_type or 'WebPage' in page:
self.page = self._target_page = self.tab = page
else: # Tab、Frame
self.page = page.page
self._target_page = page
self.tab = page.tab if 'ChromiumFrame' in page_type else page
self.address = page.address self.address = page.address
node = page.run_cdp('DOM.describeNode', backendNodeId=ele.ids.backend_id)['node'] node = page.run_cdp('DOM.describeNode', backendNodeId=ele.ids.backend_id)['node']
self.frame_id = node['frameId'] self.frame_id = node['frameId']
@ -32,7 +39,7 @@ class ChromiumFrame(ChromiumBase):
if self._is_inner_frame(): if self._is_inner_frame():
self._is_diff_domain = False self._is_diff_domain = False
self.doc_ele = ChromiumElement(self.page, backend_id=node['contentDocument']['backendNodeId']) self.doc_ele = ChromiumElement(self._target_page, backend_id=node['contentDocument']['backendNodeId'])
super().__init__(page.address, page.tab_id, page.timeout) super().__init__(page.address, page.tab_id, page.timeout)
else: else:
self._is_diff_domain = True self._is_diff_domain = True
@ -65,8 +72,8 @@ class ChromiumFrame(ChromiumBase):
def _runtime_settings(self): def _runtime_settings(self):
"""重写设置浏览器运行参数方法""" """重写设置浏览器运行参数方法"""
self._timeouts = self.page.timeouts self._timeouts = self._target_page.timeouts
self._page_load_strategy = self.page.page_load_strategy self._page_load_strategy = self._target_page.page_load_strategy
def _driver_init(self, tab_id): def _driver_init(self, tab_id):
"""避免出现服务器500错误 """避免出现服务器500错误
@ -87,18 +94,18 @@ class ChromiumFrame(ChromiumBase):
if debug: if debug:
print('重新获取document') print('重新获取document')
self._frame_ele = ChromiumElement(self.page, backend_id=self._backend_id) self._frame_ele = ChromiumElement(self._target_page, backend_id=self._backend_id)
node = self.page.run_cdp('DOM.describeNode', backendNodeId=self._frame_ele.ids.backend_id)['node'] node = self._target_page.run_cdp('DOM.describeNode', backendNodeId=self._frame_ele.ids.backend_id)['node']
if self._is_inner_frame(): if self._is_inner_frame():
self._is_diff_domain = False self._is_diff_domain = False
self.doc_ele = ChromiumElement(self.page, backend_id=node['contentDocument']['backendNodeId']) self.doc_ele = ChromiumElement(self._target_page, backend_id=node['contentDocument']['backendNodeId'])
super().__init__(self.address, self.page.tab_id, self.page.timeout) super().__init__(self.address, self._target_page.tab_id, self._target_page.timeout)
self._debug = debug self._debug = debug
else: else:
self._is_diff_domain = True self._is_diff_domain = True
self._tab_obj.stop() self._tab_obj.stop()
super().__init__(self.address, self.frame_id, self.page.timeout) super().__init__(self.address, self.frame_id, self._target_page.timeout)
obj_id = super().run_js('document;', as_expr=True)['objectId'] obj_id = super().run_js('document;', as_expr=True)['objectId']
self.doc_ele = ChromiumElement(self, obj_id=obj_id) self.doc_ele = ChromiumElement(self, obj_id=obj_id)
self._debug = debug self._debug = debug
@ -109,7 +116,7 @@ class ChromiumFrame(ChromiumBase):
self._reload() self._reload()
try: try:
self.page.run_cdp('DOM.describeNode', nodeId=self.ids.node_id) self._target_page.run_cdp('DOM.describeNode', nodeId=self.ids.node_id)
except Exception: except Exception:
self._reload() self._reload()
# sleep(2) # sleep(2)
@ -126,8 +133,9 @@ class ChromiumFrame(ChromiumBase):
while self.is_alive and perf_counter() < end_time: while self.is_alive and perf_counter() < end_time:
try: try:
if self._is_diff_domain is False: if self._is_diff_domain is False:
node = self.page.run_cdp('DOM.describeNode', backendNodeId=self.ids.backend_id)['node'] node = self._target_page.run_cdp('DOM.describeNode', backendNodeId=self.ids.backend_id)['node']
self.doc_ele = ChromiumElement(self.page, backend_id=node['contentDocument']['backendNodeId']) self.doc_ele = ChromiumElement(self._target_page,
backend_id=node['contentDocument']['backendNodeId'])
else: else:
b_id = self.run_cdp('DOM.getDocument')['root']['backendNodeId'] b_id = self.run_cdp('DOM.getDocument')['root']['backendNodeId']
@ -207,7 +215,8 @@ class ChromiumFrame(ChromiumBase):
"""返回元素outerHTML文本""" """返回元素outerHTML文本"""
self._check_ok() self._check_ok()
tag = self.tag tag = self.tag
out_html = self.page.run_cdp('DOM.getOuterHTML', backendNodeId=self.frame_ele.ids.backend_id)['outerHTML'] out_html = self._target_page.run_cdp('DOM.getOuterHTML', backendNodeId=self.frame_ele.ids.backend_id)[
'outerHTML']
sign = search(rf'<{tag}.*?>', out_html).group(0) sign = search(rf'<{tag}.*?>', out_html).group(0)
return f'{sign}{self.inner_html}</{tag}>' return f'{sign}{self.inner_html}</{tag}>'
@ -296,7 +305,7 @@ class ChromiumFrame(ChromiumBase):
except ContextLossError: except ContextLossError:
try: try:
node = self.run_cdp('DOM.describeNode', backendNodeId=self.frame_ele.ids.backend_id)['node'] node = self.run_cdp('DOM.describeNode', backendNodeId=self.frame_ele.ids.backend_id)['node']
doc = ChromiumElement(self.page, backend_id=node['contentDocument']['backendNodeId']) doc = ChromiumElement(self._target_page, backend_id=node['contentDocument']['backendNodeId'])
return doc.run_js('return this.readyState;') return doc.run_js('return this.readyState;')
except: except:
pass pass
@ -525,7 +534,7 @@ class ChromiumFrame(ChromiumBase):
cx, cy = ele.locations.viewport_location cx, cy = ele.locations.viewport_location
w, h = ele.size w, h = ele.size
img_data = f'data:image/{pic_type};base64,{self.frame_ele.get_screenshot(as_base64=True)}' img_data = f'data:image/{pic_type};base64,{self.frame_ele.get_screenshot(as_base64=True)}'
body = self.page('t:body') body = self._target_page('t:body')
first_child = body('c::first-child') first_child = body('c::first-child')
if not isinstance(first_child, ChromiumElement): if not isinstance(first_child, ChromiumElement):
first_child = first_child.frame_ele first_child = first_child.frame_ele
@ -540,9 +549,9 @@ class ChromiumFrame(ChromiumBase):
new_ele.scroll.to_see(True) new_ele.scroll.to_see(True)
top = int(self.frame_ele.style('border-top').split('px')[0]) top = int(self.frame_ele.style('border-top').split('px')[0])
left = int(self.frame_ele.style('border-left').split('px')[0]) left = int(self.frame_ele.style('border-left').split('px')[0])
r = self.page.get_screenshot(path=path, as_bytes=as_bytes, as_base64=as_base64, r = self._target_page.get_screenshot(path=path, as_bytes=as_bytes, as_base64=as_base64,
left_top=(cx + left, cy + top), right_bottom=(cx + w + left, cy + h + top)) left_top=(cx + left, cy + top), right_bottom=(cx + w + left, cy + h + top))
self.page.remove_ele(new_ele) self._target_page.remove_ele(new_ele)
return r return r
def _find_elements(self, loc_or_ele, timeout=None, single=True, relative=False, raise_err=None): def _find_elements(self, loc_or_ele, timeout=None, single=True, relative=False, raise_err=None):
@ -610,7 +619,7 @@ class ChromiumFrame(ChromiumBase):
def _is_inner_frame(self): def _is_inner_frame(self):
"""返回当前frame是否同域""" """返回当前frame是否同域"""
return self.frame_id in str(self.page.run_cdp('Page.getFrameTree')['frameTree']) return self.frame_id in str(self._target_page.run_cdp('Page.getFrameTree')['frameTree'])
def _check_alive(self): def _check_alive(self):
"""检测iframe是否有效线程方法""" """检测iframe是否有效线程方法"""

3
DrissionPage/chromium_frame.pyi
View File

@ -16,6 +16,9 @@ class ChromiumFrame(ChromiumBase):
def __init__(self, page: ChromiumBase, ele: ChromiumElement): def __init__(self, page: ChromiumBase, ele: ChromiumElement):
self.page: ChromiumBase = ... self.page: ChromiumBase = ...
self._target_page = ...
self.tab = ...
self._tab_id: str = ...
self.frame_id: str = ... self.frame_id: str = ...
self._frame_ele: ChromiumElement = ... self._frame_ele: ChromiumElement = ...
self._backend_id: str = ... self._backend_id: str = ...

6
DrissionPage/chromium_page.py
View File

@ -343,6 +343,7 @@ class ChromiumPage(ChromiumBase):
p = popen(txt) p = popen(txt)
if f' {self.process_id} ' not in p.read(): if f' {self.process_id} ' not in p.read():
break break
sleep(.2)
def _on_alert_close(self, **kwargs): def _on_alert_close(self, **kwargs):
"""alert关闭时触发的方法""" """alert关闭时触发的方法"""
@ -438,8 +439,11 @@ class ChromiumTabRect(object):
class BrowserDownloadManager(object): class BrowserDownloadManager(object):
def __init__(self, page): def __init__(self, page):
self._page = page self.page = page
self.frames = {} self.frames = {}
self.pause = False
# class BaseDownloadSetter(DownloadSetter): # class BaseDownloadSetter(DownloadSetter):
# """用于设置下载参数的类""" # """用于设置下载参数的类"""
# #