Merge pull request #1062 from dbhurley/feat/plasmate-loader

feat: add PlasmateLoader as lightweight scraping backend (no Chrome needed)

Author: VinciGit00

scrapegraphai/docloaders/__init__.py

@@ -4,10 +4,12 @@
 
 from .browser_base import browser_base_fetch
 from .chromium import ChromiumLoader
+from .plasmate import PlasmateLoader
 from .scrape_do import scrape_do_fetch
 
 __all__ = [
     "browser_base_fetch",
     "ChromiumLoader",
+    "PlasmateLoader",
     "scrape_do_fetch",
 ]

scrapegraphai/docloaders/plasmate.py

@@ -0,0 +1,203 @@
+"""
+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,
+                },
+            )

scrapegraphai/nodes/fetch_node.py

@@ -83,6 +83,10 @@ def __init__(
             None if node_config is None else node_config.get("scrape_do", None)
         )
 
+        self.plasmate = (
+            None if node_config is None else node_config.get("plasmate", None)
+        )
+
         self.storage_state = (
             None if node_config is None else node_config.get("storage_state", None)
         )
@@ -351,6 +355,19 @@ def handle_web_source(self, state, source):
                     )
 
                 document = [Document(page_content=data, metadata={"source": source})]
+            elif self.plasmate is not None:
+                from ..docloaders.plasmate import PlasmateLoader
+
+                plasmate_cfg = self.plasmate if isinstance(self.plasmate, dict) else {}
+                loader = PlasmateLoader(
+                    [source],
+                    output_format=plasmate_cfg.get("output_format", "text"),
+                    timeout=plasmate_cfg.get("timeout", self.timeout or 30),
+                    selector=plasmate_cfg.get("selector"),
+                    extra_headers=plasmate_cfg.get("extra_headers", {}),
+                    fallback_to_chrome=plasmate_cfg.get("fallback_to_chrome", False),
+                )
+                document = loader.load()
             else:
                 loader = ChromiumLoader(
                     [source],