""" Dynamic flow executor — reads the active flow config from DB at runtime and runs: 1. search_step : type each search text into the search box 2. scroll_step : infinite-scroll until target item is visible or max_scrolls reached 3. click_step : click each target item id """ from __future__ import annotations import asyncio import time from dataclasses import dataclass, field from typing import Any import undetected_chromedriver as uc from selenium.common.exceptions import NoSuchElementException, TimeoutException from selenium.webdriver.common.action_chains import ActionChains from selenium.webdriver.common.by import By from selenium.webdriver.common.keys import Keys from selenium.webdriver.remote.webelement import WebElement from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support.ui import WebDriverWait from crawler.driver import human_delay class DynamicFlowError(Exception): pass @dataclass class StepResult: name: str success: bool data: dict[str, Any] = field(default_factory=dict) error: str | None = None duration_ms: int = 0 @dataclass class DynamicFlowResult: config_id: int config_name: str steps: list[StepResult] = field(default_factory=list) @property def success(self) -> bool: return all(s.success for s in self.steps) @property def failure_reason(self) -> str | None: for s in self.steps: if not s.success: return f"{s.name}: {s.error}" return None def _by(selector_type: str) -> str: return By.XPATH if selector_type.lower() == "xpath" else By.CSS_SELECTOR def _find(driver: uc.Chrome, selector: str, selector_type: str) -> WebElement: return driver.find_element(_by(selector_type), selector) def _find_all(driver: uc.Chrome, selector: str, selector_type: str) -> list[WebElement]: return driver.find_elements(_by(selector_type), selector) def _wait_for( driver: uc.Chrome, selector: str, selector_type: str, timeout: float = 15.0, ) -> WebElement: return WebDriverWait(driver, timeout).until( EC.presence_of_element_located((_by(selector_type), selector)) ) class DynamicFlow: def __init__(self, driver: uc.Chrome, cfg: dict[str, Any]) -> None: self.driver = driver self.cfg = cfg def run(self) -> DynamicFlowResult: result = DynamicFlowResult( config_id=self.cfg["id"], config_name=self.cfg["name"], ) search_texts: list[str] = self.cfg.get("search_texts_json") or [] item_ids: list[str] = self.cfg.get("target_item_ids_json") or [] for text in search_texts: step = self._run_search(text) result.steps.append(step) if not step.success: return result human_delay(0.8, 1.5) for item_id in item_ids: scroll_step = self._run_scroll_and_find(item_id) result.steps.append(scroll_step) if not scroll_step.success: continue # try next item_id, don't abort the whole flow click_step = self._run_click(item_id, scroll_step.data.get("element")) result.steps.append(click_step) human_delay(1.0, 2.5) return result # ------------------------------------------------------------------ # Step: type into search box # ------------------------------------------------------------------ def _run_search(self, text: str) -> StepResult: t0 = time.monotonic() name = f"search:{text}" try: sel = self.cfg["search_box_selector"] sel_type = self.cfg["search_box_selector_type"] if not sel: raise DynamicFlowError("search_box_selector is not configured") box = _wait_for(self.driver, sel, sel_type) box.clear() human_delay(0.3, 0.6) # Type character by character for a human feel for ch in text: box.send_keys(ch) time.sleep(0.04) human_delay(0.4, 0.8) # Submit: explicit button or Enter submit_sel = self.cfg.get("search_submit_selector", "") if submit_sel: submit_type = self.cfg.get("search_submit_selector_type", "css") btn = _wait_for(self.driver, submit_sel, submit_type, timeout=5.0) btn.click() else: box.send_keys(Keys.RETURN) # Wait for results container if configured results_sel = self.cfg.get("search_results_selector", "") if results_sel: _wait_for(self.driver, results_sel, "css", timeout=15.0) human_delay(1.0, 2.0) return StepResult(name=name, success=True, data={"text": text}, duration_ms=int((time.monotonic() - t0) * 1000)) except Exception as exc: return StepResult(name=name, success=False, error=str(exc), duration_ms=int((time.monotonic() - t0) * 1000)) # ------------------------------------------------------------------ # Step: scroll (infinite-scroll) until item selector matches # ------------------------------------------------------------------ def _run_scroll_and_find(self, item_id: str) -> StepResult: t0 = time.monotonic() name = f"scroll_find:{item_id}" try: item_sel = self._build_item_selector(item_id) item_sel_type = self.cfg.get("item_selector_type", "css") scroll_container = self.cfg.get("scroll_container_selector", "").strip() max_scrolls: int = int(self.cfg.get("max_scrolls", 30)) pause_ms: int = int(self.cfg.get("scroll_pause_ms", 1200)) no_new_ms: int = int(self.cfg.get("no_new_content_timeout_ms", 3000)) # Check if already visible before scrolling found = self._find_item_visible(item_sel, item_sel_type) if found: return StepResult(name=name, success=True, data={"item_id": item_id, "scrolls": 0, "element": found}, duration_ms=int((time.monotonic() - t0) * 1000)) last_height = self._get_scroll_height(scroll_container) stale_since: float | None = None for scroll_num in range(1, max_scrolls + 1): self._scroll_down(scroll_container) time.sleep(pause_ms / 1000) found = self._find_item_visible(item_sel, item_sel_type) if found: return StepResult(name=name, success=True, data={"item_id": item_id, "scrolls": scroll_num, "element": found}, duration_ms=int((time.monotonic() - t0) * 1000)) new_height = self._get_scroll_height(scroll_container) if new_height == last_height: if stale_since is None: stale_since = time.monotonic() elif (time.monotonic() - stale_since) * 1000 >= no_new_ms: raise DynamicFlowError( f"No new content after {no_new_ms} ms — reached end of page " f"without finding item '{item_id}'" ) else: stale_since = None last_height = new_height raise DynamicFlowError( f"Item '{item_id}' not found after {max_scrolls} scrolls" ) except DynamicFlowError: raise except Exception as exc: return StepResult(name=name, success=False, error=str(exc), duration_ms=int((time.monotonic() - t0) * 1000)) # ------------------------------------------------------------------ # Step: click the found element # ------------------------------------------------------------------ def _run_click(self, item_id: str, element: Any) -> StepResult: t0 = time.monotonic() name = f"click:{item_id}" try: if element is None: raise DynamicFlowError("No element reference from scroll step") el: WebElement = element # Scroll element into view and click self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", el) human_delay(0.3, 0.7) try: el.click() except Exception: # Fallback: JS click self.driver.execute_script("arguments[0].click();", el) human_delay(0.8, 1.5) return StepResult(name=name, success=True, data={"item_id": item_id, "url_after": self.driver.current_url}, duration_ms=int((time.monotonic() - t0) * 1000)) except Exception as exc: return StepResult(name=name, success=False, error=str(exc), duration_ms=int((time.monotonic() - t0) * 1000)) # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ def _build_item_selector(self, item_id: str) -> str: template: str = self.cfg.get("item_selector_template", "") if not template: raise DynamicFlowError("item_selector_template is not configured") return template.replace("{item_id}", item_id) def _find_item_visible(self, selector: str, selector_type: str) -> WebElement | None: try: els = _find_all(self.driver, selector, selector_type) for el in els: if el.is_displayed(): return el except NoSuchElementException: pass return None def _get_scroll_height(self, container_sel: str) -> int: if container_sel: try: el = _find(self.driver, container_sel, "css") return int(self.driver.execute_script("return arguments[0].scrollHeight", el)) except Exception: pass return int(self.driver.execute_script("return document.body.scrollHeight")) def _scroll_down(self, container_sel: str) -> None: if container_sel: try: el = _find(self.driver, container_sel, "css") self.driver.execute_script( "arguments[0].scrollTop = arguments[0].scrollHeight", el ) return except Exception: pass self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight)")