283 lines
11 KiB
Python
283 lines
11 KiB
Python
"""
|
|
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)")
|