Digipay and speedtest added

2026-06-27 16:18:33 +03:30
parent 64886b2b10
commit 744550deaa
8 changed files with 421 additions and 16 deletions
@@ -0,0 +1,133 @@
+"""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']"),
+]
+
+
+def _wait_spa_ready(driver: uc.Chrome, timeout: float = 20.0) -> None:
+    """Wait for DOM ready + Vue/React hydration (no pending network requests)."""
+    WebDriverWait(driver, timeout).until(
+        lambda d: d.execute_script("return document.readyState") == "complete"
+    )
+    # Extra pause for SPA JS to hydrate
+    time.sleep(1.5)
+
+
+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))
+            )
+            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)