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

完善连接阶段逻辑和alert处理逻辑

Browse Source
This commit is contained in:
g1879 2023-12-18 01:07:11 +08:00
parent 05cac3b69a
commit 13cb1d7f53
6 changed files with 46 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
DrissionPage/_base/driver.py
View File

@ -37,6 +37,7 @@ class Driver(object):
self._stopped = Event()
self.event_handlers = {}
self.immediate_event_handlers = {}
self.method_results = {}
self.event_queue = Queue()
@ -63,12 +64,14 @@ class Driver(object):
print(f'发> {message_json}')
break
if timeout is not None:
timeout = perf_counter() + timeout
end_time = perf_counter() + timeout if timeout is not None else None
self.method_results[ws_id] = Queue()
try:
self._ws.send(message_json)
if timeout == 0:
self.method_results.pop(ws_id, None)
return {'id': ws_id, 'result': {}}
except (OSError, WebSocketConnectionClosedException):
self.method_results.pop(ws_id, None)
return None
@ -82,14 +85,12 @@ class Driver(object):
except Empty:
if self.alert_flag:
self.alert_flag = False
result = {'result': {'message': 'alert exists.'}}
self.method_results.pop(ws_id, None)
return result
return {'result': {'message': 'alert exists.'}}
elif timeout is not None and perf_counter() > timeout:
result = {'error': {'message': 'timeout'}}
elif timeout is not None and perf_counter() > end_time:
self.method_results.pop(ws_id, None)
return result
return {'error': {'message': 'timeout'}}
continue
@ -119,8 +120,12 @@ class Driver(object):
if 'method' in msg:
if msg['method'].startswith('Page.javascriptDialog'):
self.alert_flag = msg['method'].endswith('Opening')
self.event_queue.put(msg)
if msg['method'] in self.immediate_event_handlers:
function = self.immediate_event_handlers.get(msg['method'])
if function:
function(**msg['params'])
else:
self.event_queue.put(msg)
elif msg.get('id') in self.method_results:
self.method_results[msg['id']].put(msg)
@ -138,11 +143,7 @@ class Driver(object):
function = self.event_handlers.get(event['method'])
if function:
# if self._debug:
# print(f'开始执行 {function.__name__}')
function(**event['params'])
# if self._debug:
# print(f'执行 {function.__name__}完毕')
self.event_queue.task_done()
@ -156,8 +157,8 @@ class Driver(object):
if self._stopped.is_set():
return {'error': 'tab closed', 'type': 'tab_closed'}
timeout = kwargs.pop("_timeout", 20)
result = self._send({"method": _method, "params": kwargs}, timeout=timeout)
timeout = kwargs.pop('_timeout', 20)
result = self._send({'method': _method, 'params': kwargs}, timeout=timeout)
if result is None:
return {'error': 'tab closed', 'type': 'tab_closed'}
if 'result' not in result and 'error' in result:
@ -199,19 +200,21 @@ class Driver(object):
self.event_queue.queue.clear()
return True
def set_callback(self, event, callback):
def set_callback(self, event, callback, immediate=False):
"""绑定cdp event和回调方法
:param event: cdp event
:param callback: 绑定到cdp event的回调方法
:param immediate: 是否要立即处理的动作
:return: None
"""
handler = self.immediate_event_handlers if immediate else self.event_handlers
if callback:
self.event_handlers[event] = callback
handler[event] = callback
else:
self.event_handlers.pop(event, None)
handler.pop(event, None)
def __str__(self):
return f"<Driver {self.id}>"
return f'<Driver {self.id}>'
__repr__ = __str__
@ -233,7 +236,7 @@ class BrowserDriver(Driver):
self.browser = browser
def __repr__(self):
return f"<BrowserDriver {self.id}>"
return f'<BrowserDriver {self.id}>'
def get(self, url):
r = get(url, headers={'Connection': 'close'})

3
DrissionPage/_base/driver.pyi
View File

