mirror of
https://github.com/CloakHQ/CloakBrowser.git
synced 2026-06-23 11:41:46 +02:00
feat(humanize): add Playwright-style actionability checks (#228)
* feat(humanize): add Playwright-style actionability checks to all interaction methods
Humanized locator/page methods now perform pre-action validation matching
Playwright's native behavior: attached, visible, enabled, editable, stable,
and receives-pointer-events checks with retry loop and backoff.
- New error hierarchy: ActionabilityError base with ElementNotAttachedError,
ElementNotVisibleError, ElementNotStableError, ElementNotEnabledError,
ElementNotEditableError, ElementNotReceivingEventsError
- force=True parameter skips all actionability checks (matches Playwright)
- Shared deadline across all steps (checks + scroll + stable + pointer)
- Post-scroll stability check only runs when scroll actually happened
- Chained methods (type/fill/check/uncheck/press) skip inner click checks
but still run pointer-events check at actual click coordinates
- Frame methods now forward kwargs (force, timeout, human_config)
- Locator patches forward force via _forward_kwargs
- Python sync + async, JS/TS implementation
* fix(humanize): forward human_config in all chained methods, use evaluate args in handle pointer checks
- Add human_config=kwargs.get("human_config") to check/uncheck/select_option/press inner calls (sync+async+JS)
- Convert check_pointer_events_handle from f-string interpolation to evaluate args pattern (sync+async+JS)
* fix(humanize): strip custom kwargs before forwarding to Playwright select_option
originals.select_option(**kwargs) passes human_config/force to Playwright
which rejects unknown kwargs with TypeError.
This commit is contained in:
Cloak-HQ
+398
-73
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,342 @@
|
||||
"""Playwright-style actionability checks for the humanize layer (sync).
|
||||
|
||||
Checks: attached, visible, stable, enabled, editable, receives pointer events.
|
||||
Retry loop with backoff matching Playwright internals: [100, 250, 500, 1000]ms.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from typing import Any, FrozenSet, Optional, Tuple
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Error hierarchy — all subclass RuntimeError for backward compat
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class ActionabilityError(RuntimeError):
|
||||
"""Base for all actionability failures."""
|
||||
|
||||
def __init__(self, selector: str, check: str, message: str):
|
||||
self.selector = selector
|
||||
self.check = check
|
||||
super().__init__(f"Element {selector!r} failed {check} check: {message}")
|
||||
|
||||
|
||||
class ElementNotAttachedError(ActionabilityError):
|
||||
def __init__(self, selector: str):
|
||||
super().__init__(selector, "attached", "element not found in DOM")
|
||||
|
||||
|
||||
class ElementNotVisibleError(ActionabilityError):
|
||||
def __init__(self, selector: str):
|
||||
super().__init__(selector, "visible", "element is not visible")
|
||||
|
||||
|
||||
class ElementNotStableError(ActionabilityError):
|
||||
def __init__(self, selector: str):
|
||||
super().__init__(selector, "stable", "element position is still changing")
|
||||
|
||||
|
||||
class ElementNotEnabledError(ActionabilityError):
|
||||
def __init__(self, selector: str):
|
||||
super().__init__(selector, "enabled", "element is disabled")
|
||||
|
||||
|
||||
class ElementNotEditableError(ActionabilityError):
|
||||
def __init__(self, selector: str):
|
||||
super().__init__(selector, "editable", "element is not editable")
|
||||
|
||||
|
||||
class ElementNotReceivingEventsError(ActionabilityError):
|
||||
def __init__(self, selector: str, covering_tag: str = "unknown"):
|
||||
super().__init__(
|
||||
selector,
|
||||
"pointer_events",
|
||||
f"element is covered by <{covering_tag}>",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Check-set constants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
CHECKS_CLICK: FrozenSet[str] = frozenset({"attached", "visible", "enabled", "pointer_events"})
|
||||
CHECKS_HOVER: FrozenSet[str] = frozenset({"attached", "visible", "pointer_events"})
|
||||
CHECKS_INPUT: FrozenSet[str] = frozenset({"attached", "visible", "enabled", "editable", "pointer_events"})
|
||||
CHECKS_FOCUS: FrozenSet[str] = frozenset({"attached", "visible", "enabled"})
|
||||
CHECKS_CHECK: FrozenSet[str] = frozenset({"attached", "visible", "enabled", "pointer_events"})
|
||||
|
||||
_BACKOFF_MS = [100, 250, 500, 1000]
|
||||
|
||||
|
||||
def _backoff_sleep(attempt: int) -> None:
|
||||
idx = min(attempt, len(_BACKOFF_MS) - 1)
|
||||
time.sleep(_BACKOFF_MS[idx] / 1000.0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pre-scroll actionability: attached, visible, enabled, editable
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def ensure_actionable(
|
||||
page: Any,
|
||||
selector: str,
|
||||
checks: FrozenSet[str],
|
||||
timeout: float = 30000,
|
||||
force: bool = False,
|
||||
) -> None:
|
||||
"""Wait for element to pass actionability checks (pre-scroll).
|
||||
|
||||
Retries with backoff until *timeout* ms elapsed.
|
||||
Raises a specific ``ActionabilityError`` subclass on failure.
|
||||
If *force* is True, returns immediately.
|
||||
"""
|
||||
if force:
|
||||
return
|
||||
|
||||
deadline = time.monotonic() + timeout / 1000.0
|
||||
attempt = 0
|
||||
last_error: Optional[ActionabilityError] = None
|
||||
|
||||
while True:
|
||||
remaining_ms = max(0, (deadline - time.monotonic()) * 1000)
|
||||
if remaining_ms <= 0:
|
||||
if last_error is not None:
|
||||
raise last_error
|
||||
raise ActionabilityError(selector, "timeout", "timeout expired before first check")
|
||||
|
||||
try:
|
||||
loc = page.locator(selector).first
|
||||
|
||||
if "attached" in checks:
|
||||
try:
|
||||
loc.wait_for(state="attached", timeout=max(1, min(remaining_ms, 2000)))
|
||||
except Exception:
|
||||
raise ElementNotAttachedError(selector)
|
||||
|
||||
if "visible" in checks:
|
||||
if not loc.is_visible():
|
||||
raise ElementNotVisibleError(selector)
|
||||
|
||||
if "enabled" in checks:
|
||||
if not loc.is_enabled():
|
||||
raise ElementNotEnabledError(selector)
|
||||
|
||||
if "editable" in checks:
|
||||
if not loc.is_editable():
|
||||
raise ElementNotEditableError(selector)
|
||||
|
||||
return
|
||||
|
||||
except ActionabilityError as e:
|
||||
last_error = e
|
||||
if time.monotonic() >= deadline:
|
||||
raise last_error
|
||||
_backoff_sleep(attempt)
|
||||
attempt += 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Post-scroll stability check
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _boxes_differ(a: dict, b: dict) -> bool:
|
||||
return (
|
||||
abs(a["x"] - b["x"]) > 1
|
||||
or abs(a["y"] - b["y"]) > 1
|
||||
or abs(a["width"] - b["width"]) > 1
|
||||
or abs(a["height"] - b["height"]) > 1
|
||||
)
|
||||
|
||||
|
||||
def ensure_stable(
|
||||
page: Any,
|
||||
selector: str,
|
||||
timeout: float = 5000,
|
||||
) -> None:
|
||||
"""Wait for element position to stabilize (two samples 100ms apart).
|
||||
|
||||
Only call after scroll — skip if element was already in viewport.
|
||||
"""
|
||||
deadline = time.monotonic() + timeout / 1000.0
|
||||
attempt = 0
|
||||
|
||||
while True:
|
||||
remaining_ms = max(0, (deadline - time.monotonic()) * 1000)
|
||||
if remaining_ms <= 0:
|
||||
raise ElementNotStableError(selector)
|
||||
|
||||
loc = page.locator(selector).first
|
||||
box1 = loc.bounding_box(timeout=max(1, min(remaining_ms, 1000)))
|
||||
if box1 is None:
|
||||
raise ElementNotAttachedError(selector)
|
||||
|
||||
time.sleep(0.1)
|
||||
|
||||
box2 = loc.bounding_box(timeout=max(1, min(remaining_ms, 1000)))
|
||||
if box2 is None:
|
||||
raise ElementNotAttachedError(selector)
|
||||
|
||||
if not _boxes_differ(box1, box2):
|
||||
return
|
||||
|
||||
if time.monotonic() >= deadline:
|
||||
raise ElementNotStableError(selector)
|
||||
|
||||
_backoff_sleep(attempt)
|
||||
attempt += 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pointer-events check (post-scroll, at actual click coordinates)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_POINTER_EVENTS_LOCATOR_JS = """(expected, coords) => {
|
||||
const target = document.elementFromPoint(coords.x, coords.y);
|
||||
if (!target) return { hit: false, reason: 'no_element_at_point', covering: 'none' };
|
||||
let node = target;
|
||||
while (node) { if (node === expected) return { hit: true }; node = node.parentNode; }
|
||||
if (expected.contains(target)) return { hit: true };
|
||||
return { hit: false, reason: 'covered', covering: target.tagName || 'unknown' };
|
||||
}"""
|
||||
|
||||
_POINTER_EVENTS_HANDLE_JS = """(expected, coords) => {
|
||||
const target = document.elementFromPoint(coords.x, coords.y);
|
||||
if (!target) return { hit: false, reason: 'no_element_at_point', covering: 'none' };
|
||||
let node = target;
|
||||
while (node) { if (node === expected) return { hit: true }; node = node.parentNode; }
|
||||
if (expected.contains(target)) return { hit: true };
|
||||
return { hit: false, reason: 'covered', covering: target.tagName || 'unknown' };
|
||||
}"""
|
||||
|
||||
|
||||
def check_pointer_events(
|
||||
page: Any,
|
||||
selector: str,
|
||||
x: float,
|
||||
y: float,
|
||||
stealth: Any = None,
|
||||
timeout: float = 5000,
|
||||
) -> None:
|
||||
"""Check that elementFromPoint(x, y) hits the expected element.
|
||||
|
||||
Uses locator.evaluate() so all Playwright selector types work
|
||||
(text=, role=, XPath, CSS, etc.). Retries with backoff for transient overlays.
|
||||
"""
|
||||
deadline = time.monotonic() + timeout / 1000.0
|
||||
attempt = 0
|
||||
coords = {"x": x, "y": y}
|
||||
|
||||
while True:
|
||||
try:
|
||||
loc = page.locator(selector).first
|
||||
result = loc.evaluate(_POINTER_EVENTS_LOCATOR_JS, coords)
|
||||
except Exception as exc:
|
||||
logger.debug("pointer_events check failed for %r: %s", selector, exc)
|
||||
result = None
|
||||
|
||||
if result and result.get("hit", False):
|
||||
return
|
||||
|
||||
covering = (result or {}).get("covering", "unknown")
|
||||
|
||||
if time.monotonic() >= deadline:
|
||||
raise ElementNotReceivingEventsError(selector, covering)
|
||||
|
||||
_backoff_sleep(attempt)
|
||||
attempt += 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ElementHandle variant
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def ensure_actionable_handle(
|
||||
page: Any,
|
||||
el: Any,
|
||||
checks: FrozenSet[str],
|
||||
timeout: float = 30000,
|
||||
force: bool = False,
|
||||
) -> None:
|
||||
"""Actionability checks for ElementHandle (no selector needed).
|
||||
|
||||
Uses Playwright's wait_for_element_state where available.
|
||||
"""
|
||||
if force:
|
||||
return
|
||||
|
||||
deadline = time.monotonic() + timeout / 1000.0
|
||||
attempt = 0
|
||||
last_error: Optional[ActionabilityError] = None
|
||||
label = "<ElementHandle>"
|
||||
|
||||
while True:
|
||||
remaining_ms = max(0, (deadline - time.monotonic()) * 1000)
|
||||
if remaining_ms <= 0:
|
||||
if last_error is not None:
|
||||
raise last_error
|
||||
raise ActionabilityError(label, "timeout", "timeout expired before first check")
|
||||
|
||||
try:
|
||||
if "visible" in checks:
|
||||
try:
|
||||
el.wait_for_element_state("visible", timeout=max(1, min(remaining_ms, 2000)))
|
||||
except Exception:
|
||||
raise ElementNotVisibleError(label)
|
||||
|
||||
if "enabled" in checks:
|
||||
try:
|
||||
el.wait_for_element_state("enabled", timeout=max(1, min(remaining_ms, 2000)))
|
||||
except Exception:
|
||||
raise ElementNotEnabledError(label)
|
||||
|
||||
if "editable" in checks:
|
||||
try:
|
||||
el.wait_for_element_state("editable", timeout=max(1, min(remaining_ms, 2000)))
|
||||
except Exception:
|
||||
raise ElementNotEditableError(label)
|
||||
|
||||
return
|
||||
|
||||
except ActionabilityError as e:
|
||||
last_error = e
|
||||
if time.monotonic() >= deadline:
|
||||
raise last_error
|
||||
_backoff_sleep(attempt)
|
||||
attempt += 1
|
||||
|
||||
|
||||
def check_pointer_events_handle(
|
||||
page: Any,
|
||||
el: Any,
|
||||
x: float,
|
||||
y: float,
|
||||
timeout: float = 5000,
|
||||
) -> None:
|
||||
"""Pointer-events check for ElementHandle."""
|
||||
deadline = time.monotonic() + timeout / 1000.0
|
||||
attempt = 0
|
||||
|
||||
coords = {"x": x, "y": y}
|
||||
|
||||
while True:
|
||||
try:
|
||||
result = el.evaluate(_POINTER_EVENTS_HANDLE_JS, coords)
|
||||
except Exception:
|
||||
result = None
|
||||
|
||||
if result and result.get("hit", False):
|
||||
return
|
||||
|
||||
covering = (result or {}).get("covering", "unknown")
|
||||
|
||||
if time.monotonic() >= deadline:
|
||||
raise ElementNotReceivingEventsError("<ElementHandle>", covering)
|
||||
|
||||
_backoff_sleep(attempt)
|
||||
attempt += 1
|
||||
@@ -0,0 +1,247 @@
|
||||
"""Playwright-style actionability checks for the humanize layer (async).
|
||||
|
||||
Async mirror of actionability.py — same logic, uses asyncio.sleep and await.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
from typing import Any, FrozenSet, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
from .actionability import (
|
||||
ActionabilityError,
|
||||
ElementNotAttachedError,
|
||||
ElementNotVisibleError,
|
||||
ElementNotStableError,
|
||||
ElementNotEnabledError,
|
||||
ElementNotEditableError,
|
||||
ElementNotReceivingEventsError,
|
||||
_BACKOFF_MS,
|
||||
_boxes_differ,
|
||||
_POINTER_EVENTS_LOCATOR_JS,
|
||||
_POINTER_EVENTS_HANDLE_JS,
|
||||
)
|
||||
|
||||
|
||||
async def _async_backoff_sleep(attempt: int) -> None:
|
||||
idx = min(attempt, len(_BACKOFF_MS) - 1)
|
||||
await asyncio.sleep(_BACKOFF_MS[idx] / 1000.0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pre-scroll actionability
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def async_ensure_actionable(
|
||||
page: Any,
|
||||
selector: str,
|
||||
checks: FrozenSet[str],
|
||||
timeout: float = 30000,
|
||||
force: bool = False,
|
||||
) -> None:
|
||||
if force:
|
||||
return
|
||||
|
||||
deadline = time.monotonic() + timeout / 1000.0
|
||||
attempt = 0
|
||||
last_error: Optional[ActionabilityError] = None
|
||||
|
||||
while True:
|
||||
remaining_ms = max(0, (deadline - time.monotonic()) * 1000)
|
||||
if remaining_ms <= 0:
|
||||
if last_error is not None:
|
||||
raise last_error
|
||||
raise ActionabilityError(selector, "timeout", "timeout expired before first check")
|
||||
|
||||
try:
|
||||
loc = page.locator(selector).first
|
||||
|
||||
if "attached" in checks:
|
||||
try:
|
||||
await loc.wait_for(state="attached", timeout=max(1, min(remaining_ms, 2000)))
|
||||
except Exception:
|
||||
raise ElementNotAttachedError(selector)
|
||||
|
||||
if "visible" in checks:
|
||||
if not await loc.is_visible():
|
||||
raise ElementNotVisibleError(selector)
|
||||
|
||||
if "enabled" in checks:
|
||||
if not await loc.is_enabled():
|
||||
raise ElementNotEnabledError(selector)
|
||||
|
||||
if "editable" in checks:
|
||||
if not await loc.is_editable():
|
||||
raise ElementNotEditableError(selector)
|
||||
|
||||
return
|
||||
|
||||
except ActionabilityError as e:
|
||||
last_error = e
|
||||
if time.monotonic() >= deadline:
|
||||
raise last_error
|
||||
await _async_backoff_sleep(attempt)
|
||||
attempt += 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Post-scroll stability check
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def async_ensure_stable(
|
||||
page: Any,
|
||||
selector: str,
|
||||
timeout: float = 5000,
|
||||
) -> None:
|
||||
deadline = time.monotonic() + timeout / 1000.0
|
||||
attempt = 0
|
||||
|
||||
while True:
|
||||
remaining_ms = max(0, (deadline - time.monotonic()) * 1000)
|
||||
if remaining_ms <= 0:
|
||||
raise ElementNotStableError(selector)
|
||||
|
||||
loc = page.locator(selector).first
|
||||
box1 = await loc.bounding_box(timeout=max(1, min(remaining_ms, 1000)))
|
||||
if box1 is None:
|
||||
raise ElementNotAttachedError(selector)
|
||||
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
box2 = await loc.bounding_box(timeout=max(1, min(remaining_ms, 1000)))
|
||||
if box2 is None:
|
||||
raise ElementNotAttachedError(selector)
|
||||
|
||||
if not _boxes_differ(box1, box2):
|
||||
return
|
||||
|
||||
if time.monotonic() >= deadline:
|
||||
raise ElementNotStableError(selector)
|
||||
|
||||
await _async_backoff_sleep(attempt)
|
||||
attempt += 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pointer-events check
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def async_check_pointer_events(
|
||||
page: Any,
|
||||
selector: str,
|
||||
x: float,
|
||||
y: float,
|
||||
stealth: Any = None,
|
||||
timeout: float = 5000,
|
||||
) -> None:
|
||||
deadline = time.monotonic() + timeout / 1000.0
|
||||
attempt = 0
|
||||
coords = {"x": x, "y": y}
|
||||
|
||||
while True:
|
||||
try:
|
||||
loc = page.locator(selector).first
|
||||
result = await loc.evaluate(_POINTER_EVENTS_LOCATOR_JS, coords)
|
||||
except Exception as exc:
|
||||
logger.debug("pointer_events check failed for %r: %s", selector, exc)
|
||||
result = None
|
||||
|
||||
if result and result.get("hit", False):
|
||||
return
|
||||
|
||||
covering = (result or {}).get("covering", "unknown")
|
||||
|
||||
if time.monotonic() >= deadline:
|
||||
raise ElementNotReceivingEventsError(selector, covering)
|
||||
|
||||
await _async_backoff_sleep(attempt)
|
||||
attempt += 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ElementHandle variant
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def async_ensure_actionable_handle(
|
||||
page: Any,
|
||||
el: Any,
|
||||
checks: FrozenSet[str],
|
||||
timeout: float = 30000,
|
||||
force: bool = False,
|
||||
) -> None:
|
||||
if force:
|
||||
return
|
||||
|
||||
deadline = time.monotonic() + timeout / 1000.0
|
||||
attempt = 0
|
||||
last_error: Optional[ActionabilityError] = None
|
||||
label = "<ElementHandle>"
|
||||
|
||||
while True:
|
||||
remaining_ms = max(0, (deadline - time.monotonic()) * 1000)
|
||||
if remaining_ms <= 0:
|
||||
if last_error is not None:
|
||||
raise last_error
|
||||
raise ActionabilityError(label, "timeout", "timeout expired before first check")
|
||||
|
||||
try:
|
||||
if "visible" in checks:
|
||||
try:
|
||||
await el.wait_for_element_state("visible", timeout=max(1, min(remaining_ms, 2000)))
|
||||
except Exception:
|
||||
raise ElementNotVisibleError(label)
|
||||
|
||||
if "enabled" in checks:
|
||||
try:
|
||||
await el.wait_for_element_state("enabled", timeout=max(1, min(remaining_ms, 2000)))
|
||||
except Exception:
|
||||
raise ElementNotEnabledError(label)
|
||||
|
||||
if "editable" in checks:
|
||||
try:
|
||||
await el.wait_for_element_state("editable", timeout=max(1, min(remaining_ms, 2000)))
|
||||
except Exception:
|
||||
raise ElementNotEditableError(label)
|
||||
|
||||
return
|
||||
|
||||
except ActionabilityError as e:
|
||||
last_error = e
|
||||
if time.monotonic() >= deadline:
|
||||
raise last_error
|
||||
await _async_backoff_sleep(attempt)
|
||||
attempt += 1
|
||||
|
||||
|
||||
async def async_check_pointer_events_handle(
|
||||
page: Any,
|
||||
el: Any,
|
||||
x: float,
|
||||
y: float,
|
||||
timeout: float = 5000,
|
||||
) -> None:
|
||||
deadline = time.monotonic() + timeout / 1000.0
|
||||
attempt = 0
|
||||
|
||||
coords = {"x": x, "y": y}
|
||||
|
||||
while True:
|
||||
try:
|
||||
result = await el.evaluate(_POINTER_EVENTS_HANDLE_JS, coords)
|
||||
except Exception:
|
||||
result = None
|
||||
|
||||
if result and result.get("hit", False):
|
||||
return
|
||||
|
||||
covering = (result or {}).get("covering", "unknown")
|
||||
|
||||
if time.monotonic() >= deadline:
|
||||
raise ElementNotReceivingEventsError("<ElementHandle>", covering)
|
||||
|
||||
await _async_backoff_sleep(attempt)
|
||||
attempt += 1
|
||||
@@ -26,7 +26,7 @@ def _get_element_box(page: Any, selector: str, timeout: float = 30000) -> Option
|
||||
"""
|
||||
try:
|
||||
el = page.locator(selector).first
|
||||
return el.bounding_box(timeout=timeout)
|
||||
return el.bounding_box(timeout=max(1, timeout))
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
@@ -50,7 +50,7 @@ def human_scroll_into_view(
|
||||
get_box: Callable[[], Optional[dict]],
|
||||
cursor_x: float, cursor_y: float,
|
||||
cfg: HumanConfig,
|
||||
) -> Tuple[dict, float, float]:
|
||||
) -> Tuple[dict, float, float, bool]:
|
||||
"""Humanized scrolling that uses an arbitrary ``get_box`` callable
|
||||
instead of a CSS selector.
|
||||
|
||||
@@ -58,6 +58,9 @@ def human_scroll_into_view(
|
||||
``ElementHandle.scroll_into_view_if_needed`` / ``Locator.scroll_into_view_if_needed``
|
||||
(handle-based) so the same accelerate \u2192 cruise \u2192 decelerate \u2192 overshoot
|
||||
behavior runs everywhere.
|
||||
|
||||
Returns ``(box, cursor_x, cursor_y, did_scroll)`` \u2014 *did_scroll* is False
|
||||
when the element was already in the viewport.
|
||||
"""
|
||||
viewport = page.viewport_size
|
||||
if not viewport:
|
||||
@@ -71,7 +74,7 @@ def human_scroll_into_view(
|
||||
raise RuntimeError("Element not found while scrolling into view")
|
||||
|
||||
if _is_in_viewport(box, viewport_height, cfg):
|
||||
return box, cursor_x, cursor_y
|
||||
return box, cursor_x, cursor_y, False
|
||||
|
||||
# Move cursor into scroll area
|
||||
scroll_area_x = round(viewport_width * rand(0.3, 0.7))
|
||||
@@ -139,7 +142,7 @@ def human_scroll_into_view(
|
||||
if box is None:
|
||||
raise RuntimeError("Element lost after scrolling into view")
|
||||
|
||||
return box, cursor_x, cursor_y
|
||||
return box, cursor_x, cursor_y, True
|
||||
|
||||
|
||||
def scroll_to_element(
|
||||
@@ -149,12 +152,14 @@ def scroll_to_element(
|
||||
cursor_x: float, cursor_y: float,
|
||||
cfg: HumanConfig,
|
||||
timeout: float = 30000,
|
||||
) -> Tuple[dict, float, float]:
|
||||
) -> Tuple[dict, float, float, bool]:
|
||||
"""Selector-based humanized scroll.
|
||||
|
||||
``timeout`` is forwarded to ``locator.bounding_box(timeout=...)`` so callers
|
||||
such as ``page.click('#x', timeout=5000)`` can wait longer for slow elements
|
||||
(#172). Default matches Playwright's 30000ms when not specified.
|
||||
|
||||
Returns ``(box, cursor_x, cursor_y, did_scroll)``.
|
||||
"""
|
||||
return human_scroll_into_view(
|
||||
page, raw,
|
||||
|
||||
@@ -23,7 +23,7 @@ async def _get_element_box_async(
|
||||
elements (#172)."""
|
||||
try:
|
||||
el = page.locator(selector).first
|
||||
return await el.bounding_box(timeout=timeout)
|
||||
return await el.bounding_box(timeout=max(1, timeout))
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
@@ -47,13 +47,16 @@ async def async_human_scroll_into_view(
|
||||
get_box: Callable[[], Awaitable[Optional[dict]]],
|
||||
cursor_x: float, cursor_y: float,
|
||||
cfg: HumanConfig,
|
||||
) -> Tuple[dict, float, float]:
|
||||
) -> Tuple[dict, float, float, bool]:
|
||||
"""Humanized scrolling using an arbitrary async ``get_box`` callable.
|
||||
|
||||
Used by both ``async_scroll_to_element`` (selector-based) and the
|
||||
ElementHandle / Locator ``scroll_into_view_if_needed`` patches so all
|
||||
scrolling paths share the same accelerate \u2192 cruise \u2192 decelerate
|
||||
\u2192 overshoot behavior.
|
||||
|
||||
Returns ``(box, cursor_x, cursor_y, did_scroll)`` \u2014 *did_scroll* is False
|
||||
when the element was already in the viewport.
|
||||
"""
|
||||
viewport = page.viewport_size
|
||||
if not viewport:
|
||||
@@ -67,7 +70,7 @@ async def async_human_scroll_into_view(
|
||||
raise RuntimeError("Element not found while scrolling into view")
|
||||
|
||||
if _is_in_viewport(box, viewport_height, cfg):
|
||||
return box, cursor_x, cursor_y
|
||||
return box, cursor_x, cursor_y, False
|
||||
|
||||
# Move cursor into scroll area
|
||||
scroll_area_x = round(viewport_width * rand(0.3, 0.7))
|
||||
@@ -135,7 +138,7 @@ async def async_human_scroll_into_view(
|
||||
if box is None:
|
||||
raise RuntimeError("Element lost after scrolling into view")
|
||||
|
||||
return box, cursor_x, cursor_y
|
||||
return box, cursor_x, cursor_y, True
|
||||
|
||||
|
||||
async def async_scroll_to_element(
|
||||
@@ -145,12 +148,14 @@ async def async_scroll_to_element(
|
||||
cursor_x: float, cursor_y: float,
|
||||
cfg: HumanConfig,
|
||||
timeout: float = 30000,
|
||||
) -> Tuple[dict, float, float]:
|
||||
) -> Tuple[dict, float, float, bool]:
|
||||
"""Selector-based humanized scroll (async).
|
||||
|
||||
``timeout`` is forwarded to ``locator.bounding_box(timeout=...)`` so callers
|
||||
such as ``page.click('#x', timeout=5000)`` can wait longer for slow elements
|
||||
(#172). Default matches Playwright's 30000ms when not specified.
|
||||
|
||||
Returns ``(box, cursor_x, cursor_y, did_scroll)``.
|
||||
"""
|
||||
async def _get():
|
||||
return await _get_element_box_async(page, selector, timeout)
|
||||
|
||||
Reference in New Issue
Block a user