import asyncio
import base64
import os
import random
import json
import logging
import hashlib
from typing import (
Any,
Awaitable,
Callable,
Dict,
Optional,
Tuple,
Union,
cast,
List,
Literal,
)
import warnings
from playwright.async_api import Locator
from playwright.async_api import Error as PlaywrightError
from playwright.async_api import TimeoutError as PlaywrightTimeoutError
from playwright.async_api import Download, Page, BrowserContext
from .utils.animation_utils import AnimationUtilsPlaywright
from .utils.webpage_text_utils import WebpageTextUtilsPlaywright
from ..url_status_manager import UrlStatusManager
from .types import (
InteractiveRegion,
VisualViewport,
interactiveregion_from_dict,
visualviewport_from_dict,
)
warnings.filterwarnings(action="ignore", module="markitdown")
logger = logging.getLogger(__name__)
# Some of the Code for clicking coordinates and keypresses adapted from https://github.com/openai/openai-cua-sample-app/blob/main/computers/base_playwright.py
# Copyright 2025 OpenAI - MIT License
CUA_KEY_TO_PLAYWRIGHT_KEY = {
"/": "Divide",
"\\": "Backslash",
"alt": "Alt",
"arrowdown": "ArrowDown",
"arrowleft": "ArrowLeft",
"arrowright": "ArrowRight",
"arrowup": "ArrowUp",
"backspace": "Backspace",
"capslock": "CapsLock",
"cmd": "Meta",
"ctrl": "Control",
"delete": "Delete",
"end": "End",
"enter": "Enter",
"esc": "Escape",
"home": "Home",
"insert": "Insert",
"option": "Alt",
"pagedown": "PageDown",
"pageup": "PageUp",
"shift": "Shift",
"space": " ",
"super": "Meta",
"tab": "Tab",
"win": "Meta",
}
class PlaywrightController:
"""
A helper class to allow Playwright to interact with web pages to perform actions such as clicking, filling, and scrolling.
Args:
downloads_folder (str, optional): The folder to save downloads to. If None, downloads are not saved. Default: None
animate_actions (bool, optional): Whether to animate the actions (create fake cursor to click). Default: False
viewport_width (int, optional): The width of the viewport. Default: 1440
viewport_height (int, optional): The height of the viewport. Default: 1440
_download_handler (callable, None], optional): A function to handle downloads.
to_resize_viewport (bool, optional): Whether to resize the viewport. Default: True
timeout_load (int | float, optional): Amount of time (in secs) to wait before timeout on actions. Default: 1
sleep_after_action (int | float, optional): Amount of time (in secs) to sleep after performing action. Default: 1
single_tab_mode (bool, optional): If True, forces navigation to happen in the same tab rather than opening new tabs/windows. Default: False
url_status_manager (UrlStatusManager, optional): A list of websites to allow or deny. If None, all websites are allowed.
url_validation_callback (callable, optional): A callback function to validate URLs. It should return a tuple of (str, bool) where the str is a failure string and bool indicates if the URL is allowed.
"""
def __init__(
self,
downloads_folder: str | None = None,
animate_actions: bool = False,
viewport_width: int = 1440,
viewport_height: int = 1440,
_download_handler: Optional[Callable[[Download], None]] = None,
to_resize_viewport: bool = True,
timeout_load: Union[int, float] = 1,
sleep_after_action: Union[int, float] = 0.1,
single_tab_mode: bool = False,
url_status_manager: UrlStatusManager | None = None,
url_validation_callback: Optional[
Callable[[str], Awaitable[Tuple[str, bool]]]
] = None,
) -> None:
"""
Initialize the PlaywrightController.
"""
assert isinstance(animate_actions, bool)
assert isinstance(viewport_width, int)
assert isinstance(viewport_height, int)
assert viewport_height > 0
assert viewport_width > 0
assert timeout_load > 0
self.animate_actions = animate_actions
self.downloads_folder = downloads_folder
self.viewport_width = viewport_width
self.viewport_height = viewport_height
self._download_handler = _download_handler
self.to_resize_viewport = to_resize_viewport
self._timeout_load = timeout_load
self._sleep_after_action = sleep_after_action
self.single_tab_mode = single_tab_mode
self._url_status_manager = url_status_manager
self._url_validation_callback = url_validation_callback
self._page_script: str = ""
self._markdown_converter: Optional[Any] | None = None
# Create animation utils instance
self._animation = AnimationUtilsPlaywright()
# Use animation utils for cursor position tracking
self.last_cursor_position = self._animation.last_cursor_position
# Read page_script
with open(
os.path.join(os.path.abspath(os.path.dirname(__file__)), "page_script.js"),
"rt",
encoding="utf-8",
) as fh:
self._page_script = fh.read()
# Initialize WebpageTextUtils
self._text_utils = WebpageTextUtilsPlaywright()
async def on_new_page(self, page: Page) -> None:
"""
Handle actions to perform on a new page.
Args:
page (Page): The Playwright page object.
"""
assert page is not None
awaiting_approval = False
tentative_url = page.url
tentative_url_approved = True
# If the page is not whitelisted, block the site before asking if the user wants to allow it
if self._url_status_manager and not self._url_status_manager.is_url_allowed(
tentative_url
):
await page.route("**/*", lambda route: route.abort("blockedbyclient"))
try:
# This will raise an exception, but we don't care about it
await page.reload()
except PlaywrightError:
pass
await page.unroute("**/*")
if self._url_validation_callback is not None:
awaiting_approval = True
_, tentative_url_approved = await self._url_validation_callback(
tentative_url
)
# Wait for page load
try:
await page.wait_for_load_state(timeout=15000)
except PlaywrightTimeoutError:
logger.warning("Page load timeout, page might not be loaded")
# refresh
await page.reload()
try:
await page.wait_for_load_state(timeout=15000)
except Exception:
# stop page loading
await page.evaluate("window.stop()")
except Exception:
pass
if awaiting_approval and tentative_url_approved:
# Visit the page if permission has been given
await self.visit_page(page, tentative_url)
page.on("download", self._download_handler) # type: ignore
# check if there is a need to resize the viewport
page_viewport_size = page.viewport_size
if self.to_resize_viewport and self.viewport_width and self.viewport_height:
if (
page_viewport_size is None
or page_viewport_size["width"] != self.viewport_width
or page_viewport_size["height"] != self.viewport_height
):
await page.set_viewport_size(
{"width": self.viewport_width, "height": self.viewport_height}
)
await page.add_init_script(
path=os.path.join(
os.path.abspath(os.path.dirname(__file__)), "page_script.js"
)
)
async def _ensure_page_ready(self, page: Page) -> None:
"""
Ensure the page is properly configured before performing any action.
Args:
page (Page): The Playwright page object.
"""
assert page is not None
# bring the page to front
await page.bring_to_front()
await self.on_new_page(page)
async def get_current_url_title(self, page: Page) -> Tuple[str, str]:
"""
Get the current URL and title of the page.
Args:
page (Page): The Playwright page object.
Returns:
A tuple containing:
- str: The current URL of the page.
- str: The title of the page.
"""
try:
url = page.url
title = await page.title()
return url, title
except Exception:
logger.warning("Error getting current URL and title, returning unknown")
return "Unknown", "Unknown"
async def get_screenshot(self, page: Page, path: str | None = None) -> bytes:
"""
Capture a screenshot of the current page.
Args:
page (Page): The Playwright page object.
path (str, optional): The file path to save the screenshot. If None, the screenshot will be returned as bytes. Default: None
"""
try:
screenshot = await page.screenshot(path=path, timeout=7000)
return screenshot
except Exception:
logger.warning(
"Screenshot failed, page might not be loaded, stopping page and taking screenshot again"
)
# refresh the page
await page.reload()
await self._ensure_page_ready(page)
# try again
screenshot = await page.screenshot(path=path, timeout=10000)
return screenshot
async def sleep(self, page: Page, duration: Union[int, float]) -> None:
"""
Pause the execution for a specified duration.
Args:
page (Page): The Playwright page object.
duration (int | float): The duration to sleep in seconds.
"""
await self._ensure_page_ready(page)
await page.wait_for_timeout(duration * 1000)
async def get_interactive_rects(self, page: Page) -> Dict[str, InteractiveRegion]:
"""
Retrieve interactive regions from the web page.
Args:
page (Page): The Playwright page object.
Returns:
Dict[str, InteractiveRegion]: A dictionary of interactive regions.
"""
await self._ensure_page_ready(page)
# Read the regions from the DOM
try:
await page.evaluate(self._page_script)
except Exception:
pass
result = cast(
Dict[str, Dict[str, Any]],
await page.evaluate("WebSurfer.getInteractiveRects();"),
)
# Convert the results into appropriate types
assert isinstance(result, dict)
typed_results: Dict[str, InteractiveRegion] = {}
for k in result:
assert isinstance(k, str)
typed_results[k] = interactiveregion_from_dict(result[k])
return typed_results
async def get_visual_viewport(self, page: Page) -> VisualViewport:
"""
Retrieve the visual viewport of the web page.
Args:
page (Page): The Playwright page object.
Returns:
VisualViewport: The visual viewport of the page.
"""
await self._ensure_page_ready(page)
try:
await page.evaluate(self._page_script)
except Exception:
pass
return visualviewport_from_dict(
await page.evaluate("WebSurfer.getVisualViewport();")
)
async def get_focused_rect_id(self, page: Page) -> str:
"""
Retrieve the ID of the currently focused element.
Args:
page (Page): The Playwright page object.
Returns:
str: The ID of the focused element.
"""
await self._ensure_page_ready(page)
try:
await page.evaluate(self._page_script)
except Exception:
pass
result = await page.evaluate("WebSurfer.getFocusedElementId();")
return str(result)
async def get_page_metadata(self, page: Page) -> Dict[str, Any]:
"""
Retrieve metadata from the web page.
Args:
page (Page): The Playwright page object.
Returns:
Dict[str, Any]: A dictionary of page metadata.
"""
await self._ensure_page_ready(page)
try:
await page.evaluate(self._page_script)
except Exception:
pass
result = await page.evaluate("WebSurfer.getPageMetadata();")
assert isinstance(result, dict)
return cast(Dict[str, Any], result)
async def go_back(self, page: Page) -> bool:
"""
Navigate back to the previous page.
Args:
page (Page): The Playwright page object.
Returns:
bool: True if navigation was successful, False otherwise.
"""
await self._ensure_page_ready(page)
response = await page.go_back(
wait_until="load", timeout=self._timeout_load * 1000
)
if response is None:
return False
else:
return True
async def go_forward(self, page: Page) -> bool:
"""
Navigate forward to the next page.
Args:
page (Page): The Playwright page object.
Returns:
bool: True if navigation was successful, False otherwise.
"""
await self._ensure_page_ready(page)
response = await page.go_forward(
wait_until="load", timeout=self._timeout_load * 1000
)
if response is None:
return False
else:
return True
async def visit_page(self, page: Page, url: str) -> Tuple[bool, bool]:
"""
Visit a specified URL.
Args:
page (Page): The Playwright page object.
url (str): The URL to visit.
Returns:
A tuple containing:
- bool: Whether to reset prior metadata hash.
- bool: Whether to reset the last download.
"""
await self._ensure_page_ready(page)
reset_prior_metadata_hash = False
reset_last_download = False
download = None
download_future: asyncio.Task[Download] | None = None
# If downloads are enabled, start listening for a download event before navigation.
if self.downloads_folder:
try:
# Create a timeout for the download listener - use a longer timeout
download_future = asyncio.create_task(
page.wait_for_event( # type: ignore
"download", timeout=self._timeout_load * 5000
)
)
except Exception as e:
logger.warning(f"Failed to set up download listener: {e}")
download_future = None
try:
# Attempt normal navigation.
await page.goto(url)
await page.wait_for_load_state("load", timeout=self._timeout_load * 1000)
await asyncio.sleep(1)
reset_prior_metadata_hash = True
except Exception as e:
# If a navigation error occurs and it's likely due to a download,
# use the already waiting download event.
if self.downloads_folder and "net::ERR_ABORTED" in str(e):
if download_future:
try:
# Wait for download with a reasonable timeout
download = await asyncio.wait_for(
download_future, timeout=self._timeout_load * 1000
)
except asyncio.TimeoutError:
logger.warning(
"Download timeout exceeded, continuing without download"
)
except Exception as download_error:
logger.warning(f"Download error: {download_error}")
else:
raise
finally:
# Clean up the download future if it exists and hasn't completed
if download_future and not download_future.done():
download_future.cancel()
# Process any download that was detected
if download:
assert self.downloads_folder is not None
fname = os.path.join(self.downloads_folder, download.suggested_filename)
await download.save_as(fname)
message = (
f""
f"Successfully downloaded '{download.suggested_filename}' to local path:
{fname}
"
f""
)
data_uri = "data:text/html;base64," + base64.b64encode(
message.encode("utf-8")
).decode("utf-8")
await page.goto(data_uri)
reset_last_download = True
if self._sleep_after_action > 0:
await page.wait_for_timeout(self._sleep_after_action * 1000)
return reset_prior_metadata_hash, reset_last_download
async def refresh_page(self, page: Page) -> None:
"""
Refresh the current page.
Args:
page (Page): The Playwright page object.
"""
await self._ensure_page_ready(page)
await page.reload()
await page.wait_for_load_state("load", timeout=self._timeout_load * 1000)
async def page_down(self, page: Page) -> None:
"""
Scroll the page down by one viewport height minus 50 pixels.
Updates cursor position if animations are enabled.
When animations are enabled, scrolls smoothly instead of jumping.
Args:
page (Page): The Playwright page object.
"""
await self._ensure_page_ready(page)
scroll_amount = self.viewport_height - 50
if self.animate_actions:
# Smooth scrolling in smaller increments
steps = 5 # Reduced from 10 for faster animation
step_amount = scroll_amount / steps
for _ in range(steps):
await page.evaluate(f"window.scrollBy(0, {step_amount});")
await asyncio.sleep(0.02) # Reduced from 0.05 for faster animation
# Move cursor with the scroll using gradual animation
x, y = self.last_cursor_position
new_y = max(0, min(y - scroll_amount, self.viewport_height))
await self._animation.gradual_cursor_animation(page, x, y, x, new_y)
else:
# Regular instant scroll
await page.evaluate(f"window.scrollBy(0, {scroll_amount});")
async def page_up(self, page: Page) -> None:
"""
Scroll the page up by one viewport height minus 50 pixels.
Updates cursor position if animations are enabled.
When animations are enabled, scrolls smoothly instead of jumping.
Args:
page (Page): The Playwright page object.
"""
await self._ensure_page_ready(page)
scroll_amount = self.viewport_height - 50
if self.animate_actions:
# Smooth scrolling in smaller increments
steps = 5 # Reduced from 10 for faster animation
step_amount = scroll_amount / steps
for _ in range(steps):
await page.evaluate(f"window.scrollBy(0, -{step_amount});")
await asyncio.sleep(0.02) # Reduced from 0.05 for faster animation
# Move cursor with the scroll using gradual animation
x, y = self.last_cursor_position
new_y = max(0, min(y + scroll_amount, self.viewport_height))
await self._animation.gradual_cursor_animation(page, x, y, x, new_y)
else:
# Regular instant scroll
await page.evaluate(f"window.scrollBy(0, -{scroll_amount});")
async def scroll_mousewheel(
self, page: Page, direction: str, pixels: int = 400
) -> None:
"""
Scroll the page using mouse wheel for a specific number of pixels.
Updates cursor position if animations are enabled.
Args:
page (Page): The Playwright page object.
direction (str): Direction to scroll ("up" or "down").
pixels (int, optional): Number of pixels to scroll. Default: 400.
"""
await self._ensure_page_ready(page)
# Determine scroll direction (negative for up, positive for down)
scroll_amount = pixels if direction.lower() == "down" else -pixels
if self.animate_actions:
# Smooth scrolling in smaller increments
steps = 4
step_amount = scroll_amount / steps
for _ in range(steps):
await page.evaluate(f"window.scrollBy(0, {step_amount});")
await asyncio.sleep(0.015)
# Move cursor with the scroll using gradual animation
x, y = self.last_cursor_position
# Calculate new cursor position relative to scroll direction
cursor_offset = min(pixels / 3, 100) # Limit cursor movement
new_y = (
y - cursor_offset if direction.lower() == "down" else y + cursor_offset
)
new_y = max(0, min(new_y, self.viewport_height))
await self._animation.gradual_cursor_animation(page, x, y, x, new_y)
else:
# Regular instant scroll
await page.evaluate(f"window.scrollBy(0, {scroll_amount});")
if self._sleep_after_action > 0:
await page.wait_for_timeout(self._sleep_after_action * 1000)
async def click_id(
self,
context: BrowserContext,
page: Page,
identifier: str,
hold: float = 0.0,
button: Literal["left", "right"] = "left",
) -> Optional[Page]:
"""
Click an element with the given identifier. Handles regular clicks, right clicks, holding the mouse button, new page creation, and file downloads.
Args:
context (BrowserContext): The Playwright browser context.
page (Page): The Playwright page object.
identifier (str): The element identifier.
hold (float, optional): Seconds to hold the mouse button down before releasing. Default: 0.0
button (Literal["left", "right"], optional): Mouse button to use. Default: "left"
Returns:
Optional[Page]: The new page object if a popup was created, None otherwise.
Raises:
ValueError: If the element is not visible on the page.
TimeoutError: If the element does not appear within the timeout period.
"""
await self._ensure_page_ready(page)
new_page: Optional[Page] = None
selector = f"[__elementId='{identifier}']"
try:
# Wait for the element to be visible and scroll it into view
await page.wait_for_selector(
selector, state="visible", timeout=self._timeout_load * 1000
)
target = page.locator(selector)
await target.scroll_into_view_if_needed()
except PlaywrightTimeoutError:
raise ValueError(
f"Element with identifier {identifier} not found or not visible"
)
# Retrieve bounding box to determine the center for clicking
box = await target.bounding_box()
if not box:
raise ValueError(
f"Element with identifier {identifier} is not visible on the page."
)
center_x = box["x"] + box["width"] / 2
center_y = box["y"] + box["height"] / 2
# In single tab mode, override target attributes to avoid opening a new tab
if self.single_tab_mode:
await target.evaluate("""
el => {
// Remove target attribute from clicked element and all _blank links/forms
el.removeAttribute('target');
document.querySelectorAll('a[target=_blank], form[target=_blank]')
.forEach(e => e.removeAttribute('target'));
}
""")
download = None
download_future: asyncio.Task[Download] | None = None
# Start listening for a download event if downloads are enabled
if self.downloads_folder:
try:
download_future = asyncio.create_task(
page.wait_for_event( # type: ignore
"download", timeout=self._timeout_load * 2000
)
)
except Exception as e:
logger.warning(f"Failed to set up download listener: {e}")
download_future = None
async def perform_click() -> Optional[Page]:
nonlocal download
try:
if self.single_tab_mode:
await page.mouse.move(center_x, center_y, steps=1)
if hold == 0.0 and button == "left":
await page.mouse.click(center_x, center_y)
else:
await page.mouse.down(button=button)
if hold > 0:
await asyncio.sleep(hold)
await page.mouse.up(button=button)
return None
else:
# Create a task to wait for a new page event
new_page_promise: asyncio.Task[Page] = asyncio.create_task(
context.wait_for_event( # type: ignore
"page", timeout=self._timeout_load * 1000
)
)
# Perform the click
await page.mouse.move(center_x, center_y, steps=1)
if hold == 0.0 and button == "left":
await page.mouse.click(center_x, center_y, delay=10)
else:
await page.mouse.down(button=button)
if hold > 0:
await asyncio.sleep(hold)
await page.mouse.up(button=button)
try:
# Wait for the new page to open
new_page = await new_page_promise
await self._ensure_page_ready(new_page)
return new_page
except PlaywrightTimeoutError:
# No new page opened within timeout
return None
except Exception:
raise
# Optionally animate the click
if self.animate_actions:
await self.add_cursor_box(page, identifier)
start_x, start_y = self.last_cursor_position
await self.gradual_cursor_animation(
page, start_x, start_y, center_x, center_y
)
new_page = await perform_click()
# Handle any download that occurred
if download_future:
try:
if not download:
# Use asyncio.wait_for with a reasonable timeout
try:
download = await asyncio.wait_for(
download_future, timeout=self._timeout_load * 1000
)
except asyncio.TimeoutError:
# No download occurred within the timeout period
logger.debug("No download detected within timeout period")
pass
if download:
logger.info(
f"Downloading {download.suggested_filename} to {self.downloads_folder}"
)
assert self.downloads_folder is not None
fname = os.path.join(
self.downloads_folder, download.suggested_filename
)
await download.save_as(fname)
except Exception as e:
logger.debug(f"Error handling download: {e}")
finally:
if not download_future.done():
download_future.cancel()
if self.animate_actions:
await self.remove_cursor_box(page, identifier)
if self._sleep_after_action > 0:
await page.wait_for_timeout(self._sleep_after_action * 1000)
return new_page
async def hover_id(self, page: Page, identifier: str) -> None:
"""
Hover the mouse over the element with the given identifier.
Args:
page (Page): The Playwright page object.
identifier (str): The element identifier.
"""
await self._ensure_page_ready(page)
target = page.locator(f"[__elementId='{identifier}']")
# See if it exists
try:
await target.wait_for(timeout=5000)
except PlaywrightTimeoutError:
raise ValueError("No such element.") from None
# Hover over it
await target.scroll_into_view_if_needed()
await asyncio.sleep(0.3)
box = cast(Dict[str, Union[int, float]], await target.bounding_box())
try:
if self.animate_actions:
await self.add_cursor_box(page, identifier)
# Move cursor to the box slowly
start_x, start_y = self.last_cursor_position
end_x, end_y = box["x"] + box["width"] / 2, box["y"] + box["height"] / 2
await self.gradual_cursor_animation(
page, start_x, start_y, end_x, end_y
)
await asyncio.sleep(0.1)
await page.mouse.move(
box["x"] + box["width"] / 2, box["y"] + box["height"] / 2
)
await self.remove_cursor_box(page, identifier)
else:
await page.mouse.move(
box["x"] + box["width"] / 2, box["y"] + box["height"] / 2
)
except Exception:
if self.animate_actions:
await self.remove_cursor_box(page, identifier)
await target.hover()
async def fill_id(
self,
page: Page,
identifier: str,
value: str,
press_enter: bool = True,
delete_existing_text: bool = False,
) -> None:
"""
Fill the element with the given identifier with the specified value.
Works with text inputs, textareas, and comboboxes.
Args:
page (Page): The Playwright page object.
identifier (str): The element identifier.
value (str): The value to fill.
press_enter (bool, optional): Whether to press enter after filling. Default: True
delete_existing_text (bool, optional): Whether to delete existing text before filling. Default: False
"""
await self._ensure_page_ready(page)
await page.wait_for_selector(f"[__elementId='{identifier}']", state="visible")
target = page.locator(f"[__elementId='{identifier}']")
await target.scroll_into_view_if_needed()
# See if it exists
try:
await target.wait_for(timeout=5000)
except PlaywrightTimeoutError:
raise ValueError("No such element.") from None
# Fill it
box = cast(Dict[str, Union[int, float]], await target.bounding_box())
if self.single_tab_mode:
# Remove target attributes to prevent new tabs
await target.evaluate("""
el => el.removeAttribute('target')
// Remove 'target' on all tags
for (const a of document.querySelectorAll('a[target=_blank]')) {
a.removeAttribute('target');
}
// Remove 'target' on all