Open target URL during Chrome startup

crawler/batch.py

@@ -35,7 +35,11 @@ def _single_run(
         target_url or "<missing>",
     )
     try:
-        with driver_session(headless=headless, recording_name=identifier) as driver:
+        with driver_session(
+            headless=headless,
+            recording_name=identifier,
+            initial_url=target_url,
+        ) as driver:
             FreshSessionScenario(driver, target_url).run()
             result = DynamicFlow(driver, cfg).run() if cfg else None
 

crawler/driver.py

@@ -13,9 +13,10 @@ import tempfile
 import time
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Generator
+from typing import Any, Generator, cast
 
 import undetected_chromedriver as uc
+from selenium.common.exceptions import WebDriverException
 from selenium.webdriver.chrome.options import Options
 from selenium_stealth import stealth
 
@@ -90,10 +91,13 @@ def _build_options(
     chrome_binary: str,
     profile_dir: str | None = None,
     display: str | None = None,
+    initial_url: str | None = None,
 ) -> Options:
     opts = Options()
     opts.binary_location = chrome_binary
-    opts.page_load_strategy = "eager"
+    # Never let ChromeDriver session creation block on the initial New Tab
+    # renderer. Scenarios perform their own explicit URL/readiness waits.
+    opts.page_load_strategy = "none"
     if headless:
         opts.add_argument("--headless=new")
     if profile_dir:
@@ -108,6 +112,10 @@ def _build_options(
     opts.add_argument("--disable-blink-features=AutomationControlled")
     opts.add_argument("--disable-infobars")
     opts.add_argument("--window-size=1920,1080")
+    if initial_url and initial_url.startswith(("http://", "https://")):
+        # Opening the configured page as a Chrome process argument avoids
+        # waiting for the initial New Tab renderer before navigation begins.
+        opts.add_argument(initial_url)
     return opts
 
 
@@ -115,14 +123,27 @@ def make_driver(
     headless: bool | None = None,
     profile_dir: str | None = None,
     display: str | None = None,
+    initial_url: str | None = None,
 ) -> uc.Chrome:
     """Return a stealthed undetected Chrome instance."""
     use_headless = config.headless if headless is None else headless
     chrome_binary, major = _detect_chrome()
     ua = random.choice(_USER_AGENTS)
-    opts = _build_options(ua, use_headless, chrome_binary, profile_dir, display)
+    opts = _build_options(
+        ua,
+        use_headless,
+        chrome_binary,
+        profile_dir,
+        display,
+        initial_url,
+    )
 
-    log.debug("Starting ChromeDriver (headless=%s, Chrome %d)", use_headless, major)
+    log.info(
+        "Starting ChromeDriver (headless=%s, Chrome %d, initial_url=%s)",
+        use_headless,
+        major,
+        initial_url or "<none>",
+    )
 
     driver = uc.Chrome(
         options=opts,
@@ -130,6 +151,23 @@ def make_driver(
         version_main=major,
         use_subprocess=True,
     )
+    log.info("ChromeDriver connected")
+
+    if initial_url and initial_url.startswith(("http://", "https://")):
+        navigation = cast(
+            dict[str, Any],
+            driver.execute_cdp_cmd(  # type: ignore[reportUnknownMemberType]
+                "Page.navigate", {"url": initial_url}
+            ),
+        )
+        error_text = navigation.get("errorText")
+        if error_text:
+            raise WebDriverException(
+                f"Chrome rejected initial navigation to {initial_url}: {error_text}"
+            )
+        log.info("Initial navigation dispatched to %s", initial_url)
+
+    log.info("Applying stealth settings")
 
     if display:
         driver.set_window_size(1920, 1080)
@@ -191,6 +229,7 @@ def _terminate_profile_processes(profile_dir: str) -> None:
 def driver_session(
     headless: bool | None = None,
     recording_name: str | None = None,
+    initial_url: str | None = None,
 ) -> Generator[uc.Chrome, None, None]:
     profile_dir = tempfile.mkdtemp(prefix="seed-chrome-")
     driver: uc.Chrome | None = None
@@ -207,6 +246,7 @@ def driver_session(
             effective_headless,
             profile_dir=profile_dir,
             display=display,
+            initial_url=initial_url,
         )
         yield driver
     finally:

crawler/scenarios/fresh_session.py

@@ -13,9 +13,6 @@ 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."""
 
@@ -32,6 +29,11 @@ def _normalise_url(url: str) -> str:
     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,
@@ -76,10 +78,18 @@ class FreshSessionScenario:
                 )
 
         try:
-            self.wait.until(lambda d: d.current_url not in _BLANK_URLS)
+            # 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 left Chrome's blank page"
+                f"navigation to {self.url} never reached a web page "
+                f"(Chrome remained at {current_url})"
             ) from exc
 
         human_delay(1.0, 2.0)

crawler/tasks.py

@@ -314,7 +314,11 @@ def run_batch_job(batch_id: str, item: dict[str, Any]) -> None:
         )
 
         try:
-            with driver_session(headless=headless, recording_name=identifier) as driver:
+            with driver_session(
+                headless=headless,
+                recording_name=identifier,
+                initial_url=target_url,
+            ) as driver:
                 FreshSessionScenario(driver, target_url).run()
                 if cfg:
                     if scenario == "digipay":

main.py

@@ -73,7 +73,7 @@ def run_otp(phone: str, cfg_id: int | None = None) -> None:
     target_url: str = flow_cfg.get("target_url") or "" if flow_cfg else ""
     t0 = time.monotonic()
     try:
-        with driver_session() as driver:
+        with driver_session(initial_url=target_url) as driver:
             OTPLoginScenario(driver, phone, otp_resolver, target_url).run()
             result = DynamicFlow(driver, flow_cfg).run() if flow_cfg else None
     except Exception as exc:
@@ -91,7 +91,7 @@ def run_fresh(cfg_id: int | None = None) -> None:
     target_url: str = flow_cfg.get("target_url") or "" if flow_cfg else ""
     t0 = time.monotonic()
     try:
-        with driver_session() as driver:
+        with driver_session(initial_url=target_url) as driver:
             FreshSessionScenario(driver, target_url).run()
             result = DynamicFlow(driver, flow_cfg).run() if flow_cfg else None
     except Exception as exc: