cloakhq--cloakbrowser/cloakbrowser/human/scroll.py

"""cloakbrowser-human — Human-like scrolling via mouse wheel events."""

from __future__ import annotations

import math
import random
import time
from typing import Any, Callable, Optional, Tuple

from .config import HumanConfig, rand, rand_range, rand_int_range, sleep_ms
from .mouse import RawMouse, human_move
from .stealth_dom import build_box_js, eval_parsed, OK, NOT_FOUND, UNSUPPORTED, _VIEWPORT_JS


def _is_in_viewport(bounds: dict, viewport_height: int, cfg: HumanConfig) -> bool:
    top_edge = bounds["y"]
    bottom_edge = bounds["y"] + bounds["height"]
    zone_top = viewport_height * cfg.scroll_target_zone[0]
    zone_bottom = viewport_height * cfg.scroll_target_zone[1]
    return top_edge >= zone_top and bottom_edge <= zone_bottom


def _get_element_box(page: Any, selector: str, timeout: float = 30000) -> Optional[dict]:
    """Locate ``selector`` and return its bounding box.

    The ``timeout`` is forwarded to Playwright's ``boundingBox(timeout=...)``
    so callers can extend it for slow-loading elements (#172).

    Reads geometry through the isolated world when available; a not-found is
    retried briefly in-world (SPA re-renders) and only an *unsupported* selector
    reaches Playwright's ``bounding_box``.
    """
    world = getattr(page, "_stealth_world", None)
    if world is not None:
        status, data = eval_parsed(world, build_box_js(selector))
        if status == OK:
            return data["box"]
        if status == NOT_FOUND:
            deadline = time.monotonic() + min(timeout, 2000) / 1000.0
            while time.monotonic() < deadline:
                time.sleep(0.05)
                status, data = eval_parsed(world, build_box_js(selector))
                if status == OK:
                    return data["box"]
                if status == UNSUPPORTED:
                    break
            if status != UNSUPPORTED:
                return None
        # UNSUPPORTED -> Playwright fallback below
    try:
        el = page.locator(selector).first
        return el.bounding_box(timeout=max(1, timeout))
    except Exception:
        return None


_SCROLL_JS = (
    "(() => { const e = document.scrollingElement || document.documentElement;"
    " return { y: window.scrollY, maxY: Math.max(0, e.scrollHeight - e.clientHeight) }; })()"
)


def _read_scroll_state(page: Any) -> dict:
    """Current vertical scroll offset and the maximum scrollable offset."""
    world = getattr(page, "_stealth_world", None)
    if world is not None:
        try:
            return world.evaluate(_SCROLL_JS)
        except Exception:
            pass
    try:
        return page.evaluate(_SCROLL_JS)
    except Exception:
        return {"y": 0, "maxY": 0}


def _smooth_wheel(raw: RawMouse, delta: int, cfg: HumanConfig) -> None:
    """Send one logical scroll as a burst of small wheel events (like real inertia)."""
    abs_d = abs(delta)
    sign = 1 if delta > 0 else -1
    sent = 0
    while sent < abs_d:
        step_size = rand(20, 40)
        chunk = min(step_size, abs_d - sent)
        raw.wheel(0, round(chunk) * sign)
        sent += chunk
        sleep_ms(rand(8, 20))


