继续修改抓包功能,未完成

This commit is contained in:
g1879 2023-04-26 19:01:00 +08:00
parent 283bbfe06f
commit f817fcee5c
3 changed files with 22 additions and 7 deletions

View File

@ -1063,6 +1063,7 @@ class NetworkListener(object):
self._count = None
self._caught = 0 # 已获取到的数量
self._all_tabs = False # 是否监听所有tab
def set_targets(self, targets, is_regex=False, count=None):
"""指定要等待的数据包
@ -1092,12 +1093,23 @@ class NetworkListener(object):
else:
self.stop()
def start(self):
driver = self._page.browser_driver if self._all_tabs else self._page.driver
driver.set_listener('Fetch.requestPaused', self._request_paused)
patterns = []
for i in self._targets:
patterns.append({'requestStage': 'Request', 'urlPattern': i})
patterns.append({'requestStage': 'Response', 'urlPattern': i})
if patterns:
driver.call_method('Fetch.enable', patterns=patterns)
else:
driver.call_method('Fetch.enable')
def stop(self):
"""停止监听数据包"""
self._page.run_cdp('Network.disable')
self._page.driver.Network.requestWillBeSent = None
self._page.driver.Network.responseReceived = None
self._page.driver.Network.loadingFinished = None
driver = self._page.browser_driver if self._all_tabs else self._page.driver
driver.set_listener('Fetch.requestPaused', None)
driver.call_method('Fetch.disable')
def listen(self, timeout=None, any_one=False):
"""等待指定数据包加载完成
@ -1126,6 +1138,9 @@ class NetworkListener(object):
self._caught = 0
return r
def _request_paused(self, **kwargs):
pass
def _request_will_sent(self, **kwargs):
"""接收到请求时的回调函数"""
for target in self._targets:

View File

@ -53,7 +53,7 @@ class ChromiumDriver(object):
def stop(self) -> bool: ...
def set_listener(self, event: str, callback: Callable) -> Union[Callable, None, bool]: ...
def set_listener(self, event: str, callback: Union[Callable, None]) -> Union[Callable, None, bool]: ...
def get_listener(self, event: str) -> Union[Callable, None]: ...

View File

@ -26,8 +26,8 @@ class WebPage(SessionPage, ChromiumPage, BasePage):
def __init__(self,
mode: str = 'd',
timeout: float = None,
driver_or_options: Union[ChromiumDriver, ChromiumOptions, DriverOptions] = None,
session_or_options: Union[Session, SessionOptions] = None) -> None:
driver_or_options: Union[ChromiumDriver, ChromiumOptions, DriverOptions, bool] = None,
session_or_options: Union[Session, SessionOptions, bool] = None) -> None:
self._mode: str = ...
self._has_driver: bool = ...
self._has_session: bool = ...