From c19c91bb8169378e13852eb7b13807e504035268 Mon Sep 17 00:00:00 2001 From: winet Date: Sun, 17 Dec 2023 19:57:44 +1000 Subject: [PATCH] Working devtools response catch --- stolichki/browser.py | 53 +++++++++++++++++++++++++++++++++++--------- 1 file changed, 42 insertions(+), 11 deletions(-) diff --git a/stolichki/browser.py b/stolichki/browser.py index 6822cb6..be40ccb 100644 --- a/stolichki/browser.py +++ b/stolichki/browser.py @@ -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 - - except (NoSuchElementException, ElementNotVisibleException): - return False + logging.info("Loading element was founded") + return True + + 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")