# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
import asyncio
import datetime
import io
import json
import os
import random
import re
import shutil
import urllib.parse
from copy import deepcopy
from typing import (
TYPE_CHECKING,
Any,
BinaryIO,
Coroutine,
Dict,
List,
Literal,
Optional,
Tuple,
TypedDict,
Union,
cast,
)
from PIL import Image, ImageDraw, ImageFont
if TYPE_CHECKING:
from camel.agents import ChatAgent
from camel.logger import get_logger
from camel.messages import BaseMessage
from camel.models import BaseModelBackend, ModelFactory
from camel.toolkits.base import BaseToolkit
from camel.toolkits.function_tool import FunctionTool
from camel.toolkits.video_analysis_toolkit import VideoAnalysisToolkit
from camel.types import ModelPlatformType, ModelType
from camel.utils import (
dependencies_required,
retry_on_error,
sanitize_filename,
)
logger = get_logger(__name__)
TOP_NO_LABEL_ZONE = 20
AVAILABLE_ACTIONS_PROMPT = """
1. `fill_input_id(identifier: Union[str, int], text: str)`: Fill an input
field (e.g. search box) with the given text and press Enter.
2. `click_id(identifier: Union[str, int])`: Click an element with the given ID.
3. `hover_id(identifier: Union[str, int])`: Hover over an element with the
given ID.
4. `download_file_id(identifier: Union[str, int])`: Download a file with the
given ID. It returns the path to the downloaded file. If the file is
successfully downloaded, you can stop the simulation and report the path to
the downloaded file for further processing.
5. `scroll_to_bottom()`: Scroll to the bottom of the page.
6. `scroll_to_top()`: Scroll to the top of the page.
7. `scroll_up()`: Scroll up the page. It is suitable when you want to see the
elements above the current viewport.
8. `scroll_down()`: Scroll down the page. It is suitable when you want to see
the elements below the current viewport. If the webpage does not change, It
means that the webpage has scrolled to the bottom.
9. `back()`: Navigate back to the previous page. This is useful when you want
to go back to the previous page, as current page is not useful.
10. `stop()`: Stop the action process, because the task is completed or failed
(impossible to find the answer). In this situation, you should provide your
answer in your output.
11. `get_url()`: Get the current URL of the current page.
12. `find_text_on_page(search_text: str)`: Find the next given text on the
current whole page, and scroll the page to the targeted text. It is equivalent
to pressing Ctrl + F and searching for the text, and is powerful when you want
to fast-check whether the current page contains some specific text.
13. `visit_page(url: str)`: Go to the specific url page.
14. `click_blank_area()`: Click a blank area of the page to unfocus the
current element. It is useful when you have clicked an element but it cannot
unfocus itself (e.g. Menu bar) to automatically render the updated webpage.
15. `ask_question_about_video(question: str)`: Ask a question about the
current webpage which contains video, e.g. youtube websites.
"""
ASYNC_ACTIONS = [
"fill_input_id",
"click_id",
"hover_id",
"download_file_id",
"scroll_up",
"scroll_down",
"scroll_to_bottom",
"scroll_to_top",
"back",
"stop",
"find_text_on_page",
"visit_page",
"click_blank_area",
]
ACTION_WITH_FEEDBACK_LIST = [
'ask_question_about_video',
'download_file_id',
'find_text_on_page',
]
# Code from magentic-one
class DOMRectangle(TypedDict):
x: Union[int, float]
y: Union[int, float]
width: Union[int, float]
height: Union[int, float]
top: Union[int, float]
right: Union[int, float]
bottom: Union[int, float]
left: Union[int, float]
class VisualViewport(TypedDict):
height: Union[int, float]
width: Union[int, float]
offsetLeft: Union[int, float]
offsetTop: Union[int, float]
pageLeft: Union[int, float]
pageTop: Union[int, float]
scale: Union[int, float]
clientWidth: Union[int, float]
clientHeight: Union[int, float]
scrollWidth: Union[int, float]
scrollHeight: Union[int, float]
class InteractiveRegion(TypedDict):
tag_name: str
role: str
aria_name: str
v_scrollable: bool
rects: List[DOMRectangle]
def extract_function_name(s: str) -> str:
r"""Extract the pure function name from a string (without parameters or
parentheses)
Args:
s (str): Input string, e.g., `1.`**`click_id(14)`**, `scroll_up()`,
`'visit_page(url)'`, etc.
Returns:
str: Pure function name (e.g., `click_id`, `scroll_up`, `visit_page`)
"""
# 1. Strip leading/trailing whitespace and enclosing backticks or quotes
s = s.strip().strip('`"\'')
# Strip any leading numeric prefix like " 12. " or "3. "
s = re.sub(r'^\s*\d+\.\s*', '', s)
# 3. Match a Python-valid identifier followed by an opening parenthesis
match = re.match(r'^([A-Za-z_]\w*)\s*\(', s)
if match:
return match.group(1)
# 4. Fallback: take everything before the first space or '('
return re.split(r'[ (\n]', s, maxsplit=1)[0]
def _get_str(d: Any, k: str) -> str:
r"""Safely retrieve a string value from a dictionary."""
if k not in d:
raise KeyError(f"Missing required key: '{k}'")
val = d[k]
if isinstance(val, str):
return val
raise TypeError(
f"Expected a string for key '{k}', " f"but got {type(val).__name__}"
)
def _get_number(d: Any, k: str) -> Union[int, float]:
r"""Safely retrieve a number (int or float) from a dictionary"""
val = d[k]
if isinstance(val, (int, float)):
return val
raise TypeError(
f"Expected a number (int/float) for key "
f"'{k}', but got {type(val).__name__}"
)
def _get_bool(d: Any, k: str) -> bool:
r"""Safely retrieve a boolean value from a dictionary."""
val = d[k]
if isinstance(val, bool):
return val
raise TypeError(
f"Expected a boolean for key '{k}', " f"but got {type(val).__name__}"
)
def _parse_json_output(text: str) -> Dict[str, Any]:
r"""Extract JSON output from a string."""
markdown_pattern = r'```(?:json)?\s*(.*?)\s*```'
markdown_match = re.search(markdown_pattern, text, re.DOTALL)
if markdown_match:
text = markdown_match.group(1).strip()
triple_quotes_pattern = r'"""(?:json)?\s*(.*?)\s*"""'
triple_quotes_match = re.search(triple_quotes_pattern, text, re.DOTALL)
if triple_quotes_match:
text = triple_quotes_match.group(1).strip()
try:
return json.loads(text)
except json.JSONDecodeError:
try:
fixed_text = re.sub(
r'`([^`]*?)`(?=\s*[:,\[\]{}]|$)', r'"\1"', text
)
return json.loads(fixed_text)
except json.JSONDecodeError:
result = {}
try:
bool_pattern = r'"(\w+)"\s*:\s*(true|false)'
for match in re.finditer(bool_pattern, text, re.IGNORECASE):
key, value = match.groups()
result[key] = value.lower() == "true"
str_pattern = r'"(\w+)"\s*:\s*"([^"]*)"'
for match in re.finditer(str_pattern, text):
key, value = match.groups()
result[key] = value
num_pattern = r'"(\w+)"\s*:\s*(-?\d+(?:\.\d+)?)'
for match in re.finditer(num_pattern, text):
key, value = match.groups()
try:
result[key] = int(value)
except ValueError:
result[key] = float(value)
empty_str_pattern = r'"(\w+)"\s*:\s*""'
for match in re.finditer(empty_str_pattern, text):
key = match.group(1)
result[key] = ""
if result:
return result
logger.warning(f"Failed to parse JSON output: {text}")
return {}
except Exception as e:
logger.warning(f"Error while extracting fields from JSON: {e}")
return {}
def _reload_image(image: Image.Image):
buffer = io.BytesIO()
image.save(buffer, format="PNG")
buffer.seek(0)
return Image.open(buffer)
def dom_rectangle_from_dict(rect: Dict[str, Any]) -> DOMRectangle:
r"""Create a DOMRectangle object from a dictionary."""
return DOMRectangle(
x=_get_number(rect, "x"),
y=_get_number(rect, "y"),
width=_get_number(rect, "width"),
height=_get_number(rect, "height"),
top=_get_number(rect, "top"),
right=_get_number(rect, "right"),
bottom=_get_number(rect, "bottom"),
left=_get_number(rect, "left"),
)
def interactive_region_from_dict(region: Dict[str, Any]) -> InteractiveRegion:
r"""Create an :class:`InteractiveRegion` object from a dictionary."""
typed_rects: List[DOMRectangle] = []
for rect in region["rects"]:
typed_rects.append(dom_rectangle_from_dict(rect))
return InteractiveRegion(
tag_name=_get_str(region, "tag_name"),
role=_get_str(region, "role"),
aria_name=_get_str(region, "aria-name"),
v_scrollable=_get_bool(region, "v-scrollable"),
rects=typed_rects,
)
def visual_viewport_from_dict(viewport: Dict[str, Any]) -> VisualViewport:
r"""Create a :class:`VisualViewport` object from a dictionary."""
return VisualViewport(
height=_get_number(viewport, "height"),
width=_get_number(viewport, "width"),
offsetLeft=_get_number(viewport, "offsetLeft"),
offsetTop=_get_number(viewport, "offsetTop"),
pageLeft=_get_number(viewport, "pageLeft"),
pageTop=_get_number(viewport, "pageTop"),
scale=_get_number(viewport, "scale"),
clientWidth=_get_number(viewport, "clientWidth"),
clientHeight=_get_number(viewport, "clientHeight"),
scrollWidth=_get_number(viewport, "scrollWidth"),
scrollHeight=_get_number(viewport, "scrollHeight"),
)
def add_set_of_mark(
screenshot: Union[bytes, Image.Image, io.BufferedIOBase],
ROIs: Dict[str, InteractiveRegion],
) -> Tuple[Image.Image, List[str], List[str], List[str]]:
if isinstance(screenshot, Image.Image):
return _add_set_of_mark(screenshot, ROIs)
if isinstance(screenshot, bytes):
screenshot = io.BytesIO(screenshot)
image = Image.open(cast(BinaryIO, screenshot))
comp, visible_rects, rects_above, rects_below = _add_set_of_mark(
image, ROIs
)
image.close()
return comp, visible_rects, rects_above, rects_below
def _add_set_of_mark(
screenshot: Image.Image, ROIs: Dict[str, InteractiveRegion]
) -> Tuple[Image.Image, List[str], List[str], List[str]]:
r"""Add a set of marks to the screenshot.
Args:
screenshot (Image.Image): The screenshot to add marks to.
ROIs (Dict[str, InteractiveRegion]): The regions to add marks to.
Returns:
Tuple[Image.Image, List[str], List[str], List[str]]: A tuple
containing the screenshot with marked ROIs, ROIs fully within the
images, ROIs located above the visible area, and ROIs located below
the visible area.
"""
visible_rects: List[str] = list()
rects_above: List[str] = list() # Scroll up to see
rects_below: List[str] = list() # Scroll down to see
fnt = ImageFont.load_default(14)
base = screenshot.convert("L").convert("RGBA")
overlay = Image.new("RGBA", base.size)
draw = ImageDraw.Draw(overlay)
for r in ROIs:
for rect in ROIs[r]["rects"]:
# Empty rectangles
if not rect or rect["width"] == 0 or rect["height"] == 0:
continue
# TODO: add scroll left and right?
horizontal_center = (rect["right"] + rect["left"]) / 2.0
vertical_center = (rect["top"] + rect["bottom"]) / 2.0
is_within_horizon = 0 <= horizontal_center < base.size[0]
is_above_viewport = vertical_center < 0
is_below_viewport = vertical_center >= base.size[1]
if is_within_horizon:
if is_above_viewport:
rects_above.append(r)
elif is_below_viewport:
rects_below.append(r)
else: # Fully visible
visible_rects.append(r)
_draw_roi(draw, int(r), fnt, rect)
comp = Image.alpha_composite(base, overlay)
overlay.close()
return comp, visible_rects, rects_above, rects_below
def _draw_roi(
draw: ImageDraw.ImageDraw,
idx: int,
font: ImageFont.FreeTypeFont | ImageFont.ImageFont,
rect: DOMRectangle,
) -> None:
r"""Draw a ROI on the image.
Args:
draw (ImageDraw.ImageDraw): The draw object.
idx (int): The index of the ROI.
font (ImageFont.FreeTypeFont | ImageFont.ImageFont): The font.
rect (DOMRectangle): The DOM rectangle.
"""
color = _get_random_color(idx)
text_color = _get_text_color(color)
roi = ((rect["left"], rect["top"]), (rect["right"], rect["bottom"]))
label_location = (rect["right"], rect["top"])
label_anchor = "rb"
if label_location[1] <= TOP_NO_LABEL_ZONE:
label_location = (rect["right"], rect["bottom"])
label_anchor = "rt"
draw.rectangle(
roi, outline=color, fill=(color[0], color[1], color[2], 48), width=2
)
bbox = draw.textbbox(
label_location,
str(idx),
font=font,
anchor=label_anchor,
align="center",
)
bbox = (bbox[0] - 3, bbox[1] - 3, bbox[2] + 3, bbox[3] + 3)
draw.rectangle(bbox, fill=color)
draw.text(
label_location,
str(idx),
fill=text_color,
font=font,
anchor=label_anchor,
align="center",
)
def _get_text_color(
bg_color: Tuple[int, int, int, int],
) -> Tuple[int, int, int, int]:
r"""Determine the ideal text color (black or white) for contrast.
Args:
bg_color: The background color (R, G, B, A).
Returns:
A tuple representing black or white color for text.
"""
luminance = bg_color[0] * 0.3 + bg_color[1] * 0.59 + bg_color[2] * 0.11
return (0, 0, 0, 255) if luminance > 120 else (255, 255, 255, 255)
def _get_random_color(identifier: int) -> Tuple[int, int, int, int]:
r"""Generate a consistent random RGBA color based on the identifier.
Args:
identifier: The ID used as a seed to ensure color consistency.
Returns:
A tuple representing (R, G, B, A) values.
"""
rnd = random.Random(int(identifier))
r = rnd.randint(0, 255)
g = rnd.randint(125, 255)
b = rnd.randint(0, 50)
color = [r, g, b]
# TODO: check why shuffle is needed?
rnd.shuffle(color)
color.append(255)
return cast(Tuple[int, int, int, int], tuple(color))
class AsyncBaseBrowser:
def __init__(
self,
headless=True,
cache_dir: Optional[str] = None,
channel: Literal["chrome", "msedge", "chromium"] = "chromium",
cookie_json_path: Optional[str] = None,
):
r"""
Initialize the asynchronous browser core.
Args:
headless (bool): Whether to run the browser in headless mode.
cache_dir (Union[str, None]): The directory to store cache files.
channel (Literal["chrome", "msedge", "chromium"]): The browser
channel to use. Must be one of "chrome", "msedge", or
"chromium".
cookie_json_path (Optional[str]): Path to a JSON file containing
authentication cookies and browser storage state. If provided
and the file exists, the browser will load this state to
maintain authenticated sessions without requiring manual login.
Returns:
None
"""
from playwright.async_api import (
async_playwright,
)
self.history: list = []
self.headless = headless
self.channel = channel
self.playwright = async_playwright()
self.page_history: list = []
self.cookie_json_path = cookie_json_path
# Set the cache directory
self.cache_dir = "tmp/" if cache_dir is None else cache_dir
os.makedirs(self.cache_dir, exist_ok=True)
# Load the page script
abs_dir_path = os.path.dirname(os.path.abspath(__file__))
page_script_path = os.path.join(abs_dir_path, "page_script.js")
try:
with open(page_script_path, "r", encoding='utf-8') as f:
self.page_script = f.read()
f.close()
except FileNotFoundError:
raise FileNotFoundError(
f"Page script file not found at path: {page_script_path}"
)
async def async_init(self) -> None:
r"""Asynchronously initialize the browser."""
# Start Playwright asynchronously (only needed in async mode).
if not getattr(self, "playwright_started", False):
await self._ensure_browser_installed()
self.playwright_server = await self.playwright.start()
self.playwright_started = True
# Launch the browser asynchronously.
self.browser = await self.playwright_server.chromium.launch(
headless=self.headless, channel=self.channel
)
# Check if cookie file exists before using it to maintain
# authenticated sessions. This prevents errors when the cookie file
# doesn't exist
if self.cookie_json_path and os.path.exists(self.cookie_json_path):
self.context = await self.browser.new_context(
accept_downloads=True, storage_state=self.cookie_json_path
)
else:
self.context = await self.browser.new_context(
accept_downloads=True,
)
# Create a new page asynchronously.
self.page = await self.context.new_page()
def init(self) -> Coroutine[Any, Any, None]:
r"""Initialize the browser asynchronously."""
return self.async_init()
def clean_cache(self) -> None:
r"""Delete the cache directory and its contents."""
if os.path.exists(self.cache_dir):
shutil.rmtree(self.cache_dir)
async def async_wait_for_load(self, timeout: int = 20) -> None:
r"""
Asynchronously Wait for a certain amount of time for the page to load.
Args:
timeout (int): Timeout in seconds.
"""
timeout_ms = timeout * 1000
await self.page.wait_for_load_state("load", timeout=timeout_ms)
# TODO: check if this is needed
await asyncio.sleep(2)
def wait_for_load(self, timeout: int = 20) -> Coroutine[Any, Any, None]:
r"""Wait for a certain amount of time for the page to load.
Args:
timeout (int): Timeout in seconds.
"""
return self.async_wait_for_load(timeout)
async def async_click_blank_area(self) -> None:
r"""Asynchronously click a blank area of the page to unfocus
the current element."""
await self.page.mouse.click(0, 0)
await self.wait_for_load()
def click_blank_area(self) -> Coroutine[Any, Any, None]:
r"""Click a blank area of the page to unfocus the current element."""
return self.async_click_blank_area()
async def async_visit_page(self, url: str) -> None:
r"""Visit a page with the given URL."""
await self.page.goto(url)
await self.wait_for_load()
self.page_url = url
@retry_on_error()
def visit_page(self, url: str) -> Coroutine[Any, Any, None]:
r"""Visit a page with the given URL."""
return self.async_visit_page(url)
def ask_question_about_video(self, question: str) -> str:
r"""Ask a question about the video on the current page,
such as YouTube video.
Args:
question (str): The question to ask.
Returns:
str: The answer to the question.
"""
current_url = self.get_url()
# Confirm with user before proceeding due to potential slow
# processing time
confirmation_message = (
f"Do you want to analyze the video on the current "
f"page({current_url})? This operation may take a long time.(y/n): "
)
user_confirmation = input(confirmation_message)
if user_confirmation.lower() not in ['y', 'yes']:
return "User cancelled the video analysis."
model = None
if hasattr(self, 'web_agent_model'):
model = self.web_agent_model
video_analyzer = VideoAnalysisToolkit(model=model)
result = video_analyzer.ask_question_about_video(current_url, question)
return result
@retry_on_error()
async def async_get_screenshot(
self, save_image: bool = False
) -> Tuple[Image.Image, Union[str, None]]:
r"""Asynchronously get a screenshot of the current page.
Args:
save_image (bool): Whether to save the image to the cache
directory.
Returns:
Tuple[Image.Image, str]: A tuple containing the screenshot
image and the path to the image file if saved, otherwise
:obj:`None`.
"""
image_data = await self.page.screenshot(timeout=60000)
image = Image.open(io.BytesIO(image_data))
file_path = None
if save_image:
# Get url name to form a file name
# Use urlparser for a safer extraction the url name
parsed_url = urllib.parse.urlparse(self.page_url)
# Max length is set to 241 as there are 10 characters for the
# timestamp and 4 characters for the file extension:
url_name = sanitize_filename(str(parsed_url.path), max_length=241)
timestamp = datetime.datetime.now().strftime("%m%d%H%M%S")
file_path = os.path.join(
self.cache_dir, f"{url_name}_{timestamp}.png"
)
with open(file_path, "wb") as f:
image.save(f, "PNG")
f.close()
return image, file_path
@retry_on_error()
def get_screenshot(
self, save_image: bool = False
) -> Coroutine[Any, Any, Tuple[Image.Image, Union[str, None]]]:
r"""Get a screenshot of the current page.
Args:
save_image (bool): Whether to save the image to the cache
directory.
Returns:
Tuple[Image.Image, str]: A tuple containing the screenshot
image and the path to the image file if saved, otherwise
:obj:`None`.
"""
return self.async_get_screenshot(save_image)
async def async_capture_full_page_screenshots(
self, scroll_ratio: float = 0.8
) -> List[str]:
r"""Asynchronously capture full page screenshots by scrolling the
page with a buffer zone.
Args:
scroll_ratio (float): The ratio of viewport height to scroll each
step (default: 0.8).
Returns:
List[str]: A list of paths to the captured screenshots.
"""
screenshots = []
scroll_height = await self.page.evaluate("document.body.scrollHeight")
assert self.page.viewport_size is not None
viewport_height = self.page.viewport_size["height"]
current_scroll = 0
screenshot_index = 1
max_height = scroll_height - viewport_height
scroll_step = int(viewport_height * scroll_ratio)
last_height = 0
while True:
logger.debug(
f"Current scroll: {current_scroll}, max_height: "
f"{max_height}, step: {scroll_step}"
)
_, file_path = await self.get_screenshot(save_image=True)
screenshots.append(file_path)
await self.page.evaluate(f"window.scrollBy(0, {scroll_step})")
# Allow time for content to load
await asyncio.sleep(0.5)
current_scroll = await self.page.evaluate("window.scrollY")
# Break if there is no significant scroll
if abs(current_scroll - last_height) < viewport_height * 0.1:
break
last_height = current_scroll
screenshot_index += 1
return screenshots
def capture_full_page_screenshots(
self, scroll_ratio: float = 0.8
) -> Coroutine[Any, Any, List[str]]:
r"""Capture full page screenshots by scrolling the page with
a buffer zone.
Args:
scroll_ratio (float): The ratio of viewport height to scroll each
step (default: 0.8).
Returns:
List[str]: A list of paths to the captured screenshots.
"""
return self.async_capture_full_page_screenshots(scroll_ratio)
async def async_get_visual_viewport(self) -> VisualViewport:
r"""Asynchronously get the visual viewport of the current page.
Returns:
VisualViewport: The visual viewport of the current page.
"""
try:
await self.page.evaluate(self.page_script)
except Exception as e:
logger.warning(f"Error evaluating page script: {e}")
return visual_viewport_from_dict(
await self.page.evaluate(
"MultimodalWebSurfer.getVisualViewport();"
)
)
def get_visual_viewport(self) -> Coroutine[Any, Any, VisualViewport]:
r"""Get the visual viewport of the current page."""
return self.async_get_visual_viewport()
async def async_get_interactive_elements(
self,
) -> Dict[str, InteractiveRegion]:
r"""Asynchronously get the interactive elements of the current page.
Returns:
Dict[str, InteractiveRegion]: A dictionary containing the
interactive elements of the current page.
"""
try:
await self.page.evaluate(self.page_script)
except Exception as e:
logger.warning(f"Error evaluating page script: {e}")
result = cast(
Dict[str, Dict[str, Any]],
await self.page.evaluate(
"MultimodalWebSurfer.getInteractiveRects();"
),
)
typed_results: Dict[str, InteractiveRegion] = {}
for k in result:
typed_results[k] = interactive_region_from_dict(result[k])
return typed_results # type: ignore[return-value]
def get_interactive_elements(
self,
) -> Coroutine[Any, Any, Dict[str, InteractiveRegion]]:
r"""Get the interactive elements of the current page.
Returns:
Dict[str, InteractiveRegion]: A dictionary of interactive elements.
"""
return self.async_get_interactive_elements()
async def async_get_som_screenshot(
self,
save_image: bool = False,
) -> Tuple[Image.Image, Union[str, None]]:
r"""Asynchronously get a screenshot of the current viewport
with interactive elements marked.
Args:
save_image (bool): Whether to save the image to the cache
directory.
Returns:
Tuple[Image.Image, str]: A tuple containing the screenshot
image and the path to the image file.
"""
await self.wait_for_load()
screenshot, _ = await self.get_screenshot(save_image=False)
rects = await self.get_interactive_elements()
file_path: str | None = None
comp, _, _, _ = add_set_of_mark(
screenshot,
rects, # type: ignore[arg-type]
)
if save_image:
parsed_url = urllib.parse.urlparse(self.page_url)
# Max length is set to 241 as there are 10 characters for the
# timestamp and 4 characters for the file extension:
url_name = sanitize_filename(str(parsed_url.path), max_length=241)
timestamp = datetime.datetime.now().strftime("%m%d%H%M%S")
file_path = os.path.join(
self.cache_dir, f"{url_name}_{timestamp}.png"
)
with open(file_path, "wb") as f:
comp.save(f, "PNG")
f.close()
return comp, file_path
def get_som_screenshot(
self,
save_image: bool = False,
) -> Coroutine[Any, Any, Tuple[Image.Image, Union[str, None]]]:
r"""Get a screenshot of the current viewport with interactive elements
marked.
Args:
save_image (bool): Whether to save the image to the cache
directory.
Returns:
Tuple[Image.Image, str]: A tuple containing the screenshot image
and the path to the image file.
"""
return self.async_get_som_screenshot(save_image)
async def async_scroll_up(self) -> None:
r"""Asynchronously scroll up the page."""
await self.page.keyboard.press("PageUp")
def scroll_up(self) -> Coroutine[Any, Any, None]:
r"""Scroll up the page."""
return self.async_scroll_up()
async def async_scroll_down(self) -> None:
r"""Asynchronously scroll down the page."""
await self.page.keyboard.press("PageDown")
def scroll_down(self) -> Coroutine[Any, Any, None]:
r"""Scroll down the page."""
return self.async_scroll_down()
def get_url(self) -> str:
r"""Get the URL of the current page."""
return self.page.url
async def async_click_id(self, identifier: Union[str, int]) -> None:
r"""Asynchronously click an element with the given ID.
Args:
identifier (Union[str, int]): The ID of the element to click.
"""
if isinstance(identifier, int):
identifier = str(identifier)
target = self.page.locator(f"[__elementId='{identifier}']")
try:
await target.wait_for(timeout=5000)
except (TimeoutError, Exception) as e:
logger.debug(f"Error during click operation: {e}")
raise ValueError("No such element.") from None
await target.scroll_into_view_if_needed()
new_page = None
try:
async with self.page.expect_event(
"popup", timeout=1000
) as page_info:
box = cast(
Dict[str, Union[int, float]], await target.bounding_box()
)
await self.page.mouse.click(
box["x"] + box["width"] / 2, box["y"] + box["height"] / 2
)
new_page = await page_info.value
# If a new page is opened, switch to it
if new_page:
self.page_history.append(deepcopy(self.page.url))
self.page = new_page
except (TimeoutError, Exception) as e:
logger.debug(f"Error during click operation: {e}")
pass
await self.wait_for_load()
def click_id(
self, identifier: Union[str, int]
) -> Coroutine[Any, Any, None]:
r"""Click an element with the given identifier."""
return self.async_click_id(identifier)
async def async_extract_url_content(self) -> str:
r"""Asynchronously extract the content of the current page."""
content = await self.page.content()
return content
def extract_url_content(self) -> Coroutine[Any, Any, str]:
r"""Extract the content of the current page."""
return self.async_extract_url_content()
async def async_download_file_id(self, identifier: Union[str, int]) -> str:
r"""Asynchronously download a file with the given selector.
Args:
identifier (Union[str, int]): The identifier of the file
to download.
Returns:
str: The path to the downloaded file.
"""
if isinstance(identifier, int):
identifier = str(identifier)
try:
target = self.page.locator(f"[__elementId='{identifier}']")
except (TimeoutError, Exception) as e:
logger.debug(f"Error during download operation: {e}")
logger.warning(
f"Element with identifier '{identifier}' not found."
)
return f"Element with identifier '{identifier}' not found."
await target.scroll_into_view_if_needed()
file_path = os.path.join(self.cache_dir)
await self.wait_for_load()
try:
async with self.page.expect_download(
timeout=5000
) as download_info:
await target.click()
download = await download_info.value
file_name = download.suggested_filename
file_path = os.path.join(file_path, file_name)
await download.save_as(file_path)
return f"Downloaded file to path '{file_path}'."
except Exception as e:
logger.debug(f"Error during download operation: {e}")
return f"Failed to download file with identifier '{identifier}'."
def download_file_id(
self, identifier: Union[str, int]
) -> Coroutine[Any, Any, str]:
r"""Download a file with the given identifier."""
return self.async_download_file_id(identifier)
async def async_fill_input_id(
self, identifier: Union[str, int], text: str
) -> str:
r"""Asynchronously fill an input field with the given text, and then
press Enter.
Args:
identifier (Union[str, int]): The identifier of the input field.
text (str): The text to fill.
Returns:
str: The result of the action.
"""
if isinstance(identifier, int):
identifier = str(identifier)
try:
target = self.page.locator(f"[__elementId='{identifier}']")
except (TimeoutError, Exception) as e:
logger.debug(f"Error during fill operation: {e}")
logger.warning(
f"Element with identifier '{identifier}' not found."
)
return f"Element with identifier '{identifier}' not found."
await target.scroll_into_view_if_needed()
await target.focus()
try:
await target.fill(text)
except Exception as e:
logger.debug(f"Error during fill operation: {e}")
await target.press_sequentially(text)
await target.press("Enter")
await self.wait_for_load()
return (
f"Filled input field '{identifier}' with text '{text}' "
f"and pressed Enter."
)
def fill_input_id(
self, identifier: Union[str, int], text: str
) -> Coroutine[Any, Any, str]:
r"""Fill an input field with the given text, and then press Enter."""
return self.async_fill_input_id(identifier, text)
async def async_scroll_to_bottom(self) -> str:
r"""Asynchronously scroll to the bottom of the page."""
await self.page.evaluate(
"window.scrollTo(0, document.body.scrollHeight);"
)
await self.wait_for_load()
return "Scrolled to the bottom of the page."
def scroll_to_bottom(self) -> Coroutine[Any, Any, str]:
r"""Scroll to the bottom of the page."""
return self.async_scroll_to_bottom()
async def async_scroll_to_top(self) -> str:
r"""Asynchronously scroll to the top of the page."""
await self.page.evaluate("window.scrollTo(0, 0);")
await self.wait_for_load()
return "Scrolled to the top of the page."
def scroll_to_top(self) -> Coroutine[Any, Any, str]:
r"""Scroll to the top of the page."""
return self.async_scroll_to_top()
async def async_hover_id(self, identifier: Union[str, int]) -> str:
r"""Asynchronously hover over an element with the given identifier.
Args:
identifier (Union[str, int]): The identifier of the element
to hover over.
Returns:
str: The result of the action.
"""
if isinstance(identifier, int):
identifier = str(identifier)
try:
target = self.page.locator(f"[__elementId='{identifier}']")
except (TimeoutError, Exception) as e:
logger.debug(f"Error during hover operation: {e}")
logger.warning(
f"Element with identifier '{identifier}' not found."
)
return f"Element with identifier '{identifier}' not found."
await target.scroll_into_view_if_needed()
await target.hover()
await self.wait_for_load()
return f"Hovered over element with identifier '{identifier}'."
def hover_id(
self, identifier: Union[str, int]
) -> Coroutine[Any, Any, str]:
r"""Hover over an element with the given identifier."""
return self.async_hover_id(identifier)
async def async_find_text_on_page(self, search_text: str) -> str:
r"""Asynchronously find the next given text on the page.It is
equivalent to pressing Ctrl + F and searching for the text.
Args:
search_text (str): The text to search for.
Returns:
str: The result of the action.
"""
script = f"""
(function() {{
let text = "{search_text}";
let found = window.find(text);
if (!found) {{
let elements = document.querySelectorAll(
"*:not(script):not(style)"
);
for (let el of elements) {{
if (el.innerText && el.innerText.includes(text)) {{
el.scrollIntoView({{
behavior: "smooth",
block: "center"
}});
el.style.backgroundColor = "yellow";
el.style.border = '2px solid red';
return true;
}}
}}
return false;
}}
return true;
}})();
"""
found = await self.page.evaluate(script)
await self.wait_for_load()
if found:
return f"Found text '{search_text}' on the page."
else:
return f"Text '{search_text}' not found on the page."
def find_text_on_page(self, search_text: str) -> Coroutine[Any, Any, str]:
r"""Find the next given text on the page, and scroll the page to
the targeted text. It is equivalent to pressing Ctrl + F and
searching for the text.
Args:
search_text (str): The text to search for.
Returns:
str: The result of the action.
"""
return self.async_find_text_on_page(search_text)
async def async_back(self) -> None:
r"""Asynchronously navigate back to the previous page."""
page_url_before = self.page.url
await self.page.go_back()
page_url_after = self.page.url
if page_url_after == "about:blank":
await self.visit_page(page_url_before)
if page_url_before == page_url_after:
# If the page is not changed, try to use the history
if len(self.page_history) > 0:
await self.visit_page(self.page_history.pop())
await asyncio.sleep(1)
await self.wait_for_load()
def back(self) -> Coroutine[Any, Any, None]:
r"""Navigate back to the previous page."""
return self.async_back()
async def async_close(self) -> None:
r"""Asynchronously close the browser."""
await self.browser.close()
def close(self) -> Coroutine[Any, Any, None]:
r"""Close the browser."""
return self.async_close()
async def async_show_interactive_elements(self) -> None:
r"""Asynchronously show simple interactive elements on
the current page."""
await self.page.evaluate(self.page_script)
await self.page.evaluate("""
() => {
document.querySelectorAll(
'a, button, input, select, textarea, ' +
'[tabindex]:not([tabindex="-1"]), ' +
'[contenteditable="true"]'
).forEach(el => {
el.style.border = '2px solid red';
});
}
""")
def show_interactive_elements(self) -> Coroutine[Any, Any, None]:
r"""Show simple interactive elements on the current page."""
return self.async_show_interactive_elements()
async def async_get_webpage_content(self) -> str:
r"""Asynchronously extract the content of the current page and convert
it to markdown."""
from html2text import html2text
await self.wait_for_load()
html_content = await self.page.content()
markdown_content = html2text(html_content)
return markdown_content
@retry_on_error()
def get_webpage_content(self) -> Coroutine[Any, Any, str]:
r"""Extract the content of the current page."""
return self.async_get_webpage_content()
async def async_ensure_browser_installed(self) -> None:
r"""Ensure the browser is installed."""
import platform
import sys
try:
from playwright.async_api import async_playwright
async with async_playwright() as p:
browser = await p.chromium.launch(channel=self.channel)
await browser.close()
except Exception:
logger.info("Installing Chromium browser...")
try:
proc1 = await asyncio.create_subprocess_exec(
sys.executable,
"-m",
"playwright",
"install",
self.channel,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await proc1.communicate()
if proc1.returncode != 0:
raise RuntimeError(
f"Failed to install browser: {stderr.decode()}"
)
if platform.system().lower() == "linux":
proc2 = await asyncio.create_subprocess_exec(
sys.executable,
"-m",
"playwright",
"install-deps",
self.channel,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout2, stderr2 = await proc2.communicate()
if proc2.returncode != 0:
error_message = stderr2.decode()
raise RuntimeError(
f"Failed to install dependencies: {error_message}"
)
logger.info("Chromium browser installation completed")
except Exception as e:
raise RuntimeError(f"Installation failed: {e}")
def _ensure_browser_installed(self) -> Coroutine[Any, Any, None]:
r"""Ensure the browser is installed."""
return self.async_ensure_browser_installed()