def human_scroll_into_view(
    page: Any,
    raw: RawMouse,
    get_box: Callable[[], Optional[dict]],
    cursor_x: float, cursor_y: float,
    cfg: HumanConfig,
) -> Tuple[dict, float, float, bool]:
    """Humanized scrolling that uses an arbitrary ``get_box`` callable
    instead of a CSS selector.

    Used both by ``scroll_to_element`` (selector-based) and by
    ``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:
        # Headed launches default to no_viewport so the page tracks the real OS
        # window; page.viewport_size is then None. Read the live window dimensions
        # through the isolated world, consistent with the other geometry reads here.
        world = getattr(page, "_stealth_world", None)
        if world is not None:
            try:
                viewport = world.evaluate(_VIEWPORT_JS)
            except Exception:
                viewport = None
        if not viewport:
            viewport = page.evaluate(
                "() => ({ width: window.innerWidth, height: window.innerHeight })"
            )
    if not viewport or not viewport.get("height"):
        raise RuntimeError("Viewport size not available")

    viewport_height = viewport["height"]
    viewport_width = viewport["width"]

    box = get_box()
    if box is None:
        raise RuntimeError("Element not found while scrolling into view")

    if _is_in_viewport(box, viewport_height, cfg):
        return box, cursor_x, cursor_y, False

    # Already fully visible but off-center, with the page pinned at the boundary
    # in the needed direction: scrolling can't help, so don't waste the budget.
    fully_visible = box["y"] >= 0 and box["y"] + box["height"] <= viewport_height
    if fully_visible:
        zone_mid = viewport_height * (cfg.scroll_target_zone[0] + cfg.scroll_target_zone[1]) / 2
        need_up = box["y"] + box["height"] / 2 < zone_mid
        scroll = _read_scroll_state(page)
        if (scroll["y"] <= 0) if need_up else (scroll["y"] >= scroll["maxY"]):
            return box, cursor_x, cursor_y, False

    # Move cursor into scroll area
    scroll_area_x = round(viewport_width * rand(0.3, 0.7))
    scroll_area_y = round(viewport_height * rand(0.3, 0.7))
    human_move(raw, cursor_x, cursor_y, scroll_area_x, scroll_area_y, cfg)
    cursor_x = scroll_area_x
    cursor_y = scroll_area_y
    sleep_ms(rand_range(cfg.scroll_pre_move_delay))

    # Calculate scroll distance
    target_y = viewport_height * rand(cfg.scroll_target_zone[0], cfg.scroll_target_zone[1])
    element_center = box["y"] + box["height"] / 2
    distance_to_scroll = element_center - target_y

    direction = 1 if distance_to_scroll > 0 else -1
    abs_distance = abs(distance_to_scroll)
    avg_delta = (cfg.scroll_delta_base[0] + cfg.scroll_delta_base[1]) / 2
    total_clicks = max(3, math.ceil(abs_distance / avg_delta))
    accel_steps = rand_int_range(cfg.scroll_accel_steps)
    decel_steps = rand_int_range(cfg.scroll_decel_steps)

    # Scroll loop: accelerate → cruise → decelerate
    scrolled = 0
    for i in range(total_clicks):
        if i < accel_steps:
            delta = rand(80, 100)
            pause = rand_range(cfg.scroll_pause_slow)
        elif i >= total_clicks - decel_steps:
            delta = rand(60, 90)
            pause = rand_range(cfg.scroll_pause_slow)
        else:
            delta = rand_range(cfg.scroll_delta_base)
            pause = rand_range(cfg.scroll_pause_fast)

        delta *= 1 + (random.random() - 0.5) * 2 * cfg.scroll_delta_variance
        delta = round(delta) * direction

        _smooth_wheel(raw, delta, cfg)
        scrolled += abs(delta)
        sleep_ms(pause)

        # Check visibility every 3 steps
        if i % 3 == 2 or i == total_clicks - 1:
            box = get_box()
            if box and _is_in_viewport(box, viewport_height, cfg):
                break
        if scrolled >= abs_distance * 1.1:
            break

    # Optional overshoot + correction
    if random.random() < cfg.scroll_overshoot_chance:
        overshoot_px = round(rand_range(cfg.scroll_overshoot_px)) * direction
        _smooth_wheel(raw, overshoot_px, cfg)
        sleep_ms(rand_range(cfg.scroll_settle_delay))
        corrections = rand_int_range((1, 2))
        for _ in range(corrections):
            corr_delta = round(rand(40, 80)) * -direction
            _smooth_wheel(raw, corr_delta, cfg)
            sleep_ms(rand(100, 250))

    # Settle
    sleep_ms(rand_range(cfg.scroll_settle_delay))

    box = get_box()
    if box is None:
        raise RuntimeError("Element lost after scrolling into view")

    return box, cursor_x, cursor_y, True


def scroll_to_element(
    page: Any,
    raw: RawMouse,
    selector: str,
    cursor_x: float, cursor_y: float,
    cfg: HumanConfig,
    timeout: float = 30000,
) -> 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,
        lambda: _get_element_box(page, selector, timeout),
        cursor_x, cursor_y, cfg,
    )