core/engine/src/tools/browser.py - main - alexefan136/flowstack

core/engine/src/tools/browser.py
769 строк25 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:1976704c6
"""
Browser Tool — MCP tool для работы с веб-страницами.

Предоставляет возможности для:
- Загрузки веб-страниц
- Извлечения текста из HTML
- Получения скриншотов страниц
- Извлечения ссылок и метаданных

Используемые библиотеки:
- aiohttp: асинхронные HTTP запросы
- beautifulsoup4: парсинг HTML
- playwright (опционально): скриншоты и JavaScript rendering

Архитектурные принципы:
- Async-first дизайн
- Чистые функции для преобразований
- MCP-совместимый ToolResult
- Proper error handling
- Rate limiting и timeouts
"""
from __future__ import annotations

import asyncio
import logging
from dataclasses import dataclass
from typing import Any, ClassVar, TYPE_CHECKING
from urllib.parse import urljoin, urlparse

import aiohttp  # type: ignore[import-not-found,import-untyped]
from bs4 import BeautifulSoup  # type: ignore[import-not-found,import-untyped]

if TYPE_CHECKING:
    from playwright.async_api import Browser as PlaywrightBrowser  # type: ignore[import-not-found,import-untyped]

from src.tools.base import (
    Content,
    ImageContent,
    TextContent,
    Tool,
    ToolResult,
)


logger = logging.getLogger(__name__)


# ============================================================================
# Configuration
# ============================================================================


@dataclass
class BrowserConfig:
    """Конфигурация для browser tool."""

    # HTTP settings
    timeout_seconds: float = 30.0
    max_redirects: int = 10
    user_agent: str = (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/120.0.0.0 Safari/537.36"
    )

    # Content extraction
    max_content_length: int = 1_000_000  # 1 MB
    extract_images: bool = True
    extract_links: bool = True

    # Rate limiting
    requests_per_second: float = 2.0

    # Screenshot settings (если установлен playwright)
    screenshot_width: int = 1280
    screenshot_height: int = 720
    screenshot_full_page: bool = False

    def validate(self) -> list[str]:
        """Валидировать конфигурацию. Чистая функция."""
        errors: list[str] = []

        if self.timeout_seconds <= 0:
            errors.append("timeout_seconds must be positive")

        if self.max_redirects < 0:
            errors.append("max_redirects must be non-negative")

        if self.max_content_length <= 0:
            errors.append("max_content_length must be positive")

        if self.requests_per_second <= 0:
            errors.append("requests_per_second must be positive")

        return errors


# ============================================================================
# Response Models
# ============================================================================


@dataclass
class PageInfo:
    """Информация о загруженной странице."""

    url: str
    title: str = ""
    status_code: int = 200
    content_type: str = ""
    content_length: int = 0
    final_url: str = ""  # После редиректов
    load_time_ms: float = 0.0

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        return {
            "url": self.url,
            "title": self.title,
            "status_code": self.status_code,
            "content_type": self.content_type,
            "content_length": self.content_length,
            "final_url": self.final_url,
            "load_time_ms": self.load_time_ms,
        }


@dataclass
class ExtractedLink:
    """Извлечённая ссылка."""

    href: str
    text: str = ""
    title: str = ""
    is_external: bool = False

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        return {
            "href": self.href,
            "text": self.text,
            "title": self.title,
            "is_external": self.is_external,
        }


@dataclass
class ExtractedImage:
    """Извлечённое изображение."""

    src: str
    alt: str = ""
    title: str = ""
    width: int | None = None
    height: int | None = None

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        return {
            "src": self.src,
            "alt": self.alt,
            "title": self.title,
            "width": self.width,
            "height": self.height,
        }


# ============================================================================
# Helper Functions
# ============================================================================


def _normalize_url(url: str) -> str:
    """
    Нормализовать URL.

    Чистая функция — добавляет scheme если отсутствует.
    """
    if not url.startswith(("http://", "https://")):
        return f"https://{url}"
    return url


