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

自动关闭浏览器启动时的隐私声明

Browse Source
This commit is contained in:
g1879 2023-11-12 22:56:52 +08:00
parent 89e1b3a29f
commit 27bea5f579
11 changed files with 97 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
DrissionPage/_base/browser.py
View File

@ -51,8 +51,8 @@ class Browser(object):
break
self.run_cdp('Target.setDiscoverTargets', discover=True)
self._driver.set_listener('Target.targetDestroyed', self._onTargetDestroyed)
self._driver.set_listener('Target.targetCreated', self._onTargetCreated)
self._driver.set_callback('Target.targetDestroyed', self._onTargetDestroyed)
self._driver.set_callback('Target.targetCreated', self._onTargetCreated)
def _get_driver(self, tab_id):
"""获取对应tab id的ChromiumDriver
@ -70,7 +70,8 @@ class Browser(object):
def _onTargetDestroyed(self, **kwargs):
"""标签页关闭时执行"""
tab_id = kwargs['targetId']
self._dl_mgr.clear_tab_info(tab_id)
if hasattr(self, '_dl_mgr'):
self._dl_mgr.clear_tab_info(tab_id)
for key in [k for k, i in self._frames.items() if i == tab_id]:
self._frames.pop(key, None)
self._drivers.pop(tab_id, None)
@ -87,7 +88,7 @@ class Browser(object):
:param cmd_args: 参数
:return: 执行的结果
"""
return self._driver.call_method(cmd, **cmd_args)
return self._driver.run(cmd, **cmd_args)
@property
def driver(self):

12
DrissionPage/_base/chromium_driver.py
View File

@ -139,15 +139,15 @@ class ChromiumDriver(object):
function = self.event_handlers.get(event['method'])
if function:
if self._debug:
print(f'开始执行 {function.__name__}')
# if self._debug:
# print(f'开始执行 {function.__name__}')
function(**event['params'])
if self._debug:
print(f'执行 {function.__name__}完毕')
# if self._debug:
# print(f'执行 {function.__name__}完毕')
self.event_queue.task_done()
def call_method(self, _method, **kwargs):
def run(self, _method, **kwargs):
"""执行cdp方法
:param _method: cdp方法名
:param args: cdp参数
@ -205,7 +205,7 @@ class ChromiumDriver(object):
self.event_queue.queue.clear()
return True
def set_listener(self, event, callback):
def set_callback(self, event, callback):
"""绑定cdp event和回调方法
:param event: cdp event
:param callback: 绑定到cdp event的回调方法

4
DrissionPage/_base/chromium_driver.pyi
View File

@ -47,13 +47,13 @@ class ChromiumDriver(object):
def __getattr__(self, item: str) -> Callable: ...
def call_method(self, _method: str, **kwargs) -> dict: ...
def run(self, _method: str, **kwargs) -> dict: ...
def start(self) -> bool: ...
def stop(self) -> bool: ...
def set_listener(self, event: str, callback: Union[Callable, None]) -> None: ...
def set_callback(self, event: str, callback: Union[Callable, None]) -> None: ...
def __str__(self) -> str: ...

70
DrissionPage/_pages/chromium_base.py
View File

