Working devtools response catch
This commit is contained in:
parent
6783437582
commit
c19c91bb81
|
@ -2,6 +2,8 @@ import time
|
|||
import logging
|
||||
import uuid
|
||||
import os
|
||||
import json
|
||||
import re
|
||||
|
||||
from selenium import webdriver
|
||||
from selenium.webdriver.chrome.options import Options
|
||||
|
@ -28,6 +30,9 @@ class StolichkiDriver(webdriver.Chrome):
|
|||
self, options: Options = None, service: Service = None, keep_alive: bool = True
|
||||
) -> None:
|
||||
|
||||
self.last_resp_index = 0
|
||||
self.last_resp_url = ""
|
||||
|
||||
# assert os.environ.get("TWOCAPTCA_KEY") is not None, "Can't fins environment variable TWOCAPTCHA_KEY"
|
||||
|
||||
if options is None:
|
||||
|
@ -42,6 +47,8 @@ class StolichkiDriver(webdriver.Chrome):
|
|||
options.add_experimental_option("useAutomationExtension", False)
|
||||
options.page_load_strategy = "eager"
|
||||
|
||||
options.capabilities["goog:loggingPrefs"] = {"performance": "ALL"}
|
||||
|
||||
# self.__solver = TwoCaptcha(os.environ.get("TWOCAPTCA_KEY"))
|
||||
|
||||
super().__init__(options, service, keep_alive)
|
||||
|
@ -92,20 +99,20 @@ class StolichkiDriver(webdriver.Chrome):
|
|||
|
||||
|
||||
def __wait_for_presence(self, xpath: str, delay: int = 60):
|
||||
try:
|
||||
wait = WebDriverWait(self, delay)
|
||||
try:
|
||||
wait = WebDriverWait(self, delay)
|
||||
|
||||
wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, xpath)
|
||||
)
|
||||
wait.until(
|
||||
EC.presence_of_element_located(
|
||||
(By.XPATH, xpath)
|
||||
)
|
||||
)
|
||||
|
||||
logging.info("Loading element was founded")
|
||||
return True
|
||||
logging.info("Loading element was founded")
|
||||
return True
|
||||
|
||||
except (NoSuchElementException, ElementNotVisibleException):
|
||||
return False
|
||||
except (NoSuchElementException, ElementNotVisibleException):
|
||||
return False
|
||||
|
||||
def __edit_cookie(self, name: str, value):
|
||||
cookie = self.get_cookie(name)
|
||||
|
@ -117,6 +124,30 @@ class StolichkiDriver(webdriver.Chrome):
|
|||
|
||||
self.add_cookie(new_cookie)
|
||||
|
||||
def get_network_response(self, url_mask):
|
||||
logs = self.get_log("performance")[self.last_resp_index:]
|
||||
url_mask = re.compile(url_mask)
|
||||
for log in filter(lambda log: self.__filter_logs(log), logs):
|
||||
message = json.loads(log.get("message"))['message']
|
||||
|
||||
request_id = message["params"]["requestId"]
|
||||
resp_url = message["params"]["response"]["url"]
|
||||
|
||||
if re.fullmatch(url_mask, resp_url) and self.last_resp_url != resp_url:
|
||||
self.last_resp_index = logs.index(log)
|
||||
self.last_resp_url = resp_url
|
||||
body = self.execute_cdp_cmd("Network.getResponseBody", {"requestId": request_id})
|
||||
return json.loads(body['body'])
|
||||
|
||||
return None
|
||||
|
||||
def __filter_logs(self, log):
|
||||
message = json.loads(log.get("message"))['message']
|
||||
return (
|
||||
message.get("method") == "Network.responseReceived"
|
||||
and "json" in message["params"]["response"]["mimeType"]
|
||||
)
|
||||
|
||||
def __handle_captcha(self) -> None:
|
||||
for attempt in range(5):
|
||||
logging.info(f"Trying to solve captcha {attempt + 1}/5")
|
||||
|
|
Loading…
Reference in New Issue