def _is_external_url(base_url: str, href: str) -> bool:
    """
    Проверить, является ли ссылка внешней.

    Чистая функция.
    """
    try:
        base_domain = urlparse(base_url).netloc
        href_domain = urlparse(urljoin(base_url, href)).netloc
        return base_domain != href_domain
    except Exception:
        return True


def _extract_title(soup: BeautifulSoup) -> str:
    """
    Извлечь заголовок страницы.

    Чистая функция.
    """
    title_tag = soup.find("title")
    if title_tag and title_tag.string:
        return str(title_tag.string).strip()

    # Fallback: og:title или первый h1
    og_title = soup.find("meta", property="og:title")
    if og_title and og_title.get("content"):
        return str(og_title["content"]).strip()

    h1 = soup.find("h1")
    if h1 and h1.get_text():
        return h1.get_text().strip()

    return ""


def _extract_main_content(soup: BeautifulSoup) -> str:
    """
    Извлечь основной текст страницы.

    Чистая функция — удаляет скрипты, стили, navigation.
    """
    # Удаляем ненужные элементы
    for element in soup.find_all(["script", "style", "nav", "header", "footer", "aside"]):
        element.decompose()

    # Пробуем найти main content
    main_content = (
        soup.find("main")
        or soup.find("article")
        or soup.find(attrs={"role": "main"})
        or soup.find("div", class_="content")
        or soup.body
    )

    if not main_content:
        return soup.get_text(separator="\n", strip=True)

    return main_content.get_text(separator="\n", strip=True)


def _extract_links(
    soup: BeautifulSoup, base_url: str
) -> list[ExtractedLink]:
    """
    Извлечь все ссылки со страницы.

    Чистая функция.
    """
    links: list[ExtractedLink] = []
    seen_hrefs: set[str] = set()

    for a_tag in soup.find_all("a", href=True):
        href = str(a_tag["href"])

        # Пропускаем якоря и javascript
        if href.startswith(("#", "javascript:", "mailto:", "tel:")):
            continue

        # Абсолютный URL
        absolute_url = urljoin(base_url, href)

        # Дедупликация
        if absolute_url in seen_hrefs:
            continue
        seen_hrefs.add(absolute_url)

        links.append(
            ExtractedLink(
                href=absolute_url,
                text=a_tag.get_text(strip=True),
                title=str(a_tag.get("title", "")),
                is_external=_is_external_url(base_url, href),
            )
        )

    return links


def _extract_images(
    soup: BeautifulSoup, base_url: str
) -> list[ExtractedImage]:
    """
    Извлечь все изображения со страницы.

    Чистая функция.
    """
    images: list[ExtractedImage] = []
    seen_srcs: set[str] = set()

    for img_tag in soup.find_all("img", src=True):
        src = str(img_tag["src"])

        # Пропускаем data URI
        if src.startswith("data:"):
            continue

        # Абсолютный URL
        absolute_src = urljoin(base_url, src)

        # Дедупликация
        if absolute_src in seen_srcs:
            continue
        seen_srcs.add(absolute_src)

        width = img_tag.get("width")
        height = img_tag.get("height")

        images.append(
            ExtractedImage(
                src=absolute_src,
                alt=str(img_tag.get("alt", "")),
                title=str(img_tag.get("title", "")),
                width=int(width) if width and str(width).isdigit() else None,
                height=int(height) if height and str(height).isdigit() else None,
            )
        )

    return images


# ============================================================================
# Browser Client
# ============================================================================


