init

crawler/scenarios/fresh_session.py

@@ -0,0 +1,52 @@
+"""Scenario 2 — fresh cookie jar per run so the site treats the visitor as a new user."""
+from __future__ import annotations
+
+import undetected_chromedriver as uc
+from selenium.common.exceptions import TimeoutException
+from selenium.webdriver.support.ui import WebDriverWait
+
+from config import config
+from crawler.driver import human_delay, make_driver
+
+
+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.
+    """
+
+    def __init__(self, driver: uc.Chrome, timeout: int = 30) -> None:
+        self.driver = driver
+        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) {}"
+        )
+
+        self.driver.get(config.target_url)
+        human_delay(2.0, 4.0)
+
+        # Wait for the page to reach a ready state
+        self.wait.until(
+            lambda d: d.execute_script("return document.readyState") == "complete"
+        )
+
+        title = self.driver.title
+        current_url = self.driver.current_url
+        cookies = self.driver.get_cookies()
+
+        return {
+            "title": title,
+            "url": current_url,
+            "cookie_count": len(cookies),
+            "cookies": cookies,
+        }

crawler/scenarios/otp_login.py

@@ -0,0 +1,106 @@
+"""Scenario 1 — SMS-OTP login with a batch of phone numbers."""
+from __future__ import annotations
+
+import time
+from typing import Callable
+
+import undetected_chromedriver as uc
+from selenium.common.exceptions import TimeoutException
+from selenium.webdriver.common.by import By
+from selenium.webdriver.support import expected_conditions as EC
+from selenium.webdriver.support.ui import WebDriverWait
+
+from config import config
+from crawler.driver import human_delay
+
+
+class OTPLoginError(Exception):
+    pass
+
+
+class OTPLoginScenario:
+    """
+    Drives an SMS-OTP login flow.
+
+    The caller supplies an `otp_resolver` — a callable that receives the
+    phone number and returns the OTP string.  Wire it up to your SMS
+    gateway / SIM-management API.
+    """
+
+    def __init__(
+        self,
+        driver: uc.Chrome,
+        phone: str,
+        otp_resolver: Callable[[str], str],
+        timeout: int = 30,
+    ) -> None:
+        self.driver = driver
+        self.phone = phone
+        self.otp_resolver = otp_resolver
+        self.wait = WebDriverWait(driver, timeout)
+
+    # ------------------------------------------------------------------
+    # Override these selectors to match the target site's actual HTML.
+    # ------------------------------------------------------------------
+    _PHONE_FIELD_SELECTOR = (By.CSS_SELECTOR, "input[type='tel'], input[name='phone']")
+    _SEND_OTP_BTN_SELECTOR = (By.CSS_SELECTOR, "button[type='submit']")
+    _OTP_FIELD_SELECTOR = (By.CSS_SELECTOR, "input[name='otp'], input[autocomplete='one-time-code']")
+    _CONFIRM_BTN_SELECTOR = (By.CSS_SELECTOR, "button[type='submit']")
+    _SUCCESS_INDICATOR = (By.CSS_SELECTOR, "[data-testid='home'], .dashboard, #main-content")
+
+    def run(self) -> dict[str, object]:
+        """Execute the OTP login. Returns a result dict."""
+        self.driver.get(config.target_url)
+        human_delay(1.5, 3.0)
+
+        try:
+            self._enter_phone()
+            self._request_otp()
+            otp = self._resolve_otp()
+            self._enter_otp(otp)
+            self._confirm_login()
+            self._wait_for_success()
+        except TimeoutException as exc:
+            raise OTPLoginError(f"Timeout during OTP login for {self.phone}") from exc
+
+        return {"phone": self.phone, "cookies": self.driver.get_cookies()}
+
+    def _enter_phone(self) -> None:
+        field = self.wait.until(EC.element_to_be_clickable(self._PHONE_FIELD_SELECTOR))
+        field.clear()
+        human_delay(0.3, 0.7)
+        for char in self.phone:
+            field.send_keys(char)
+            time.sleep(0.05)
+
+    def _request_otp(self) -> None:
+        btn = self.wait.until(EC.element_to_be_clickable(self._SEND_OTP_BTN_SELECTOR))
+        human_delay(0.5, 1.2)
+        btn.click()
+
+    def _resolve_otp(self) -> str:
+        # Poll for the OTP from the external resolver (SMS gateway / webhook)
+        for attempt in range(12):
+            human_delay(5.0, 8.0)
+            otp = self.otp_resolver(self.phone)
+            if otp:
+                return otp
+            if attempt == 11:
+                raise OTPLoginError(f"OTP not received for {self.phone} after 12 attempts")
+        return ""  # unreachable but satisfies type checker
+
+    def _enter_otp(self, otp: str) -> None:
+        field = self.wait.until(EC.element_to_be_clickable(self._OTP_FIELD_SELECTOR))
+        field.clear()
+        human_delay(0.3, 0.7)
+        for char in otp:
+            field.send_keys(char)
+            time.sleep(0.08)
+
+    def _confirm_login(self) -> None:
+        btn = self.wait.until(EC.element_to_be_clickable(self._CONFIRM_BTN_SELECTOR))
+        human_delay(0.5, 1.0)
+        btn.click()
+
+    def _wait_for_success(self) -> None:
+        self.wait.until(EC.presence_of_element_located(self._SUCCESS_INDICATOR))