@ -100,13 +100,17 @@ class ChromiumBase(BasePage):
"""
self._is_loading = True
self._driver = self.browser._get_driver(tab_id)
self._alert = Alert()
self._driver.set_listener('Page.javascriptDialogOpening', self._on_alert_open)
self._driver.set_listener('Page.javascriptDialogClosed', self._on_alert_close)
if self._driver.run('Target.getTargetInfo',
targetId=self._target_id)['targetInfo']['url'] == 'chrome://privacy-sandbox-dialog/notice':
self._driver = close_privacy_dialog(self)
self._driver.call_method('DOM.enable')
self._driver.call_method('Page.enable')
self._driver.call_method('Emulation.setFocusEmulationEnabled', enabled=True)
self._alert = Alert()
self._driver.set_callback('Page.javascriptDialogOpening', self._on_alert_open)
self._driver.set_callback('Page.javascriptDialogClosed', self._on_alert_close)
self._driver.run('DOM.enable')
self._driver.run('Page.enable')
self._driver.run('Emulation.setFocusEmulationEnabled', enabled=True)
r = self.run_cdp('Page.getFrameTree')
for i in findall(r"'id': '(.*?)'", str(r)):
@ -114,13 +118,13 @@ class ChromiumBase(BasePage):
if not hasattr(self, '_frame_id'):
self._frame_id = r['frameTree']['frame']['id']
self._driver.set_listener('Page.frameStartedLoading', self._onFrameStartedLoading)
self._driver.set_listener('Page.frameNavigated', self._onFrameNavigated)
self._driver.set_listener('Page.domContentEventFired', self._onDomContentEventFired)
self._driver.set_listener('Page.loadEventFired', self._onLoadEventFired)
self._driver.set_listener('Page.frameStoppedLoading', self._onFrameStoppedLoading)
self._driver.set_listener('Page.frameAttached', self._onFrameAttached)
self._driver.set_listener('Page.frameDetached', self._onFrameDetached)
self._driver.set_callback('Page.frameStartedLoading', self._onFrameStartedLoading)
self._driver.set_callback('Page.frameNavigated', self._onFrameNavigated)
self._driver.set_callback('Page.domContentEventFired', self._onDomContentEventFired)
self._driver.set_callback('Page.loadEventFired', self._onLoadEventFired)
self._driver.set_callback('Page.frameStoppedLoading', self._onFrameStoppedLoading)
self._driver.set_callback('Page.frameAttached', self._onFrameAttached)
self._driver.set_callback('Page.frameDetached', self._onFrameDetached)
def _get_document(self):
if self._is_reading:
@ -231,7 +235,7 @@ class ChromiumBase(BasePage):
files = self._upload_list if kwargs['mode'] == 'selectMultiple' else self._upload_list[:1]
self.run_cdp('DOM.setFileInputFiles', files=files, backendNodeId=kwargs['backendNodeId'])
self.driver.set_listener('Page.fileChooserOpened', None)
self.driver.set_callback('Page.fileChooserOpened', None)
self.run_cdp('Page.setInterceptFileChooserDialog', enabled=False)
self._upload_list = None
@ -416,7 +420,7 @@ class ChromiumBase(BasePage):
:param cmd_args: 参数
:return: 执行的结果
"""
r = self.driver.call_method(cmd, **cmd_args)
r = self.driver.run(cmd, **cmd_args)
if ERROR not in r:
return r
@ -822,9 +826,9 @@ class ChromiumBase(BasePage):
res_text = self._alert.text
if self._alert.type == 'prompt':
self.driver.call_method('Page.handleJavaScriptDialog', accept=accept, promptText=send)
self.driver.run('Page.handleJavaScriptDialog', accept=accept, promptText=send)
else:
self.driver.call_method('Page.handleJavaScriptDialog', accept=accept)
self.driver.run('Page.handleJavaScriptDialog', accept=accept)
return res_text
def _on_alert_close(self, **kwargs):
@ -1032,3 +1036,35 @@ class Alert(object):
self.defaultPrompt = None
self.response_accept = None
self.response_text = None
def close_privacy_dialog(page):
"""关闭隐私声明弹窗
:param page: ChromiumBase对象
:return: ChromiumDriver对象
"""
tid = page.tab_id
page._driver.run('Runtime.enable')
page._driver.run('DOM.enable')
page._driver.run('DOM.getDocument')
sid = page._driver.run('DOM.performSearch', query='//*[name()="privacy-sandbox-notice-dialog-app"]',
includeUserAgentShadowDOM=True)['searchId']
r = page._driver.run('DOM.getSearchResults', searchId=sid, fromIndex=0, toIndex=1)['nodeIds'][0]
while True:
try:
r = page._driver.run('DOM.describeNode', nodeId=r)['node']['shadowRoots'][0]['backendNodeId']
break
except KeyError:
pass
page._driver.run('DOM.discardSearchResults', searchId=sid)
r = page._driver.run('DOM.resolveNode', backendNodeId=r)['object']['objectId']
r = page._driver.run('Runtime.callFunctionOn', objectId=r,
functionDeclaration='function()'
'{return this.getElementById("ackButton");}')['result']['objectId']
page._driver.run('Runtime.callFunctionOn', objectId=r, functionDeclaration='function(){return this.click();}')
while True:
new_tid = page.browser.tabs[0]
if new_tid != tid:
break
sleep(.1)
return page.browser._get_driver(new_tid)

2
DrissionPage/_pages/chromium_frame.py
View File

@ -90,7 +90,7 @@ class ChromiumFrame(ChromiumBase):
except:
self.browser.driver.get(f'http://{self.address}/json')
super()._driver_init(tab_id)
self._driver.set_listener('Inspector.detached', self._onInspectorDetached)
self._driver.set_callback('Inspector.detached', self._onInspectorDetached)
def _reload(self):
"""重新获取document"""

2
DrissionPage/_pages/web_page.py
View File

@ -314,7 +314,7 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
if self._has_driver:
self.change_mode('s')
try:
self.driver.call_method('Browser.close')
self.driver.run('Browser.close')
except Exception:
pass
self._driver.stop()

18
DrissionPage/_units/action_chains.py
View File

@ -59,7 +59,7 @@ class ActionChains:
cx = x + offset_x
cy = y + offset_y
self._dr.call_method('Input.dispatchMouseEvent', type='mouseMoved', x=cx, y=cy, modifiers=self.modifier)
self._dr.run('Input.dispatchMouseEvent', type='mouseMoved', x=cx, y=cy, modifiers=self.modifier)
self.curr_x = cx
self.curr_y = cy
return self
@ -72,8 +72,8 @@ class ActionChains:
"""
self.curr_x += offset_x
self.curr_y += offset_y
self._dr.call_method('Input.dispatchMouseEvent', type='mouseMoved', x=self.curr_x, y=self.curr_y,
modifiers=self.modifier)
self._dr.run('Input.dispatchMouseEvent', type='mouseMoved', x=self.curr_x, y=self.curr_y,
modifiers=self.modifier)
return self
def click(self, on_ele=None):
@ -171,8 +171,8 @@ class ActionChains:
"""
if on_ele:
self.move_to(on_ele)
self._dr.call_method('Input.dispatchMouseEvent', type='mousePressed', button=button, clickCount=count,
x=self.curr_x, y=self.curr_y, modifiers=self.modifier)
self._dr.run('Input.dispatchMouseEvent', type='mousePressed', button=button, clickCount=count,
x=self.curr_x, y=self.curr_y, modifiers=self.modifier)
return self
def _release(self, button):
@ -180,8 +180,8 @@ class ActionChains:
:param button: 要释放的按键
:return: self
"""
self._dr.call_method('Input.dispatchMouseEvent', type='mouseReleased', button=button, clickCount=1,
x=self.curr_x, y=self.curr_y, modifiers=self.modifier)
self._dr.run('Input.dispatchMouseEvent', type='mouseReleased', button=button, clickCount=1,
x=self.curr_x, y=self.curr_y, modifiers=self.modifier)
return self
def scroll(self, delta_x=0, delta_y=0, on_ele=None):
@ -193,8 +193,8 @@ class ActionChains:
"""
if on_ele:
self.move_to(on_ele)
self._dr.call_method('Input.dispatchMouseEvent', type='mouseWheel', x=self.curr_x, y=self.curr_y,
deltaX=delta_x, deltaY=delta_y, modifiers=self.modifier)
self._dr.run('Input.dispatchMouseEvent', type='mouseWheel', x=self.curr_x, y=self.curr_y,
deltaX=delta_x, deltaY=delta_y, modifiers=self.modifier)
return self
def up(self, pixel):

4
DrissionPage/_units/download_manager.py
View File

@ -27,8 +27,8 @@ class DownloadManager(object):
self._tab_missions = {} # {tab_id: DownloadMission}
self._flags = {} # {tab_id: [bool, DownloadMission]}
self._browser.driver.set_listener('Browser.downloadProgress', self._onDownloadProgress)
self._browser.driver.set_listener('Browser.downloadWillBegin', self._onDownloadWillBegin)
self._browser.driver.set_callback('Browser.downloadProgress', self._onDownloadProgress)
self._browser.driver.set_callback('Browser.downloadWillBegin', self._onDownloadWillBegin)
self._browser.run_cdp('Browser.setDownloadBehavior', downloadPath=self._page.download_path,
behavior='allowAndName', eventsEnabled=True)

33
DrissionPage/_units/network_listener.py
View File

