from __future__ import annotations import asyncio from dataclasses import dataclass, field from typing import Any from browser_core import BrowserBridgeError, BrowserManager POINTER_BACKEND = "playwright-page-mouse" _ALLOWED_BUTTONS = {"left", "middle", "right"} _INTERACTIVE_SELECTOR = 'a,button,input,textarea,select,[role="button"],[role="link"]' @dataclass class PointerState: x: int = 0 y: int = 0 buttons: set[str] = field(default_factory=set) _POINTER_STATES: dict[tuple[str, str], PointerState] = {} def _button(value: str) -> str: selected = str(value or "left").strip().casefold() if selected not in _ALLOWED_BUTTONS: raise BrowserBridgeError("POINTER_BUTTON_INVALID", "button must be left, middle, or right.") return selected def _state(owner: str, session_id: str) -> PointerState: return _POINTER_STATES.setdefault((owner.casefold(), session_id), PointerState()) async def _point(manager: BrowserManager, session: Any, x: int, y: int) -> tuple[Any, int, int]: page = await manager._sync_page(session) viewport = page.viewport_size or {"width": 1280, "height": 900} px, py = int(x), int(y) width = int(viewport.get("width") or 0) height = int(viewport.get("height") or 0) if px < 0 or py < 0 or px >= width or py >= height: raise BrowserBridgeError( "POINTER_COORDINATE_OUT_OF_VIEWPORT", "Pointer coordinates must be inside the current isolated cloud-browser viewport.", details={"x": px, "y": py, "viewport_width": width, "viewport_height": height}, ) return page, px, py async def _guard_raw_pointer_target(page: Any, x: int, y: int) -> None: """Fail closed on normal interactive controls; raw pointer is for canvas/custom surfaces only.""" controls = page.locator(_INTERACTIVE_SELECTOR) try: count = min(await controls.count(), 200) except Exception as exc: raise BrowserBridgeError( "POINTER_TARGET_GUARD_UNAVAILABLE", "The browser could not verify the raw-pointer target safely.", details={"error_type": type(exc).__name__}, ) from exc for index in range(count): control = controls.nth(index) try: box = await control.bounding_box(timeout=400) except Exception: continue if not box: continue left = float(box.get("x") or 0) top = float(box.get("y") or 0) right = left + float(box.get("width") or 0) bottom = top + float(box.get("height") or 0) if not (left <= x < right and top <= y < bottom): continue attrs: list[str] = [] for name in ("type", "role", "name", "aria-label", "title", "placeholder", "autocomplete"): try: value = await control.get_attribute(name, timeout=300) except Exception: value = None if value: attrs.append(str(value)[:200]) try: text = (await control.inner_text(timeout=300))[:200] except Exception: text = "" raise BrowserBridgeError( "POINTER_INTERACTIVE_CONTROL_BLOCKED", "Raw pointer presses/drags are restricted to non-standard browser surfaces such as canvas. Use reviewed DOM tools for links, buttons, form fields, uploads, and submit controls.", details={"target_signature": " ".join([*attrs, text]).strip()[:500]}, ) def _result(session: Any, state: PointerState, *, status: str, button: str | None = None, duration_ms: int | None = None, steps: int | None = None) -> dict[str, Any]: return { "ok": True, "status": status, "session_id": session.session_id, "url": session.page.url, "x": state.x, "y": state.y, "button": button, "buttons_down": sorted(state.buttons), "duration_ms": duration_ms, "steps": steps, "pointer_backend": POINTER_BACKEND, "desktop_input_used": False, "submitted": False, } async def mouse_move(manager: BrowserManager, owner: str, session_id: str, x: int, y: int) -> dict[str, Any]: session = await manager.get(owner, session_id, touch=False) async with session.lock: await manager.revalidate_locked_session(owner, session_id, session) page, px, py = await _point(manager, session, x, y) await page.mouse.move(px, py) state = _state(owner, session_id) state.x, state.y = px, py return _result(session, state, status="BROWSER_POINTER_MOVED") async def mouse_down(manager: BrowserManager, owner: str, session_id: str, button: str = "left", *, user_confirmed_pointer_action: bool = False) -> dict[str, Any]: selected = _button(button) session = await manager.get(owner, session_id, touch=False) async with session.lock: await manager.revalidate_locked_session(owner, session_id, session) state = _state(owner, session_id) page, px, py = await _point(manager, session, state.x, state.y) if selected in state.buttons: raise BrowserBridgeError("POINTER_BUTTON_ALREADY_DOWN", "This pointer button is already held down.") await _guard_raw_pointer_target(page, px, py) await page.mouse.down(button=selected) state.buttons.add(selected) return _result(session, state, status="BROWSER_POINTER_DOWN", button=selected) async def mouse_up(manager: BrowserManager, owner: str, session_id: str, button: str = "left") -> dict[str, Any]: selected = _button(button) session = await manager.get(owner, session_id, touch=False) async with session.lock: await manager.revalidate_locked_session(owner, session_id, session) state = _state(owner, session_id) page, _px, _py = await _point(manager, session, state.x, state.y) if selected not in state.buttons: raise BrowserBridgeError("POINTER_BUTTON_NOT_DOWN", "This pointer button is not currently held down.") await page.mouse.up(button=selected) state.buttons.discard(selected) return _result(session, state, status="BROWSER_POINTER_UP", button=selected) async def mouse_click(manager: BrowserManager, owner: str, session_id: str, x: int, y: int, *, button: str = "left", click_count: int = 1) -> dict[str, Any]: """Atomically click a canvas/custom surface in the isolated cloud browser. Standard browser controls remain guarded so ordinary pages retain explicit semantic controls for submit, authentication and sensitive fields. This is deliberately a browser-page mouse operation, never Windows desktop input. """ selected = _button(button) count = max(1, min(2, int(click_count))) session = await manager.get(owner, session_id, touch=False) async with session.lock: await manager.revalidate_locked_session(owner, session_id, session) state = _state(owner, session_id) page, px, py = await _point(manager, session, x, y) if state.buttons: raise BrowserBridgeError( "POINTER_BUTTON_STATE_BUSY", "Release held pointer buttons before starting an atomic pointer click.", ) await _guard_raw_pointer_target(page, px, py) await page.mouse.click(px, py, button=selected, click_count=count) state.x, state.y = px, py return _result( session, state, status="BROWSER_POINTER_CLICKED", button=selected, steps=count, ) async def drag(manager: BrowserManager, owner: str, session_id: str, start_x: int, start_y: int, end_x: int, end_y: int, *, duration_ms: int = 400, button: str = "left", user_confirmed_pointer_action: bool = False) -> dict[str, Any]: selected = _button(button) duration = max(0, min(5000, int(duration_ms))) steps = max(1, min(120, round(duration / 16) if duration else 1)) session = await manager.get(owner, session_id, touch=False) async with session.lock: await manager.revalidate_locked_session(owner, session_id, session) state = _state(owner, session_id) page, sx, sy = await _point(manager, session, start_x, start_y) _page2, ex, ey = await _point(manager, session, end_x, end_y) if state.buttons: raise BrowserBridgeError( "POINTER_BUTTON_STATE_BUSY", "Release held pointer buttons before starting an atomic drag.", ) await _guard_raw_pointer_target(page, sx, sy) await _guard_raw_pointer_target(page, ex, ey) await page.mouse.move(sx, sy) state.x, state.y = sx, sy down = False try: await page.mouse.down(button=selected) down = True state.buttons.add(selected) pause = duration / steps / 1000 if duration else 0 for index in range(1, steps + 1): px = round(sx + (ex - sx) * index / steps) py = round(sy + (ey - sy) * index / steps) await page.mouse.move(px, py) state.x, state.y = px, py if pause and index < steps: await asyncio.sleep(pause) finally: if down: await page.mouse.up(button=selected) state.buttons.discard(selected) return _result( session, state, status="BROWSER_POINTER_DRAGGED", button=selected, duration_ms=duration, steps=steps, )