class BrowserClient:
    """
    Асинхронный клиент для работы с веб-страницами.

    Автоматически обрабатывает:
    - HTTP запросы с timeouts
    - Redirects
    - HTML парсинг
    - Rate limiting
    """

    def __init__(self, config: BrowserConfig | None = None):
        self.config = config or BrowserConfig()
        self._session: aiohttp.ClientSession | None = None
        self._rate_limiter = asyncio.Semaphore(int(self.config.requests_per_second))

    async def __aenter__(self) -> BrowserClient:
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        await self.close()

    async def connect(self) -> None:
        """Создать HTTP сессию."""
        if self._session is not None:
            return

        timeout = aiohttp.ClientTimeout(total=self.config.timeout_seconds)
        self._session = aiohttp.ClientSession(
            timeout=timeout,
            headers={"User-Agent": self.config.user_agent},
        )

    async def close(self) -> None:
        """Закрыть сессию."""
        if self._session is not None:
            await self._session.close()
            self._session = None

    def _require_session(self) -> aiohttp.ClientSession:
        """Получить активную сессию или выбросить ошибку."""
        if self._session is None:
            raise RuntimeError("Browser client is not connected")
        return self._session

    async def fetch_page(self, url: str) -> tuple[str, PageInfo]:
        """
        Загрузить HTML страницу.

        Args:
            url: URL страницы

        Returns:
            Tuple из (HTML content, PageInfo)

        Raises:
            aiohttp.ClientError: при HTTP ошибке
        """
        session = self._require_session()
        normalized_url = _normalize_url(url)

        start_time = asyncio.get_event_loop().time()

        async with self._rate_limiter:
            async with session.get(
                normalized_url,
                max_redirects=self.config.max_redirects,
            ) as response:
                response.raise_for_status()

                # Читаем контент с лимитом
                content_bytes = await response.read()
                if len(content_bytes) > self.config.max_content_length:
                    content_bytes = content_bytes[: self.config.max_content_length]

                content = content_bytes.decode("utf-8", errors="replace")

                load_time_ms = (asyncio.get_event_loop().time() - start_time) * 1000

                # Парсим для извлечения title
                soup = BeautifulSoup(content, "html.parser")
                title = _extract_title(soup)

                page_info = PageInfo(
                    url=normalized_url,
                    title=title,
                    status_code=response.status,
                    content_type=response.content_type or "",
                    content_length=len(content_bytes),
                    final_url=str(response.url),
                    load_time_ms=load_time_ms,
                )

                return content, page_info

    async def take_screenshot(self, url: str) -> bytes | None:
        """
        Сделать скриншот страницы.

        Требует установленный playwright.

        Args:
            url: URL страницы

        Returns:
            PNG изображение в bytes или None если playwright не установлен
        """
        try:
            from playwright.async_api import async_playwright # type: ignore[import-not-found,import-untyped]
        except ImportError:
            logger.warning(
                "playwright is not installed. "
                "Install with: pip install playwright && playwright install"
            )
            return None

        normalized_url = _normalize_url(url)

        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            page = await browser.new_page(
                viewport={
                    "width": self.config.screenshot_width,
                    "height": self.config.screenshot_height,
                }
            )

            try:
                await page.goto(normalized_url, wait_until="networkidle")
                screenshot_bytes = await page.screenshot(
                    full_page=self.config.screenshot_full_page,
                    type="png",
                )
                return screenshot_bytes
            finally:
                await browser.close()


# ============================================================================
# Browser Tool
# ============================================================================


