127 lines
4.7 KiB
Python
127 lines
4.7 KiB
Python
"""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, WebDriverException
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
|
|
from crawler.driver import human_delay
|
|
from logger import get_logger
|
|
|
|
log = get_logger(__name__)
|
|
|
|
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
|
|
|
|
|
|
def _is_web_url(url: str) -> bool:
|
|
parsed = urlsplit(url)
|
|
return parsed.scheme in {"http", "https"} and bool(parsed.netloc)
|
|
|
|
|
|
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.
|
|
|
|
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 = _normalise_url(url)
|
|
self.wait = WebDriverWait(driver, timeout)
|
|
|
|
def run(self) -> dict[str, object]:
|
|
"""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:
|
|
# Chrome's internal New Tab URLs (chrome://newtab and
|
|
# chrome://new-tab-page) are not successful navigation. Wait until
|
|
# the address bar actually contains an HTTP(S) page or redirect.
|
|
self.wait.until(lambda d: _is_web_url(d.current_url))
|
|
except TimeoutException as exc:
|
|
try:
|
|
current_url = self.driver.current_url
|
|
except WebDriverException:
|
|
current_url = "<unavailable>"
|
|
raise FreshSessionError(
|
|
f"navigation to {self.url} never reached a web page "
|
|
f"(Chrome remained at {current_url})"
|
|
) 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.
|
|
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 = cast(
|
|
list[dict[str, Any]],
|
|
self.driver.get_cookies(), # type: ignore[reportUnknownMemberType]
|
|
)
|
|
log.info("Browser reached %s", current_url)
|
|
|
|
return {
|
|
"title": title,
|
|
"url": current_url,
|
|
"cookie_count": len(cookies),
|
|
"cookies": cookies,
|
|
}
|