Navigate before initializing page state
This commit is contained in:
@@ -1,8 +1,11 @@
|
||||
"""Scenario 2 — fresh cookie jar per run so the site treats the visitor as a new user."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, cast
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
import undetected_chromedriver as uc
|
||||
from selenium.common.exceptions import TimeoutException
|
||||
from selenium.common.exceptions import TimeoutException, WebDriverException
|
||||
from selenium.webdriver.support.ui import WebDriverWait
|
||||
|
||||
from crawler.driver import human_delay
|
||||
@@ -10,56 +13,100 @@ from logger import get_logger
|
||||
|
||||
log = get_logger(__name__)
|
||||
|
||||
_BLANK_URLS = {"", "about:blank", "data:,"}
|
||||
|
||||
|
||||
class FreshSessionError(RuntimeError):
|
||||
"""The browser could not leave its initial blank page."""
|
||||
|
||||
|
||||
def _normalise_url(url: str) -> str:
|
||||
value = url.strip()
|
||||
if value and "://" not in value:
|
||||
value = f"https://{value}"
|
||||
parsed = urlsplit(value)
|
||||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
||||
raise FreshSessionError(
|
||||
"target_url must be a complete HTTP(S) URL, for example https://example.com"
|
||||
)
|
||||
return value
|
||||
|
||||
|
||||
class FreshSessionScenario:
|
||||
"""
|
||||
Opens the target URL in a brand-new browser profile (no stored cookies,
|
||||
localStorage, or cache) so each run appears as a first-time visitor.
|
||||
|
||||
Because undetected-chromedriver already creates an isolated temp profile
|
||||
per instance, simply spinning up a new driver is sufficient — but we also
|
||||
explicitly delete all cookies after loading to be safe.
|
||||
Undetected-chromedriver creates an isolated temporary profile per instance,
|
||||
so no cookie or storage clearing is needed before navigation.
|
||||
"""
|
||||
|
||||
def __init__(self, driver: uc.Chrome, url: str, timeout: int = 30) -> None:
|
||||
self.driver = driver
|
||||
self.url = url
|
||||
self.url = _normalise_url(url)
|
||||
self.wait = WebDriverWait(driver, timeout)
|
||||
|
||||
def run(self) -> dict[str, object]:
|
||||
"""Navigate to the target, clear any residual state, return page info."""
|
||||
self.driver.delete_all_cookies()
|
||||
|
||||
# Clear localStorage / sessionStorage via JS
|
||||
self.driver.execute_script(
|
||||
"try { window.localStorage.clear(); window.sessionStorage.clear(); } catch(e) {}"
|
||||
)
|
||||
|
||||
if not self.url:
|
||||
return {"title": "", "url": "", "cookie_count": 0, "cookies": []}
|
||||
"""Navigate to the target and return page information."""
|
||||
# Each driver already owns a brand-new temporary profile. Do not issue
|
||||
# renderer commands on Chrome's initial blank tab: on small servers that
|
||||
# renderer can stall before navigation is ever attempted.
|
||||
log.info("Navigating browser to %s", self.url)
|
||||
try:
|
||||
navigation = cast(
|
||||
dict[str, Any],
|
||||
self.driver.execute_cdp_cmd( # type: ignore[reportUnknownMemberType]
|
||||
"Page.navigate", {"url": self.url}
|
||||
),
|
||||
)
|
||||
error_text = navigation.get("errorText")
|
||||
if error_text:
|
||||
raise FreshSessionError(f"Chrome rejected navigation: {error_text}")
|
||||
except FreshSessionError:
|
||||
raise
|
||||
except WebDriverException:
|
||||
# Older Chrome builds may not expose Page.navigate. Keep a normal
|
||||
# WebDriver fallback, but only after navigation has been attempted.
|
||||
try:
|
||||
self.driver.get(self.url)
|
||||
except TimeoutException:
|
||||
log.warning("Page load timed out; stopping navigation")
|
||||
self.driver.execute_cdp_cmd( # type: ignore[reportUnknownMemberType]
|
||||
"Page.stopLoading", {}
|
||||
)
|
||||
|
||||
try:
|
||||
self.driver.get(self.url)
|
||||
except TimeoutException:
|
||||
log.warning(
|
||||
"Page load exceeded the browser timeout; stopping navigation and continuing"
|
||||
)
|
||||
self.driver.execute_cdp_cmd("Page.stopLoading", {})
|
||||
human_delay(2.0, 4.0)
|
||||
self.wait.until(lambda d: d.current_url not in _BLANK_URLS)
|
||||
except TimeoutException as exc:
|
||||
raise FreshSessionError(
|
||||
f"navigation to {self.url} never left Chrome's blank page"
|
||||
) from exc
|
||||
|
||||
human_delay(1.0, 2.0)
|
||||
|
||||
# Interactive is sufficient for the dynamic flow and avoids waiting on
|
||||
# analytics, ads, or other background resources indefinitely.
|
||||
try:
|
||||
self.wait.until(
|
||||
lambda d: d.execute_script("return document.readyState")
|
||||
in ("interactive", "complete")
|
||||
def document_is_ready(driver: uc.Chrome) -> bool:
|
||||
state = cast(
|
||||
str,
|
||||
driver.execute_script( # type: ignore[reportUnknownMemberType]
|
||||
"return document.readyState"
|
||||
),
|
||||
)
|
||||
return state in ("interactive", "complete")
|
||||
|
||||
try:
|
||||
self.wait.until(document_is_ready)
|
||||
except TimeoutException:
|
||||
log.warning("Document did not report a ready state; continuing with the current DOM")
|
||||
|
||||
title = self.driver.title
|
||||
current_url = self.driver.current_url
|
||||
cookies = self.driver.get_cookies()
|
||||
cookies = cast(
|
||||
list[dict[str, Any]],
|
||||
self.driver.get_cookies(), # type: ignore[reportUnknownMemberType]
|
||||
)
|
||||
log.info("Browser reached %s", current_url)
|
||||
|
||||
return {
|
||||
"title": title,
|
||||
|
||||
Reference in New Issue
Block a user