# Copyright (c) 2023 - 2025, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors # # SPDX-License-Identifier: Apache-2.0 # # Portions derived from https://github.com/microsoft/autogen are under the MIT License. # SPDX-License-Identifier: MIT import io import mimetypes import os import re import uuid from contextlib import suppress from typing import Any, Optional, Union from urllib.parse import urljoin, urlparse from .import_utils import optional_import_block, require_optional_import with optional_import_block(): import markdownify import requests from bs4 import BeautifulSoup # Optional PDF support with optional_import_block() as result: import pdfminer import pdfminer.high_level IS_PDF_CAPABLE = result.is_successful # Other optional dependencies with optional_import_block(): import pathvalidate @require_optional_import(["markdownify", "requests", "bs4", "pdfminer", "pathvalidate"], "websurfer") class SimpleTextBrowser: """(In preview) An extremely simple text-based web browser comparable to Lynx. Suitable for Agentic use.""" def __init__( self, start_page: Optional[str] = None, viewport_size: Optional[int] = 1024 * 8, downloads_folder: Optional[Union[str, None]] = None, bing_base_url: str = "https://api.bing.microsoft.com/v7.0/search", bing_api_key: Optional[Union[str, None]] = None, request_kwargs: Optional[Union[dict[str, Any], None]] = None, ): """Initialize the browser with the given parameters. Args: start_page (Optional[str], optional): The initial page to load. Defaults to None. viewport_size (Optional[int], optional): The number of characters to display per page. Defaults to 1024 * 8. downloads_folder (Optional[Union[str, None]], optional): The folder to save downloads to. Defaults to None. bing_base_url (str, optional): The base URL for Bing searches. Defaults to "https://api.bing.microsoft.com/v7.0/search". bing_api_key (Optional[Union[str, None]], optional): The API key for Bing searches. Defaults to None. request_kwargs (Optional[Union[dict[str, Any], None]], optional): Additional keyword arguments to pass to the requests library. Defaults to None. """ self.start_page: str = start_page if start_page else "about:blank" self.viewport_size = viewport_size # Applies only to the standard uri types self.downloads_folder = downloads_folder self.history: list[str] = list() self.page_title: Optional[str] = None self.viewport_current_page = 0 self.viewport_pages: list[tuple[int, int]] = list() self.set_address(self.start_page) self.bing_base_url = bing_base_url self.bing_api_key = bing_api_key self.request_kwargs = request_kwargs self._page_content = "" @property def address(self) -> str: """Return the address of the current page.""" return self.history[-1] def set_address(self, uri_or_path: str) -> None: """Set the address of the current page. Args: uri_or_path (str): The URI or path to set as the current page. """ self.history.append(uri_or_path) # Handle special URIs if uri_or_path == "about:blank": self._set_page_content("") elif uri_or_path.startswith("bing:"): self._bing_search(uri_or_path[len("bing:") :].strip()) else: if not uri_or_path.startswith("http:") and not uri_or_path.startswith("https:"): uri_or_path = urljoin(self.address, uri_or_path) self.history[-1] = uri_or_path # Update the address with the fully-qualified path self._fetch_page(uri_or_path) self.viewport_current_page = 0 @property def viewport(self) -> str: """Return the content of the current viewport.""" bounds = self.viewport_pages[self.viewport_current_page] return self.page_content[bounds[0] : bounds[1]] @property def page_content(self) -> str: """Return the full contents of the current page.""" return self._page_content def _set_page_content(self, content: str) -> None: """Sets the text content of the current page.""" self._page_content = content self._split_pages() if self.viewport_current_page >= len(self.viewport_pages): self.viewport_current_page = len(self.viewport_pages) - 1 def page_down(self) -> None: """Move the viewport down by one page.""" self.viewport_current_page = min(self.viewport_current_page + 1, len(self.viewport_pages) - 1) def page_up(self) -> None: """Move the viewport up by one page.""" self.viewport_current_page = max(self.viewport_current_page - 1, 0) def visit_page(self, path_or_uri: str) -> str: """Update the address, visit the page, and return the content of the viewport. Args: path_or_uri (str): The URI or path to visit. """ self.set_address(path_or_uri) return self.viewport def _split_pages(self) -> None: # Split only regular pages if not self.address.startswith("http:") and not self.address.startswith("https:"): self.viewport_pages = [(0, len(self._page_content))] return # Handle empty pages if len(self._page_content) == 0: self.viewport_pages = [(0, 0)] return # Break the viewport into pages self.viewport_pages = [] start_idx = 0 while start_idx < len(self._page_content): end_idx = min(start_idx + self.viewport_size, len(self._page_content)) # type: ignore[operator] # Adjust to end on a space while end_idx < len(self._page_content) and self._page_content[end_idx - 1] not in [" ", "\t", "\r", "\n"]: end_idx += 1 self.viewport_pages.append((start_idx, end_idx)) start_idx = end_idx def _bing_api_call(self, query: str) -> dict[str, dict[str, list[dict[str, Union[str, dict[str, str]]]]]]: # Make sure the key was set if self.bing_api_key is None: raise ValueError("Missing Bing API key.") # Prepare the request parameters request_kwargs = self.request_kwargs.copy() if self.request_kwargs is not None else {} if "headers" not in request_kwargs: request_kwargs["headers"] = {} request_kwargs["headers"]["Ocp-Apim-Subscription-Key"] = self.bing_api_key if "params" not in request_kwargs: request_kwargs["params"] = {} request_kwargs["params"]["q"] = query request_kwargs["params"]["textDecorations"] = False request_kwargs["params"]["textFormat"] = "raw" request_kwargs["stream"] = False # Make the request response = requests.get(self.bing_base_url, **request_kwargs) response.raise_for_status() results = response.json() return results # type: ignore[no-any-return] def _bing_search(self, query: str) -> None: results = self._bing_api_call(query) web_snippets: list[str] = list() idx = 0 for page in results["webPages"]["value"]: idx += 1 web_snippets.append(f"{idx}. [{page['name']}]({page['url']})\n{page['snippet']}") if "deepLinks" in page: for dl in page["deepLinks"]: idx += 1 web_snippets.append( f"{idx}. [{dl['name']}]({dl['url']})\n{dl.get('snippet', '')}" # type: ignore[index] ) news_snippets = list() if "news" in results: for page in results["news"]["value"]: idx += 1 news_snippets.append(f"{idx}. [{page['name']}]({page['url']})\n{page['description']}") self.page_title = f"{query} - Search" content = ( f"A Bing search for '{query}' found {len(web_snippets) + len(news_snippets)} results:\n\n## Web Results\n" + "\n\n".join(web_snippets) ) if len(news_snippets) > 0: content += "\n\n## News Results:\n" + "\n\n".join(news_snippets) self._set_page_content(content) def _fetch_page(self, url: str) -> None: try: # Prepare the request parameters request_kwargs = self.request_kwargs.copy() if self.request_kwargs is not None else {} request_kwargs["stream"] = True # Send a HTTP request to the URL response = requests.get(url, **request_kwargs) response.raise_for_status() # If the HTTP request returns a status code 200, proceed if response.status_code == 200: content_type = response.headers.get("content-type", "") for ct in ["text/html", "text/plain", "application/pdf"]: if ct in content_type.lower(): content_type = ct break if content_type == "text/html": # Get the content of the response html = "" for chunk in response.iter_content(chunk_size=512, decode_unicode=True): html += chunk soup = BeautifulSoup(html, "html.parser") # Remove javascript and style blocks for script in soup(["script", "style"]): script.extract() # Convert to markdown -- Wikipedia gets special attention to get a clean version of the page if url.startswith("https://en.wikipedia.org/"): body_elm = soup.find("div", {"id": "mw-content-text"}) title_elm = soup.find("span", {"class": "mw-page-title-main"}) if body_elm: # What's the title main_title = soup.title.string if title_elm and len(title_elm) > 0: main_title = title_elm.string webpage_text = ( "# " + main_title + "\n\n" + markdownify.MarkdownConverter().convert_soup(body_elm) ) else: webpage_text = markdownify.MarkdownConverter().convert_soup(soup) else: webpage_text = markdownify.MarkdownConverter().convert_soup(soup) # Convert newlines webpage_text = re.sub(r"\r\n", "\n", webpage_text) # Remove excessive blank lines self.page_title = soup.title.string self._set_page_content(re.sub(r"\n{2,}", "\n\n", webpage_text).strip()) elif content_type == "text/plain": # Get the content of the response plain_text = "" for chunk in response.iter_content(chunk_size=512, decode_unicode=True): plain_text += chunk self.page_title = None self._set_page_content(plain_text) elif IS_PDF_CAPABLE and content_type == "application/pdf": pdf_data = io.BytesIO(response.raw.read()) self.page_title = None self._set_page_content(pdfminer.high_level.extract_text(pdf_data)) elif self.downloads_folder is not None: # Try producing a safe filename fname = None with suppress(NameError): fname = pathvalidate.sanitize_filename(os.path.basename(urlparse(url).path)).strip() # No suitable name, so make one if fname is None: extension = mimetypes.guess_extension(content_type) if extension is None: extension = ".download" fname = str(uuid.uuid4()) + extension # Open a file for writing download_path = os.path.abspath(os.path.join(self.downloads_folder, fname)) with open(download_path, "wb") as fh: for chunk in response.iter_content(chunk_size=512): fh.write(chunk) # Return a page describing what just happened self.page_title = "Download complete." self._set_page_content(f"Downloaded '{url}' to '{download_path}'.") else: self.page_title = f"Error - Unsupported Content-Type '{content_type}'" self._set_page_content(self.page_title) else: self.page_title = "Error" self._set_page_content("Failed to retrieve " + url) except requests.exceptions.RequestException as e: self.page_title = "Error" self._set_page_content(str(e))