scrapegraphai--scrapegraph-ai/scrapegraphai/docloaders/plasmate.py

"""
PlasmateLoader — lightweight page fetcher using Plasmate (https://github.com/plasmate-labs/plasmate).

Plasmate is an open-source Rust browser engine that outputs a Structured Object Model (SOM)
instead of raw HTML. It requires no Chrome process, uses ~64MB RAM per session vs ~300MB,
and delivers 10-100x fewer tokens per page — lowering LLM costs for AI-powered scraping.

Install: pip install plasmate
Docs:    https://plasmate.app
"""

import asyncio
import subprocess
import shutil
from typing import AsyncIterator, Iterator, List, Optional

from langchain_community.document_loaders.base import BaseLoader
from langchain_core.documents import Document

from ..utils import get_logger

logger = get_logger("plasmate-loader")

_INSTALL_MSG = (
    "plasmate is required for PlasmateLoader. "
    "Install it with: pip install plasmate\n"
    "Docs: https://plasmate.app"
)


def _check_plasmate() -> str:
    """Return the path to the plasmate binary, or raise ImportError."""
    path = shutil.which("plasmate")
    if path is None:
        # Also check the Python-installed entry point location
        try:
            import plasmate as _p  # noqa: F401
            path = shutil.which("plasmate")
        except ImportError:
            pass
    if path is None:
        raise ImportError(_INSTALL_MSG)
    return path


class PlasmateLoader(BaseLoader):
    """Fetches pages using Plasmate — a lightweight Rust browser engine that outputs
    Structured Object Model (SOM) instead of raw HTML.

    Advantages over ChromiumLoader for static / server-rendered pages:
    - No Chrome/Playwright required — single binary, installs via pip
    - ~64MB RAM per session vs ~300MB for Chromium
    - 10-100x fewer tokens per page (SOM strips nav, ads, boilerplate)
    - Drops into existing ScrapeGraphAI workflows with minimal config changes

    For SPAs or pages that require JavaScript rendering, set ``fallback_to_chrome=True``
    to automatically retry with ChromiumLoader on empty or error responses.

    Attributes:
        urls: List of URLs to fetch.
        output_format: Plasmate output format — ``"text"`` (default, most compatible),
            ``"som"`` (full JSON), or ``"markdown"``.
        timeout: Per-request timeout in seconds. Defaults to 30.
        selector: Optional ARIA role or CSS id selector to scope extraction
            (e.g. ``"main"`` or ``"#content"``).
        extra_headers: Optional dict of HTTP headers to pass to each request.
        fallback_to_chrome: If True, retry with ChromiumLoader when Plasmate
            returns empty content (useful for JS-heavy SPAs). Defaults to False.
        chrome_kwargs: Extra kwargs forwarded to ChromiumLoader when fallback is used.

    Example::

        from scrapegraphai.docloaders import PlasmateLoader

        loader = PlasmateLoader(
            urls=["https://docs.python.org/3/library/json.html"],
            output_format="text",
            timeout=30,
        )
        docs = loader.load()
        print(docs[0].page_content[:500])
    """

    def __init__(
        self,
        urls: List[str],
        *,
        output_format: str = "text",
        timeout: int = 30,
        selector: Optional[str] = None,
        extra_headers: Optional[dict] = None,
        fallback_to_chrome: bool = False,
        **chrome_kwargs,
    ):
        if output_format not in ("som", "text", "markdown", "links"):
            raise ValueError(
                f"output_format must be one of 'som', 'text', 'markdown', 'links'; got {output_format!r}"
            )
        self.urls = urls
        self.output_format = output_format
        self.timeout = timeout
        self.selector = selector
        self.extra_headers = extra_headers or {}
        self.fallback_to_chrome = fallback_to_chrome
        self.chrome_kwargs = chrome_kwargs

    def _build_cmd(self, url: str) -> List[str]:
        """Build the plasmate CLI command for a given URL."""
        cmd = [
            "plasmate", "fetch", url,
            "--format", self.output_format,
            "--timeout", str(self.timeout * 1000),  # plasmate uses milliseconds
        ]
        if self.selector:
            cmd += ["--selector", self.selector]
        for key, value in self.extra_headers.items():
            cmd += ["--header", f"{key}: {value}"]
        return cmd

    def _fetch_url(self, url: str) -> str:
        """Synchronously fetch a URL via the plasmate binary."""
        binary = _check_plasmate()
        cmd = self._build_cmd(url)
        cmd[0] = binary  # use resolved path

        logger.info(f"[PlasmateLoader] Fetching: {url}")
        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=self.timeout + 5,  # outer kill timeout slightly above plasmate's
            )
            if result.returncode != 0:
                logger.warning(
                    f"[PlasmateLoader] plasmate exited {result.returncode} for {url}: {result.stderr[:200]}"
                )
                return ""
            content = result.stdout.strip()
            logger.info(f"[PlasmateLoader] Got {len(content)} chars from {url}")
            return content
        except subprocess.TimeoutExpired:
            logger.warning(f"[PlasmateLoader] Timeout fetching {url}")
            return ""
        except FileNotFoundError:
            raise ImportError(_INSTALL_MSG)

    def _fallback_fetch(self, url: str) -> str:
        """Fall back to ChromiumLoader when Plasmate returns empty content."""
        from .chromium import ChromiumLoader

        logger.info(f"[PlasmateLoader] Falling back to ChromiumLoader for: {url}")
        loader = ChromiumLoader([url], **self.chrome_kwargs)
        docs = loader.load()
        return docs[0].page_content if docs else ""

    def lazy_load(self) -> Iterator[Document]:
        """Yield Documents one at a time, fetching each URL synchronously."""
        for url in self.urls:
            content = self._fetch_url(url)

            if not content.strip() and self.fallback_to_chrome:
                content = self._fallback_fetch(url)

            if not content.strip():
                logger.warning(f"[PlasmateLoader] Empty content for {url} — skipping")
                continue

            yield Document(
                page_content=content,
                metadata={
                    "source": url,
                    "loader": "plasmate",
                    "format": self.output_format,
                },
            )

    async def _async_fetch_url(self, url: str) -> str:
        """Asynchronously fetch a URL by running the plasmate binary in a thread pool."""
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, self._fetch_url, url)

    async def alazy_load(self) -> AsyncIterator[Document]:
        """Asynchronously yield Documents, fetching all URLs concurrently."""
        tasks = [self._async_fetch_url(url) for url in self.urls]
        results = await asyncio.gather(*tasks)

        for url, content in zip(self.urls, results):
            if not content.strip() and self.fallback_to_chrome:
                content = self._fallback_fetch(url)

            if not content.strip():
                logger.warning(f"[PlasmateLoader] Empty content for {url} — skipping")
                continue

            yield Document(
                page_content=content,
                metadata={
                    "source": url,
                    "loader": "plasmate",
                    "format": self.output_format,
                },
            )