@ -34,6 +34,7 @@ class Driver(object):
_handle_event_th: Thread
_stopped: Event
event_handlers: dict
immediate_event_handlers: dict
method_results: dict
event_queue: Queue
@ -53,7 +54,7 @@ class Driver(object):
def stop(self) -> bool: ...
def set_callback(self, event: str, callback: Union[Callable, None]) -> None: ...
def set_callback(self, event: str, callback: Union[Callable, None], immediate: bool = False) -> None: ...
def __str__(self) -> str: ...

3
DrissionPage/_functions/tools.py
View File

@ -269,7 +269,8 @@ def raise_error(r):
elif error == ('tab closed', 'No target with given id found'):
raise PageClosedError
elif error == 'timeout':
raise TimeoutError
return
# raise TimeoutError
elif error == 'alert exists.':
raise AlertExistsError
elif error in ('Node does not have a layout object', 'Could not compute box model.'):

20
DrissionPage/_pages/chromium_base.py
View File

@ -123,7 +123,7 @@ class ChromiumBase(BasePage):
self._driver = self.browser._get_driver(tab_id)
self._alert = Alert()
self._driver.set_callback('Page.javascriptDialogOpening', self._on_alert_open)
self._driver.set_callback('Page.javascriptDialogOpening', self._on_alert_open, immediate=True)
self._driver.set_callback('Page.javascriptDialogClosed', self._on_alert_close)
self._driver.run('DOM.enable')
@ -837,6 +837,12 @@ class ChromiumBase(BasePage):
self.run_cdp_loaded('Network.clearBrowserCookies')
def handle_alert(self, accept=True, send=None, timeout=None, next_one=False):
r = self._handle_alert(accept=accept, send=send, timeout=timeout, next_one=next_one)
while self._has_alert:
sleep(.1)
return r
def _handle_alert(self, accept=True, send=None, timeout=None, next_one=False):
"""处理提示框，可以自动等待提示框出现
:param accept: True表示确认，False表示取消，其它值不会按按钮但依然返回文本值
:param send: 处理prompt提示框时可输入文本
@ -858,10 +864,10 @@ class ChromiumBase(BasePage):
return False
res_text = self._alert.text
if self._alert.type == 'prompt':
self.driver.run('Page.handleJavaScriptDialog', accept=accept, promptText=send)
else:
self.driver.run('Page.handleJavaScriptDialog', accept=accept)
d = {'accept': accept, '_timeout': 0}
if self._alert.type == 'prompt' and send is not None:
d['promptText'] = send
self.driver.run('Page.handleJavaScriptDialog', **d)
return res_text
def _on_alert_open(self, **kwargs):
@ -875,9 +881,9 @@ class ChromiumBase(BasePage):
self._has_alert = True
if self._alert.auto is not None:
self.handle_alert(self._alert.auto)
self._handle_alert(self._alert.auto)
elif self._alert.handle_next is not None:
self.handle_alert(self._alert.handle_next, self._alert.next_text)
self._handle_alert(self._alert.handle_next, self._alert.next_text)
self._alert.handle_next = None
def _on_alert_close(self, **kwargs):

3
DrissionPage/_pages/chromium_base.pyi
View File

@ -227,6 +227,9 @@ class ChromiumBase(BasePage):
def handle_alert(self, accept: 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): ...

4
DrissionPage/_units/clicker.py
View File

@ -147,10 +147,10 @@ class Clicker(object):
:return: None
"""
self._ele.page.run_cdp('Input.dispatchMouseEvent', type='mousePressed',
x=client_x, y=client_y, button=button, clickCount=count)
x=client_x, y=client_y, button=button, clickCount=count, _timeout=0.3)
# sleep(.05)
self._ele.page.run_cdp('Input.dispatchMouseEvent', type='mouseReleased',
x=client_x, y=client_y, button=button)
x=client_x, y=client_y, button=button, _timeout=0.2)
# -------------即将废弃--------------