crawle-snapp/crawler/scenarios/digipay.py

"""DigiPay-specific flow — search + find via onclick goToProduct URL."""
from __future__ import annotations

import time
from typing import Any

import undetected_chromedriver as uc
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait

from crawler.driver import human_delay
from crawler.dynamic_flow import DynamicFlow, StepResult, _wait_for
from logger import get_logger

log = get_logger(__name__)

_RESULTS_WAIT_CSS = "a[onclick*='goToProduct']"

# Ordered list of selectors tried to locate the search input
_SEARCH_INPUT_CANDIDATES = [
    (By.XPATH,  "/html/body/header/header/div[2]//input"),
    (By.XPATH,  "//header//input[@type='search']"),
    (By.XPATH,  "//header//input[contains(@class,'search')]"),
    (By.XPATH,  "//input[@type='search']"),
    (By.CSS_SELECTOR, "input[type='search']"),
    (By.XPATH,  "//input[contains(@placeholder,'جست')]"),  # 'جست' = search (Farsi prefix)
]

# Selectors tried to trigger / open the search box before locating the input
_SEARCH_TRIGGER_CANDIDATES = [
    (By.XPATH, "/html/body/header/header/div[2]"),
    (By.XPATH, "//header//*[contains(@class,'search')]"),
    (By.CSS_SELECTOR, "header [class*='search']"),
]


_CONTAINER_FIX_JS = """
(function() {
  function inject() {
    if (document.getElementById('__seed_fix')) return;
    var s = document.createElement('style');
    s.id = '__seed_fix';
    s.textContent = '.container{margin-left:0!important;margin-right:0!important}';
    document.head.appendChild(s);
  }
  inject();
  new MutationObserver(inject).observe(document.head, {childList: true});
  new MutationObserver(inject).observe(document.documentElement, {childList: true});
})();
"""


def _inject_container_fix(driver: uc.Chrome) -> None:
    driver.execute_script(_CONTAINER_FIX_JS)


def _wait_spa_ready(driver: uc.Chrome, timeout: float = 20.0) -> None:
    """Wait for DOM ready + Vue/React hydration (no pending network requests)."""
    # Force desktop viewport — --window-size sets outer chrome, not inner viewport
    driver.execute_cdp_cmd("Emulation.setDeviceMetricsOverride", {
        "width": 1920,
        "height": 1080,
        "deviceScaleFactor": 1,
        "mobile": False,
    })
    WebDriverWait(driver, timeout).until(
        lambda d: d.execute_script("return document.readyState") == "complete"
    )
    # Extra pause for SPA JS to hydrate
    time.sleep(1.5)
    _inject_container_fix(driver)


def _find_search_input(driver: uc.Chrome, wait: WebDriverWait) -> Any:
    """Try each candidate selector until one is clickable."""
    for by, sel in _SEARCH_INPUT_CANDIDATES:
        try:
            el = wait.until(EC.element_to_be_clickable((by, sel)))
            log.debug("Search input found via [%s:%s]", by, sel)
            return el
        except Exception:
            continue
    raise RuntimeError("Search input not found — tried all candidate selectors")


class DigiPayFlow(DynamicFlow):
    """
    DigiPay-specific flow:
    - Waits for SPA hydration before interacting.
    - Tries multiple selectors to find/open search input.
    - Waits for `a[onclick*='goToProduct']` cards after submit.
    - Item matching via onclick URL fragment or product title text.
    """

    def _run_search(self, text: str) -> StepResult:
        t0 = time.monotonic()
        name = f"search:{text}"
        log.info("[%s] DigiPay search for '%s'", name, text)
        try:
            wait_long  = WebDriverWait(self.driver, 25)
            wait_short = WebDriverWait(self.driver, 5)

            # Wait for SPA to fully hydrate
            _wait_spa_ready(self.driver)
            log.debug("[%s] SPA ready", name)

            # Try to click a trigger element to open/reveal the search input
            for by, sel in _SEARCH_TRIGGER_CANDIDATES:
                try:
                    trigger = wait_short.until(EC.element_to_be_clickable((by, sel)))
                    trigger.click()
                    log.debug("[%s] Clicked trigger [%s:%s]", name, by, sel)
                    human_delay(0.4, 0.7)
                    break
                except Exception:
                    continue

            # Locate the actual input (may now be in an overlay)
            box = _find_search_input(self.driver, wait_long)
            box.clear()
            human_delay(0.2, 0.4)

            for ch in text:
                box.send_keys(ch)
                time.sleep(0.05)

            human_delay(0.3, 0.6)
            box.send_keys(Keys.RETURN)
            log.debug("[%s] Enter sent", name)

            # Wait for product cards
            wait_long.until(
                EC.presence_of_element_located((By.CSS_SELECTOR, _RESULTS_WAIT_CSS))
            )
            _inject_container_fix(self.driver)
            human_delay(1.0, 2.0)

            ms = int((time.monotonic() - t0) * 1000)
            log.info("[%s] Search OK — %d ms", name, ms)
            return StepResult(name=name, success=True, data={"text": text}, duration_ms=ms)

        except Exception as exc:
            ms = int((time.monotonic() - t0) * 1000)
            log.error("[%s] Failed — %s", name, exc)
            return StepResult(name=name, success=False, error=str(exc), duration_ms=ms)

    def _build_item_selector(self, item_id: str) -> tuple[str, str]:
        sel_type: str = self.cfg.get("item_selector_type", "onclick")
        template: str = self.cfg.get("item_selector_template", "")
        fragment = template.replace("{item_id}", item_id) if template else item_id

        if sel_type == "onclick":
            escaped = fragment.replace("'", "\\'")
            return f"//a[contains(@onclick, '{escaped}')]", "xpath"

        if sel_type == "text":
            escaped = fragment.replace("'", "\\'")
            return f"//a[.//*[contains(normalize-space(.), '{escaped}')]]", "xpath"

        return super()._build_item_selector(item_id)