702 lines
28 KiB
Python
702 lines
28 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 math
|
|
import random
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
import undetected_chromedriver as uc
|
|
from selenium.common.exceptions import NoSuchElementException
|
|
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
|
|
from logger import get_logger
|
|
|
|
log = get_logger(__name__)
|
|
|
|
|
|
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,
|
|
label: str = "",
|
|
) -> WebElement:
|
|
log.debug(
|
|
"Waiting for %s [%s:%s] (timeout=%ss)",
|
|
label or "element",
|
|
selector_type,
|
|
selector,
|
|
timeout,
|
|
)
|
|
try:
|
|
el = WebDriverWait(driver, timeout).until(
|
|
EC.presence_of_element_located((_by(selector_type), selector))
|
|
)
|
|
log.debug("Found %s [%s:%s]", label or "element", selector_type, selector)
|
|
return el
|
|
except Exception as exc:
|
|
log.warning(
|
|
"Selector not found — %s [%s:%s]: %s",
|
|
label or "element",
|
|
selector_type,
|
|
selector,
|
|
exc,
|
|
)
|
|
raise
|
|
|
|
|
|
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"],
|
|
)
|
|
log.info("Starting flow '%s' (id=%s)", self.cfg["name"], self.cfg["id"])
|
|
|
|
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:
|
|
log.error("Search step failed for '%s' — aborting flow", text)
|
|
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:
|
|
log.warning(
|
|
"Scroll/find failed for item '%s' — skipping click", item_id
|
|
)
|
|
continue
|
|
|
|
click_step = self._run_click(item_id, scroll_step.data.get("element"))
|
|
result.steps.append(click_step)
|
|
if click_step.success and self.cfg.get("stop_on_first_click", 0):
|
|
log.info("Item '%s' clicked — stopping flow", item_id)
|
|
return result
|
|
human_delay(1.0, 2.5)
|
|
|
|
log.info(
|
|
"Flow '%s' finished — success=%s, steps=%d",
|
|
self.cfg["name"],
|
|
result.success,
|
|
len(result.steps),
|
|
)
|
|
return result
|
|
|
|
# ------------------------------------------------------------------
|
|
# Step: type into search box
|
|
# ------------------------------------------------------------------
|
|
def _run_search(self, text: str) -> StepResult:
|
|
t0 = time.monotonic()
|
|
name = f"search:{text}"
|
|
log.info("[%s] Searching for '%s'", name, 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, label="search box")
|
|
box.clear()
|
|
human_delay(0.3, 0.6)
|
|
|
|
for ch in text:
|
|
box.send_keys(ch)
|
|
time.sleep(0.04)
|
|
log.debug("[%s] Typed search text", name)
|
|
|
|
human_delay(0.4, 0.8)
|
|
|
|
submit_sel = self.cfg.get("search_submit_selector", "")
|
|
if submit_sel:
|
|
submit_type = self.cfg.get("search_submit_selector_type", "css")
|
|
log.debug(
|
|
"[%s] Clicking submit button [%s:%s]", name, submit_type, submit_sel
|
|
)
|
|
btn = _wait_for(
|
|
self.driver,
|
|
submit_sel,
|
|
submit_type,
|
|
timeout=5.0,
|
|
label="submit button",
|
|
)
|
|
btn.click()
|
|
else:
|
|
log.debug("[%s] Submitting via Enter key", name)
|
|
box.send_keys(Keys.RETURN)
|
|
|
|
results_sel = self.cfg.get("search_results_selector", "")
|
|
results_sel_type = self.cfg.get("search_results_selector_type", "css")
|
|
if results_sel:
|
|
log.debug(
|
|
"[%s] Waiting for results container [%s:%s]",
|
|
name,
|
|
results_sel_type,
|
|
results_sel,
|
|
)
|
|
_wait_for(
|
|
self.driver,
|
|
results_sel,
|
|
results_sel_type,
|
|
timeout=15.0,
|
|
label="results container",
|
|
)
|
|
|
|
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)
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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}"
|
|
log.info("[%s] Scrolling to find item '%s'", name, item_id)
|
|
try:
|
|
item_sel, item_sel_type = self._build_item_selector(item_id)
|
|
scroll_container = self.cfg.get("scroll_container_selector", "").strip()
|
|
scroll_container_type = self.cfg.get(
|
|
"scroll_container_selector_type", "css"
|
|
)
|
|
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))
|
|
pagination_sel: str = self.cfg.get("pagination_selector", "").strip()
|
|
pagination_sel_type: str = self.cfg.get("pagination_selector_type", "css")
|
|
max_pages: int = int(self.cfg.get("max_pages", 10))
|
|
pagination_wait_ms: int = int(self.cfg.get("pagination_wait_ms", 1500))
|
|
|
|
log.debug(
|
|
"[%s] Selector [%s:%s], container=[%s:'%s'], max_scrolls=%d, pagination=%s",
|
|
name,
|
|
item_sel_type,
|
|
item_sel,
|
|
scroll_container_type,
|
|
scroll_container or "window",
|
|
max_scrolls,
|
|
(
|
|
f"[{pagination_sel_type}:{pagination_sel}] max_pages={max_pages}"
|
|
if pagination_sel
|
|
else "off"
|
|
),
|
|
)
|
|
|
|
found = self._find_item_visible(item_sel, item_sel_type)
|
|
if found:
|
|
log.info("[%s] Item visible immediately (no scroll needed)", name)
|
|
return StepResult(
|
|
name=name,
|
|
success=True,
|
|
data={
|
|
"item_id": item_id,
|
|
"scrolls": 0,
|
|
"pages": 0,
|
|
"element": found,
|
|
},
|
|
duration_ms=int((time.monotonic() - t0) * 1000),
|
|
)
|
|
|
|
last_height = self._get_scroll_height(
|
|
scroll_container, scroll_container_type
|
|
)
|
|
stale_since: float | None = None
|
|
total_scrolls = 0
|
|
page_num = 0
|
|
scroll_on_page = 0
|
|
|
|
while scroll_on_page < max_scrolls:
|
|
# Check for the item BEFORE scrolling so items near the top of a
|
|
# freshly-paginated page are seen before we jump to the bottom.
|
|
found = self._find_item_visible(item_sel, item_sel_type)
|
|
if found:
|
|
ms = int((time.monotonic() - t0) * 1000)
|
|
log.info(
|
|
"[%s] Found after %d scroll(s) on page %d — %d ms",
|
|
name,
|
|
total_scrolls,
|
|
page_num + 1,
|
|
ms,
|
|
)
|
|
return StepResult(
|
|
name=name,
|
|
success=True,
|
|
data={
|
|
"item_id": item_id,
|
|
"scrolls": total_scrolls,
|
|
"pages": page_num,
|
|
"element": found,
|
|
},
|
|
duration_ms=ms,
|
|
)
|
|
|
|
scroll_on_page += 1
|
|
total_scrolls += 1
|
|
self._scroll_down(scroll_container, scroll_container_type)
|
|
time.sleep(pause_ms / 1000)
|
|
|
|
new_height = self._get_scroll_height(
|
|
scroll_container, scroll_container_type
|
|
)
|
|
if new_height == last_height:
|
|
if stale_since is None:
|
|
stale_since = time.monotonic()
|
|
log.debug(
|
|
"[%s] Page height unchanged at scroll %d — starting stale timer",
|
|
name,
|
|
scroll_on_page,
|
|
)
|
|
elif (time.monotonic() - stale_since) * 1000 >= no_new_ms:
|
|
# Infinite scroll exhausted — try pagination if configured
|
|
if pagination_sel and page_num < max_pages:
|
|
if self._try_paginate(
|
|
name,
|
|
pagination_sel,
|
|
pagination_sel_type,
|
|
pagination_wait_ms,
|
|
):
|
|
page_num += 1
|
|
scroll_on_page = 0
|
|
stale_since = None
|
|
last_height = self._get_scroll_height(
|
|
scroll_container, scroll_container_type
|
|
)
|
|
log.info(
|
|
"[%s] Advanced to pagination page %d",
|
|
name,
|
|
page_num + 1,
|
|
)
|
|
# Loop continues — item check happens at top of next iteration
|
|
else:
|
|
raise DynamicFlowError(
|
|
f"Pagination button [{pagination_sel_type}:{pagination_sel}] "
|
|
f"not visible on page {page_num + 1}"
|
|
)
|
|
elif pagination_sel and page_num >= max_pages:
|
|
raise DynamicFlowError(
|
|
f"Item '{item_id}' not found after {page_num} pagination page(s)"
|
|
)
|
|
else:
|
|
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
|
|
log.debug(
|
|
"[%s] Scroll %d (page %d) — new height %d px",
|
|
name,
|
|
scroll_on_page,
|
|
page_num + 1,
|
|
new_height,
|
|
)
|
|
|
|
raise DynamicFlowError(
|
|
f"Item '{item_id}' not found after {total_scrolls} scrolls"
|
|
)
|
|
|
|
except DynamicFlowError 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)
|
|
except Exception as exc:
|
|
ms = int((time.monotonic() - t0) * 1000)
|
|
log.error("[%s] Unexpected error — %s", name, exc)
|
|
return StepResult(name=name, success=False, error=str(exc), duration_ms=ms)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Human mouse simulation
|
|
# ------------------------------------------------------------------
|
|
def _human_click(self, el: WebElement, name: str = "") -> None:
|
|
"""
|
|
Simulate a human-like mouse approach and click:
|
|
1. Smooth scroll element into viewport
|
|
2. Dispatch JS mousemove events along a curved path
|
|
3. Trigger mouseenter / mouseover on the element
|
|
4. Click via ActionChains with random sub-element offset
|
|
5. Fallback: synthetic JS mouse event sequence
|
|
"""
|
|
# Step 1 — bring element into view smoothly
|
|
self.driver.execute_script(
|
|
"arguments[0].scrollIntoView({block:'center', behavior:'smooth'});", el
|
|
)
|
|
time.sleep(random.uniform(0.35, 0.7))
|
|
|
|
# Step 2 — get element centre in viewport coordinates
|
|
rect: dict[str, float] = self.driver.execute_script(
|
|
"""
|
|
const r = arguments[0].getBoundingClientRect();
|
|
return {x: r.left, y: r.top, w: r.width, h: r.height,
|
|
vw: window.innerWidth, vh: window.innerHeight};
|
|
""",
|
|
el,
|
|
)
|
|
cx = rect["x"] + rect["w"] / 2
|
|
cy = rect["y"] + rect["h"] / 2
|
|
vw = rect["vw"]
|
|
vh = rect["vh"]
|
|
|
|
# Step 3 — build a curved path from a random off-element start
|
|
sx = random.uniform(vw * 0.1, vw * 0.5)
|
|
sy = random.uniform(vh * 0.1, vh * 0.5)
|
|
# Control point for quadratic Bézier — pulled perpendicular to the straight line
|
|
dx, dy = cx - sx, cy - sy
|
|
perp_x = -dy * random.uniform(0.15, 0.4)
|
|
perp_y = dx * random.uniform(0.15, 0.4)
|
|
qx = sx + dx / 2 + perp_x
|
|
qy = sy + dy / 2 + perp_y
|
|
|
|
steps = random.randint(10, 18)
|
|
prev_mx, prev_my = sx, sy
|
|
for i in range(1, steps + 1):
|
|
t = i / steps
|
|
# Quadratic Bézier point
|
|
bx = (1 - t) ** 2 * sx + 2 * (1 - t) * t * qx + t**2 * cx
|
|
by = (1 - t) ** 2 * sy + 2 * (1 - t) * t * qy + t**2 * cy
|
|
# Add small Gaussian jitter that fades near the target
|
|
jitter = max(0.0, 1.0 - t) * 4.0
|
|
mx = bx + random.gauss(0, jitter)
|
|
my = by + random.gauss(0, jitter)
|
|
|
|
self.driver.execute_script(
|
|
"""
|
|
document.dispatchEvent(new MouseEvent('mousemove', {
|
|
bubbles: true, cancelable: true,
|
|
clientX: arguments[0], clientY: arguments[1]
|
|
}));
|
|
""",
|
|
int(mx),
|
|
int(my),
|
|
)
|
|
# Variable inter-step pause — faster in the middle, slower near target
|
|
speed = 0.5 + 0.5 * math.sin(math.pi * t)
|
|
time.sleep(random.uniform(0.008, 0.025) / max(speed, 0.3))
|
|
prev_mx, prev_my = mx, my
|
|
|
|
# Step 4 — hover events on the element itself
|
|
self.driver.execute_script(
|
|
"""
|
|
arguments[0].dispatchEvent(new MouseEvent('mouseenter', {bubbles: true}));
|
|
arguments[0].dispatchEvent(new MouseEvent('mouseover', {bubbles: true}));
|
|
""",
|
|
el,
|
|
)
|
|
time.sleep(random.uniform(0.1, 0.28))
|
|
|
|
# Step 5 — ActionChains click with random offset from centre
|
|
off_x = int(random.uniform(-rect["w"] * 0.28, rect["w"] * 0.28))
|
|
off_y = int(random.uniform(-rect["h"] * 0.28, rect["h"] * 0.28))
|
|
log.debug(
|
|
"[%s] Human click — target=(%.0f,%.0f) offset=(%d,%d)",
|
|
name,
|
|
cx,
|
|
cy,
|
|
off_x,
|
|
off_y,
|
|
)
|
|
try:
|
|
ActionChains(self.driver).move_to_element_with_offset(
|
|
el, off_x, off_y
|
|
).pause(random.uniform(0.05, 0.14)).click().perform()
|
|
log.debug("[%s] ActionChains click OK", name)
|
|
except Exception as ac_exc:
|
|
log.warning(
|
|
"[%s] ActionChains failed (%s) — synthetic JS click", name, ac_exc
|
|
)
|
|
self.driver.execute_script(
|
|
"""
|
|
const opts = {bubbles: true, cancelable: true,
|
|
clientX: arguments[1], clientY: arguments[2]};
|
|
arguments[0].dispatchEvent(new MouseEvent('mousedown', opts));
|
|
arguments[0].dispatchEvent(new MouseEvent('mouseup', opts));
|
|
arguments[0].dispatchEvent(new MouseEvent('click', opts));
|
|
""",
|
|
el,
|
|
int(cx + off_x),
|
|
int(cy + off_y),
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Step: click the found element
|
|
# ------------------------------------------------------------------
|
|
def _run_click(self, item_id: str, element: Any) -> StepResult:
|
|
t0 = time.monotonic()
|
|
name = f"click:{item_id}"
|
|
log.info("[%s] Clicking item '%s'", name, item_id)
|
|
try:
|
|
if element is None:
|
|
raise DynamicFlowError("No element reference from scroll step")
|
|
|
|
self._human_click(element, name)
|
|
|
|
human_delay(0.8, 1.5)
|
|
ms = int((time.monotonic() - t0) * 1000)
|
|
log.info(
|
|
"[%s] Click OK — landed on %s — %d ms",
|
|
name,
|
|
self.driver.current_url,
|
|
ms,
|
|
)
|
|
return StepResult(
|
|
name=name,
|
|
success=True,
|
|
data={"item_id": item_id, "url_after": self.driver.current_url},
|
|
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)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Pagination helper
|
|
# ------------------------------------------------------------------
|
|
def _try_paginate(
|
|
self, step_name: str, sel: str, sel_type: str, wait_ms: int
|
|
) -> bool:
|
|
"""
|
|
Find the next-page button and click it reliably.
|
|
|
|
Uses a layered approach rather than full human-mouse simulation because
|
|
pagination buttons are small and offset-based clicking is fragile:
|
|
1. Native Selenium click (triggers real browser events)
|
|
2. JS click() on the element
|
|
3. Synthetic pointer + mouse + click event dispatch
|
|
"""
|
|
log.debug("[%s] Looking for pagination button [%s:%s]", step_name, sel_type, sel)
|
|
|
|
# Search all matching elements, not just the first visible one,
|
|
# in case the button is present but momentarily off-screen.
|
|
try:
|
|
candidates = _find_all(self.driver, sel, sel_type)
|
|
except Exception as exc:
|
|
log.warning("[%s] Pagination selector failed — %s", step_name, exc)
|
|
return False
|
|
|
|
btn: WebElement | None = None
|
|
for el in candidates:
|
|
try:
|
|
if el.is_displayed() and el.is_enabled():
|
|
btn = el
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
if btn is None:
|
|
log.debug("[%s] Pagination button not found / not interactable", step_name)
|
|
return False
|
|
|
|
log.debug("[%s] Pagination button found — attempting click", step_name)
|
|
try:
|
|
self.driver.execute_script(
|
|
"arguments[0].scrollIntoView({block:'nearest', behavior:'instant'});", btn
|
|
)
|
|
time.sleep(random.uniform(0.15, 0.35))
|
|
|
|
# Layer 1: native Selenium click
|
|
try:
|
|
btn.click()
|
|
log.debug("[%s] Pagination: native click OK", step_name)
|
|
except Exception as e1:
|
|
log.debug("[%s] Pagination: native click failed (%s) — trying JS", step_name, e1)
|
|
# Layer 2: JS .click()
|
|
try:
|
|
self.driver.execute_script("arguments[0].click();", btn)
|
|
log.debug("[%s] Pagination: JS click OK", step_name)
|
|
except Exception as e2:
|
|
log.debug("[%s] Pagination: JS click failed (%s) — dispatching events", step_name, e2)
|
|
# Layer 3: synthetic pointer + mouse + click events
|
|
self.driver.execute_script(
|
|
"""
|
|
const el = arguments[0];
|
|
const r = el.getBoundingClientRect();
|
|
const cx = r.left + r.width / 2, cy = r.top + r.height / 2;
|
|
const opts = {bubbles: true, cancelable: true, clientX: cx, clientY: cy};
|
|
['pointerdown','mousedown','pointerup','mouseup','click']
|
|
.forEach(t => el.dispatchEvent(new (t.startsWith('pointer') ? PointerEvent : MouseEvent)(t, opts)));
|
|
""",
|
|
btn,
|
|
)
|
|
log.debug("[%s] Pagination: synthetic events dispatched", step_name)
|
|
|
|
log.info("[%s] Pagination button clicked — waiting %d ms", step_name, wait_ms)
|
|
time.sleep(wait_ms / 1000)
|
|
return True
|
|
|
|
except Exception as exc:
|
|
log.warning("[%s] Pagination click failed — %s", step_name, exc)
|
|
return False
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helpers
|
|
# ------------------------------------------------------------------
|
|
def _build_item_selector(self, item_id: str) -> tuple[str, str]:
|
|
"""
|
|
Return (selector, effective_type).
|
|
|
|
Priority:
|
|
1. item_url_template — renders the URL pattern and uses XPath
|
|
contains(@href, ...) to find the matching <a> element.
|
|
2. item_selector_type == "a-link" — same href-contains approach but
|
|
driven by item_selector_template instead.
|
|
3. CSS / XPath — item_selector_template used verbatim.
|
|
|
|
Href-based searches always use XPath so that arbitrary URL characters
|
|
(colons, brackets, dots, slashes) never break the selector.
|
|
"""
|
|
url_tpl: str = self.cfg.get("item_url_template", "").strip()
|
|
if url_tpl:
|
|
href = url_tpl.replace("{item_id}", item_id)
|
|
log.debug("item_url_template resolved to href pattern '%s'", href)
|
|
return f"//a[contains(@href, '{href}')]", "xpath"
|
|
|
|
sel_type: str = self.cfg.get("item_selector_type", "css")
|
|
template: str = self.cfg.get("item_selector_template", "")
|
|
|
|
if sel_type == "a-link":
|
|
href_fragment = (
|
|
template.replace("{item_id}", item_id) if template else item_id
|
|
)
|
|
return f"//a[contains(@href, '{href_fragment}')]", "xpath"
|
|
|
|
if not template:
|
|
raise DynamicFlowError("item_selector_template is not configured")
|
|
return template.replace("{item_id}", item_id), sel_type
|
|
|
|
def _find_item_visible(
|
|
self, selector: str, selector_type: str
|
|
) -> WebElement | None:
|
|
try:
|
|
els = _find_all(self.driver, selector, selector_type)
|
|
except Exception as exc:
|
|
log.debug("Selector error [%s:%s]: %s", selector_type, selector, exc)
|
|
return None
|
|
for el in els:
|
|
try:
|
|
if el.is_displayed():
|
|
return el
|
|
except Exception:
|
|
continue
|
|
return None
|
|
|
|
def _get_scroll_height(
|
|
self, container_sel: str, container_type: str = "css"
|
|
) -> int:
|
|
body_h = int(self.driver.execute_script(
|
|
"return Math.max(document.body.scrollHeight, document.documentElement.scrollHeight)"
|
|
))
|
|
if container_sel:
|
|
try:
|
|
el = _find(self.driver, container_sel, container_type)
|
|
container_h = int(self.driver.execute_script("return arguments[0].scrollHeight", el))
|
|
return max(body_h, container_h)
|
|
except Exception as exc:
|
|
log.warning(
|
|
"Could not read scrollHeight from container [%s:'%s']: %s — using body",
|
|
container_type, container_sel, exc,
|
|
)
|
|
return body_h
|
|
|
|
def _scroll_down(self, container_sel: str, container_type: str = "css") -> None:
|
|
if container_sel:
|
|
try:
|
|
el = _find(self.driver, container_sel, container_type)
|
|
self.driver.execute_script(
|
|
"""
|
|
const el = arguments[0];
|
|
el.scrollTop = el.scrollHeight;
|
|
el.dispatchEvent(new Event('scroll', {bubbles: true}));
|
|
""",
|
|
el,
|
|
)
|
|
except Exception as exc:
|
|
log.warning(
|
|
"Could not scroll container [%s:'%s']: %s — falling through to window",
|
|
container_type, container_sel, exc,
|
|
)
|
|
# Always advance the window too — most infinite-scroll triggers
|
|
# listen to window scroll events, not container scroll events.
|
|
self.driver.execute_script(
|
|
"""
|
|
window.scrollTo(0, document.body.scrollHeight);
|
|
window.dispatchEvent(new Event('scroll', {bubbles: true}));
|
|
document.dispatchEvent(new Event('scroll', {bubbles: true}));
|
|
"""
|
|
)
|