@ -78,7 +78,7 @@ class NetworkListener(object):
return
self._driver = ChromiumDriver(self._page.tab_id, 'page', self._page.address)
self._driver.call_method('Network.enable')
self._driver.run('Network.enable')
self.listening = True
self._request_ids = {}
@ -158,10 +158,10 @@ class NetworkListener(object):
:return: None
"""
if self.listening:
self._driver.set_listener('Network.requestWillBeSent', None)
self._driver.set_listener('Network.responseReceived', None)
self._driver.set_listener('Network.loadingFinished', None)
self._driver.set_listener('Network.loadingFailed', None)
self._driver.set_callback('Network.requestWillBeSent', None)
self._driver.set_callback('Network.responseReceived', None)
self._driver.set_callback('Network.loadingFinished', None)
self._driver.set_callback('Network.loadingFailed', None)
self.listening = False
if clear:
self.clear()
@ -181,12 +181,12 @@ class NetworkListener(object):
def _set_callback(self):
"""设置监听请求的回调函数"""
self._driver.set_listener('Network.requestWillBeSent', self._requestWillBeSent)
self._driver.set_listener('Network.requestWillBeSentExtraInfo', self._requestWillBeSentExtraInfo)
self._driver.set_listener('Network.responseReceived', self._response_received)
self._driver.set_listener('Network.responseReceivedExtraInfo', self._responseReceivedExtraInfo)
self._driver.set_listener('Network.loadingFinished', self._loading_finished)
self._driver.set_listener('Network.loadingFailed', self._loading_failed)
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)
self._driver.set_callback('Network.responseReceivedExtraInfo', self._responseReceivedExtraInfo)
self._driver.set_callback('Network.loadingFinished', self._loading_finished)
self._driver.set_callback('Network.loadingFailed', self._loading_failed)
def _requestWillBeSent(self, **kwargs):
"""接收到请求时的回调函数"""
@ -197,7 +197,7 @@ class NetworkListener(object):
p = self._request_ids.setdefault(rid, DataPacket(self._page.tab_id, None))
p._raw_request = kwargs
if kwargs['request'].get('hasPostData', None) and not kwargs['request'].get('postData', None):
p._raw_post_data = self._driver.call_method('Network.getRequestPostData', requestId=rid)['postData']
p._raw_post_data = self._driver.run('Network.getRequestPostData', requestId=rid)['postData']
else:
rid = kwargs['requestId']
@ -208,8 +208,8 @@ class NetworkListener(object):
p = self._request_ids.setdefault(rid, DataPacket(self._page.tab_id, target))
p._raw_request = kwargs
if kwargs['request'].get('hasPostData', None) and not kwargs['request'].get('postData', None):
p._raw_post_data = self._driver.call_method('Network.getRequestPostData',
requestId=rid)['postData']
p._raw_post_data = self._driver.run('Network.getRequestPostData',
requestId=rid)['postData']
break
self._extra_info_ids.setdefault(kwargs['requestId'], {})['obj'] = p if p else False
@ -244,7 +244,7 @@ class NetworkListener(object):
r_id = kwargs['requestId']
dp = self._request_ids.get(r_id)
if dp:
r = self._driver.call_method('Network.getResponseBody', requestId=r_id)
r = self._driver.run('Network.getResponseBody', requestId=r_id)
if 'body' in r:
dp._raw_body = r['body']
dp._base64_body = r['base64Encoded']
@ -364,9 +364,6 @@ class DataPacket(object):
:param timeout: 超时时间，None为无限等待
:return: 是否等待成功
"""
if self._raw_request['redirectHasExtraInfo'] is False:
return True
if timeout is None:
while self._responseExtraInfo is None:
sleep(.1)

4
DrissionPage/_units/screencast.py
View File

@ -35,7 +35,7 @@ class Screencast(object):
raise ValueError('save_path必须设置。')
clean_folder(self._path)
if self._mode.startswith('frugal'):
self._page.driver.set_listener('Page.screencastFrame', self._onScreencastFrame)
self._page.driver.set_callback('Page.screencastFrame', self._onScreencastFrame)
self._page.run_cdp('Page.startScreencast', everyNthFrame=1, quality=100)
elif not self._mode.startswith('js'):
@ -92,7 +92,7 @@ class Screencast(object):
return path
if self._mode.startswith('frugal'):
self._page.driver.set_listener('Page.screencastFrame', None)
self._page.driver.set_callback('Page.screencastFrame', None)
self._page.run_cdp('Page.stopScreencast')
else:
self._enable = False

8
DrissionPage/_units/setter.py
View File

@ -102,7 +102,7 @@ class ChromiumBaseSetter(object):
:return: None
"""
if not self._page._upload_list:
self._page.driver.set_listener('Page.fileChooserOpened', self._page._onFileChooserOpened)
self._page.driver.set_callback('Page.fileChooserOpened', self._page._onFileChooserOpened)
self._page.run_cdp('Page.setInterceptFileChooserDialog', enabled=True)
if isinstance(files, str):
@ -164,12 +164,6 @@ class TabSetter(ChromiumBaseSetter):
class ChromiumPageSetter(TabSetter):
def main_tab(self, tab_id=None):
"""设置主tab
:param tab_id: 标签页id，不传入则设置当前tab
:return: None
"""
self._page._main_tab = tab_id or self._page.tab_id
def tab_to_front(self, tab_or_id=None):
"""激活标签页使其处于最前面