class BrowserTool(Tool):
    """
    MCP Tool для работы с веб-страницами.

    Поддерживает операции:
    - fetch: загрузить страницу и извлечь текст
    - screenshot: сделать скриншот страницы
    - links: извлечь все ссылки
    - images: извлечь все изображения
    - metadata: извлечь метаданные (title, description, og:tags)

    Example:
        tool = BrowserTool()
        result = await tool.execute(
            operation="fetch",
            url="https://example.com",
        )
    """

    name: ClassVar[str] = "browser"
    description: ClassVar[str] = (
        "Работа с веб-страницами: загрузка, извлечение текста, "
        "скриншоты, ссылки, изображения"
    )

    input_schema: ClassVar[dict[str, Any] | None] = {
        "type": "object",
        "properties": {
            "operation": {
                "type": "string",
                "enum": ["fetch", "screenshot", "links", "images", "metadata"],
                "description": (
                    "Операция: fetch (загрузить текст), screenshot (скриншот), "
                    "links (ссылки), images (изображения), metadata (метаданные)"
                ),
            },
            "url": {
                "type": "string",
                "description": "URL веб-страницы",
            },
            "extract_content": {
                "type": "boolean",
                "default": True,
                "description": "Извлекать ли основной текст (для operation=fetch)",
            },
            "max_links": {
                "type": "integer",
                "default": 50,
                "description": "Максимум ссылок для извлечения",
            },
            "max_images": {
                "type": "integer",
                "default": 20,
                "description": "Максимум изображений для извлечения",
            },
        },
        "required": ["operation", "url"],
    }

    tags: ClassVar[list[str]] = ["web", "browser", "scraping"]
    is_read_only: ClassVar[bool] = True

    def __init__(self, config: BrowserConfig | None = None):
        self.config = config or BrowserConfig()
        self._client: BrowserClient | None = None

    async def _get_client(self) -> BrowserClient:
        """Получить или создать client."""
        if self._client is None:
            self._client = BrowserClient(self.config)
            await self._client.connect()
        return self._client

    async def execute(self, **kwargs: Any) -> ToolResult:
        """
        Выполнить browser операцию.

        Args:
            operation: Тип операции (fetch, screenshot, links, images, metadata)
            url: URL страницы
            extract_content: Извлекать ли текст (для fetch)
            max_links: Максимум ссылок
            max_images: Максимум изображений

        Returns:
            ToolResult с контентом
        """
        operation = kwargs.get("operation", "")
        url = kwargs.get("url", "")

        if not operation:
            return ToolResult.failure("Missing required parameter: operation")

        if not url:
            return ToolResult.failure("Missing required parameter: url")

        try:
            client = await self._get_client()

            if operation == "fetch":
                return await self._handle_fetch(client, url, kwargs)
            elif operation == "screenshot":
                return await self._handle_screenshot(client, url)
            elif operation == "links":
                return await self._handle_links(client, url, kwargs)
            elif operation == "images":
                return await self._handle_images(client, url, kwargs)
            elif operation == "metadata":
                return await self._handle_metadata(client, url)
            else:
                return ToolResult.failure(f"Unknown operation: {operation}")

        except aiohttp.ClientError as e:
            logger.error(f"HTTP error fetching {url}: {e}")
            return ToolResult.failure(f"HTTP error: {e}")
        except Exception as e:
            logger.exception(f"Unexpected error in browser tool: {e}")
            return ToolResult.failure(f"Unexpected error: {e}")

    async def _handle_fetch(
        self, client: BrowserClient, url: str, kwargs: dict[str, Any]
    ) -> ToolResult:
        """Обработать операцию fetch."""
        extract_content = kwargs.get("extract_content", True)

        html, page_info = await client.fetch_page(url)
        soup = BeautifulSoup(html, "html.parser")

        content_items: list[Content] = []

        # Добавляем информацию о странице
        info_text = (
            f"# {page_info.title}\n\n"
            f"**URL:** {page_info.final_url}\n"
            f"**Status:** {page_info.status_code}\n"
            f"**Load time:** {page_info.load_time_ms:.0f}ms\n\n"
        )
        content_items.append(TextContent(text=info_text))

        # Извлекаем основной текст
        if extract_content:
            main_text = _extract_main_content(soup)
            if main_text:
                content_items.append(TextContent(text=main_text))

        return ToolResult.success_result(content_items, metadata=page_info.to_dict())

    async def _handle_screenshot(
        self, client: BrowserClient, url: str
    ) -> ToolResult:
        """Обработать операцию screenshot."""
        screenshot_bytes = await client.take_screenshot(url)

        if screenshot_bytes is None:
            return ToolResult.failure(
                "Screenshot failed. Install playwright: "
                "pip install playwright && playwright install"
            )

        image_content = ImageContent.from_bytes(screenshot_bytes, "image/png")

        return ToolResult.success_result(
            [
                TextContent(text=f"Screenshot of {url}"),
                image_content,
            ],
            metadata={"url": url, "size_bytes": len(screenshot_bytes)},
        )

    async def _handle_links(
        self, client: BrowserClient, url: str, kwargs: dict[str, Any]
    ) -> ToolResult:
        """Обработать операцию links."""
        max_links = int(kwargs.get("max_links", 50))

        html, page_info = await client.fetch_page(url)
        soup = BeautifulSoup(html, "html.parser")

        all_links = _extract_links(soup, page_info.final_url)
        links = all_links[:max_links]

        links_text = f"# Links from {page_info.title}\n\n"
        links_text += f"**Total links:** {len(all_links)}\n"
        if len(all_links) > max_links:
            links_text += f"**Showing:** {max_links}\n\n"

        for i, link in enumerate(links, 1):
            links_text += f"{i}. [{link.text or link.href}]({link.href})\n"
            if link.is_external:
                links_text += "   *(external)*\n"

        return ToolResult.success_result(
            [TextContent(text=links_text)],
            metadata={
                "total_links": len(all_links),
                "shown_links": len(links),
                "links": [l.to_dict() for l in links],
            },
        )

    async def _handle_images(
        self, client: BrowserClient, url: str, kwargs: dict[str, Any]
    ) -> ToolResult:
        """Обработать операцию images."""
        max_images = int(kwargs.get("max_images", 20))

        html, page_info = await client.fetch_page(url)
        soup = BeautifulSoup(html, "html.parser")

        all_images = _extract_images(soup, page_info.final_url)
        images = all_images[:max_images]

        images_text = f"# Images from {page_info.title}\n\n"
        images_text += f"**Total images:** {len(all_images)}\n"
        if len(all_images) > max_images:
            images_text += f"**Showing:** {max_images}\n\n"

        for i, img in enumerate(images, 1):
            images_text += f"{i}. {img.src}\n"
            if img.alt:
                images_text += f"   Alt: {img.alt}\n"
            if img.width and img.height:
                images_text += f"   Size: {img.width}x{img.height}\n"

        return ToolResult.success_result(
            [TextContent(text=images_text)],
            metadata={
                "total_images": len(all_images),
                "shown_images": len(images),
                "images": [img.to_dict() for img in images],
            },
        )

    async def _handle_metadata(
        self, client: BrowserClient, url: str
    ) -> ToolResult:
        """Обработать операцию metadata."""
        html, page_info = await client.fetch_page(url)
        soup = BeautifulSoup(html, "html.parser")

        metadata: dict[str, Any] = {
            "title": page_info.title,
            "url": page_info.final_url,
            "status_code": page_info.status_code,
            "content_type": page_info.content_type,
        }

        # Извлекаем meta tags
        for meta in soup.find_all("meta"):
            name = meta.get("name") or meta.get("property")
            content = meta.get("content")
            if name and content:
                metadata[str(name)] = str(content)

        # Извлекаем link tags (canonical, alternate, etc.)
        for link in soup.find_all("link", rel=True):
            rel = link.get("rel")
            href = link.get("href")
            if rel and href:
                rel_str = " ".join(rel) if isinstance(rel, list) else str(rel)
                metadata[f"link:{rel_str}"] = str(href)

        metadata_text = f"# Metadata for {page_info.title}\n\n"
        for key, value in metadata.items():
            metadata_text += f"**{key}:** {value}\n"

        return ToolResult.success_result(
            [TextContent(text=metadata_text)],
            metadata=metadata,
        )

    async def on_after_execute(
        self, result: ToolResult, **kwargs: Any
    ) -> None:
        """Hook после выполнения — логируем использование."""
        url = kwargs.get("url", "")
        operation = kwargs.get("operation", "")
        if result.is_success():
            logger.info(f"Browser tool: {operation} {url} - success")
        else:
            logger.warning(f"Browser tool: {operation} {url} - failed")


# ============================================================================
# Exports
# ============================================================================


__all__ = [
    "BrowserConfig",
    "BrowserClient",
    "BrowserTool",
    "PageInfo",
    "ExtractedLink",
    "ExtractedImage",
]