SurfSense/surfsense_backend/app/proprietary/web_crawler/connector.py

# SurfSense proprietary crawler engine.
#
# This module is part of the ``app.proprietary`` package and is licensed
# SEPARATELY from the Apache-2.0 project root. See ``app/proprietary/LICENSE``.
# Do not relicense or redistribute this file under Apache-2.0.
"""
WebCrawler Connector Module

A single-framework (Scrapling) web crawler with Trafilatura for HTML -> markdown
extraction. Provides a unified interface for web scraping.

Fallback ladder (the ``FetchStrategy`` seam — see ``plans/backend/03a-crawler-core.md``):
  1. Scrapling AsyncFetcher    (fast static HTTP, TLS-impersonated, no subprocess)
  2. Scrapling DynamicFetcher  (full browser, run in a thread)
  3. Scrapling StealthyFetcher (patchright-Chromium anti-bot + Cloudflare solving,
                                run in a thread)

Every tier returns extracted content via the same ``CrawlOutcome`` contract, so
callers (indexer, chat tool, crawl billing) depend only on the outcome, never on
which tier produced it.
"""

import asyncio
import logging
import time
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from enum import Enum
from typing import Any

import trafilatura
import validators
from scrapling.engines.toolbelt import is_proxy_error
from scrapling.fetchers import AsyncFetcher, DynamicFetcher, StealthyFetcher

from app.utils.proxy import get_proxy_url, is_pool_backed

logger = logging.getLogger(__name__)

# Prefix for performance/timing log lines so they are easy to grep/filter.
_PERF = "[webcrawler][perf]"


class CrawlOutcomeStatus(str, Enum):
    """Deterministic per-URL crawl result, single-sourcing the billable signal."""

    SUCCESS = "success"  # a tier returned usable extracted content
    EMPTY = "empty"  # fetched, but no usable content after ALL tiers
    FAILED = "failed"  # invalid URL or every tier errored / was unavailable


@dataclass
class CrawlOutcome:
    """Explicit ``crawl_url`` result shared by every caller.

    The **billable success predicate is single-sourced**:
    ``status == CrawlOutcomeStatus.SUCCESS`` (Phase 3c meters on it). Picking a
    dataclass over a tuple lets later subplans append fields without breaking
    callers (03d adds ``captcha_attempts``/``captcha_solved`` for per-attempt
    billing; 03e's block classifier can attach a ``block_type``).
    """

    status: CrawlOutcomeStatus
    result: dict[str, Any] | None = None
    error: str | None = None
    tier: str | None = None


class WebCrawlerConnector:
    """Class for crawling web pages and extracting content."""

    async def crawl_url(self, url: str) -> CrawlOutcome:
        """
        Crawl a single URL and extract its content.

        Fallback ladder:
          1. Scrapling AsyncFetcher (fast static HTTP, TLS-impersonated)
          2. Scrapling DynamicFetcher (full browser, run in a thread)
          3. Scrapling StealthyFetcher (anti-bot stealth browser + Cloudflare
             solving, run in a thread)

        Args:
            url: URL to crawl

        Returns:
            A ``CrawlOutcome``. On ``SUCCESS``, ``result`` is a dict containing:
                - content: Extracted content (markdown)
                - metadata: Page metadata (title, description, etc.)
                - crawler_type: Identifier of the tier that produced the content
        """
        total_start = time.perf_counter()
        try:
            if not validators.url(url):
                return CrawlOutcome(
                    status=CrawlOutcomeStatus.FAILED,
                    error=f"Invalid URL: {url}",
                )

            errors: list[str] = []
            # True once any tier fetched the page but extraction yielded nothing
            # (distinguishes EMPTY from FAILED, where every tier raised/was
            # unavailable).
            reached_without_content = False

            # --- 1. Scrapling AsyncFetcher (fast static HTTP) ---
            tier_start = time.perf_counter()
            try:
                logger.info(f"[webcrawler] Using Scrapling AsyncFetcher for: {url}")
                result = await self._run_tier_with_proxy_retry(
                    "scrapling-static", lambda: self._crawl_with_async_fetcher(url)
                )
                if result:
                    self._log_tier_outcome(
                        "scrapling-static", url, tier_start, "success"
                    )
                    self._log_total(url, "scrapling-static", total_start)
                    return CrawlOutcome(
                        status=CrawlOutcomeStatus.SUCCESS,
                        result=result,
                        tier="scrapling-static",
                    )
                reached_without_content = True
                errors.append("Scrapling static: empty extraction")
                self._log_tier_outcome("scrapling-static", url, tier_start, "empty")
            except Exception as exc:
                errors.append(f"Scrapling static: {exc!s}")
                self._log_tier_outcome(
                    "scrapling-static", url, tier_start, "error", exc
                )

            # --- 2. Scrapling DynamicFetcher (full browser) ---
            tier_start = time.perf_counter()
            try:
                logger.info(f"[webcrawler] Using Scrapling DynamicFetcher for: {url}")
                result = await self._run_tier_with_proxy_retry(
                    "scrapling-dynamic", lambda: self._crawl_with_dynamic(url)
                )
                if result:
                    self._log_tier_outcome(
                        "scrapling-dynamic", url, tier_start, "success"
                    )
                    self._log_total(url, "scrapling-dynamic", total_start)
                    return CrawlOutcome(
                        status=CrawlOutcomeStatus.SUCCESS,
                        result=result,
                        tier="scrapling-dynamic",
                    )
                reached_without_content = True
                errors.append("Scrapling dynamic: empty extraction")
                self._log_tier_outcome("scrapling-dynamic", url, tier_start, "empty")
            except NotImplementedError:
                errors.append(
                    "Scrapling dynamic: event loop does not support subprocesses "
                    "(common on Windows with uvicorn --reload)"
                )
                self._log_tier_outcome(
                    "scrapling-dynamic", url, tier_start, "unavailable"
                )
            except Exception as exc:
                errors.append(f"Scrapling dynamic: {exc!s}")
                self._log_tier_outcome(
                    "scrapling-dynamic", url, tier_start, "error", exc
                )

            # --- 3. Scrapling StealthyFetcher (anti-bot, last resort) ---
            tier_start = time.perf_counter()
            try:
                logger.info(f"[webcrawler] Using Scrapling StealthyFetcher for: {url}")
                result = await self._run_tier_with_proxy_retry(
                    "scrapling-stealthy", lambda: self._crawl_with_stealthy(url)
                )
                if result:
                    self._log_tier_outcome(
                        "scrapling-stealthy", url, tier_start, "success"
                    )
                    self._log_total(url, "scrapling-stealthy", total_start)
                    return CrawlOutcome(
                        status=CrawlOutcomeStatus.SUCCESS,
                        result=result,
                        tier="scrapling-stealthy",
                    )
                reached_without_content = True
                errors.append("Scrapling stealthy: empty extraction")
                self._log_tier_outcome("scrapling-stealthy", url, tier_start, "empty")
            except NotImplementedError:
                errors.append(
                    "Scrapling stealthy: event loop does not support subprocesses "
                    "(common on Windows with uvicorn --reload)"
                )
                self._log_tier_outcome(
                    "scrapling-stealthy", url, tier_start, "unavailable"
                )
            except Exception as exc:
                errors.append(f"Scrapling stealthy: {exc!s}")
                self._log_tier_outcome(
                    "scrapling-stealthy", url, tier_start, "error", exc
                )

            self._log_total(url, "none", total_start)
            if reached_without_content:
                return CrawlOutcome(
                    status=CrawlOutcomeStatus.EMPTY,
                    error=f"No content extracted for {url}. {'; '.join(errors)}",
                )
            return CrawlOutcome(
                status=CrawlOutcomeStatus.FAILED,
                error=f"All crawl methods failed for {url}. {'; '.join(errors)}",
            )

        except Exception as e:
            self._log_total(url, "error", total_start)
            return CrawlOutcome(
                status=CrawlOutcomeStatus.FAILED,
                error=f"Error crawling URL {url}: {e!s}",
            )

    async def _run_tier_with_proxy_retry(
        self,
        tier: str,
        attempt: Callable[[], Awaitable[dict[str, Any] | None]],
    ) -> dict[str, Any] | None:
        """Run one fetch tier, retrying once on a proxy error when pool-backed.

        ``03b`` rotation: a pool-backed ``CustomProxyProvider`` yields the *next*
        endpoint on every ``get_proxy_url()`` call, so simply re-invoking the tier
        rotates the proxy. Bounded to a single extra attempt per tier — no
        unbounded fan-out on billable crawls. Single-endpoint providers
        (including the server-side-rotating ``anonymous_proxies``) skip the retry
        entirely (``is_pool_backed()`` is ``False``), since retrying the same
        static endpoint would just re-hit the same dead proxy. Non-proxy errors
        (and ``NotImplementedError`` from the browser tiers) propagate unchanged
        for the caller's existing per-tier handling.
        """
        try:
            return await attempt()
        except Exception as exc:
            if is_proxy_error(exc) and is_pool_backed():
                logger.warning(
                    "%s tier=%s proxy error; rotating endpoint, retrying once: %s",
                    _PERF,
                    tier,
                    exc,
                )
                return await attempt()
            raise

    @staticmethod
    def _log_tier_outcome(
        tier: str,
        url: str,
        tier_start: float,
        outcome: str,
        exc: Exception | None = None,
    ) -> None:
        """Log how long a single tier took and how it ended."""
        elapsed_ms = (time.perf_counter() - tier_start) * 1000
        if outcome == "error":
            logger.warning(
                "%s tier=%s url=%s elapsed_ms=%.1f outcome=error error=%s",
                _PERF,
                tier,
                url,
                elapsed_ms,
                exc,
            )
        else:
            logger.info(
                "%s tier=%s url=%s elapsed_ms=%.1f outcome=%s",
                _PERF,
                tier,
                url,
                elapsed_ms,
                outcome,
            )

    @staticmethod
    def _log_total(url: str, selected: str, total_start: float) -> None:
        """Log the total time spent across all attempted tiers."""
        total_ms = (time.perf_counter() - total_start) * 1000
        logger.info(
            "%s url=%s selected=%s total_ms=%.1f",
            _PERF,
            url,
            selected,
            total_ms,
        )

    async def _crawl_with_async_fetcher(self, url: str) -> dict[str, Any] | None:
        """
        Crawl URL using Scrapling's AsyncFetcher (static HTTP) + Trafilatura.

        AsyncFetcher is httpx/curl_cffi based and does not launch a browser
        subprocess, making it safe to call from any asyncio event loop. Returns
        ``None`` when Trafilatura cannot extract meaningful content (e.g. JS
        rendered SPAs) so the caller can fall through to the browser tiers.
        """
        fetch_start = time.perf_counter()
        # ``impersonate="chrome"`` makes curl_cffi present a real Chrome TLS
        # ClientHello (JA3/JA4) instead of its default fingerprint, keeping the
        # static tier coherent with the browser tiers' UA (see 03e §2b).
        page = await AsyncFetcher.get(
            url,
            stealthy_headers=True,
            impersonate="chrome",
            proxy=get_proxy_url(),
            timeout=20,
        )
        fetch_ms = (time.perf_counter() - fetch_start) * 1000

        status = getattr(page, "status", None)
        if status is not None and status >= 400:
            logger.info(
                "%s tier=scrapling-static url=%s fetch_ms=%.1f status=%s outcome=http_error",
                _PERF,
                url,
                fetch_ms,
                status,
            )
            return None

        return self._build_result(
            page.html_content,
            url,
            "scrapling-static",
            allow_raw_fallback=False,
            fetch_ms=fetch_ms,
            status=status,
        )

    async def _crawl_with_dynamic(self, url: str) -> dict[str, Any] | None:
        """
        Crawl URL using Scrapling's DynamicFetcher (full browser) + Trafilatura.

        Runs the sync fetch in a worker thread so it works on any event loop,
        including Windows ``SelectorEventLoop`` which cannot spawn subprocesses.
        """
        return await asyncio.to_thread(self._crawl_with_dynamic_sync, url)

    def _crawl_with_dynamic_sync(self, url: str) -> dict[str, Any] | None:
        """Synchronous DynamicFetcher crawl executed in a worker thread."""
        fetch_start = time.perf_counter()
        page = DynamicFetcher.fetch(
            url,
            headless=True,
            network_idle=True,
            timeout=30000,
            proxy=get_proxy_url(),
        )
        fetch_ms = (time.perf_counter() - fetch_start) * 1000
        return self._build_result(
            page.html_content,
            url,
            "scrapling-dynamic",
            allow_raw_fallback=False,
            fetch_ms=fetch_ms,
            status=getattr(page, "status", None),
        )

    async def _crawl_with_stealthy(self, url: str) -> dict[str, Any] | None:
        """
        Crawl URL using Scrapling's StealthyFetcher (patchright-Chromium) + Trafilatura.

        Last-resort tier with anti-bot features. Runs the sync fetch in a worker
        thread for the same event-loop-safety reasons as DynamicFetcher. Falls
        back to the raw HTML when Trafilatura extraction is empty.
        """
        return await asyncio.to_thread(self._crawl_with_stealthy_sync, url)

    def _crawl_with_stealthy_sync(self, url: str) -> dict[str, Any] | None:
        """Synchronous StealthyFetcher crawl executed in a worker thread."""
        fetch_start = time.perf_counter()
        # ``solve_cloudflare=True`` runs the full Turnstile/Interstitial challenge
        # loop; scoped to this last-resort tier only (it spins up the browser).
        page = StealthyFetcher.fetch(
            url,
            headless=True,
            network_idle=True,
            block_ads=True,
            solve_cloudflare=True,
            proxy=get_proxy_url(),
        )
        fetch_ms = (time.perf_counter() - fetch_start) * 1000
        return self._build_result(
            page.html_content,
            url,
            "scrapling-stealthy",
            allow_raw_fallback=True,
            fetch_ms=fetch_ms,
            status=getattr(page, "status", None),
        )

    def _build_result(
        self,
        raw_html: str | None,
        url: str,
        crawler_type: str,
        *,
        allow_raw_fallback: bool,
        fetch_ms: float | None = None,
        status: int | None = None,
    ) -> dict[str, Any] | None:
        """
        Extract markdown + metadata from raw HTML using Trafilatura.

        Args:
            raw_html: Raw HTML source from a fetcher.
            url: Original URL (used as the metadata source/title fallback).
            crawler_type: Identifier of the tier that produced the HTML.
            allow_raw_fallback: When True, return the raw HTML as content if
                Trafilatura cannot extract anything (used by the last-resort
                stealthy tier). When False, return ``None`` so the caller can
                fall through to the next tier.
            fetch_ms: Time spent fetching the page (for perf logging).
            status: HTTP status code returned by the fetcher (for perf logging).

        Returns:
            Result dict (content/metadata/crawler_type) or ``None``.
        """
        html_len = len(raw_html) if raw_html else 0

        if not raw_html or len(raw_html.strip()) == 0:
            self._log_build(
                crawler_type, url, fetch_ms, 0.0, status, html_len, 0, "empty_html"
            )
            return None

        extract_start = time.perf_counter()
        extracted_content: str | None = None
        trafilatura_metadata = None

        try:
            extracted_content = trafilatura.extract(
                raw_html,
                output_format="markdown",
                include_comments=False,
                include_tables=True,
                include_images=True,
                include_links=True,
            )
            trafilatura_metadata = trafilatura.extract_metadata(raw_html)

            if extracted_content and len(extracted_content.strip()) == 0:
                extracted_content = None
        except Exception:
            extracted_content = None

        extract_ms = (time.perf_counter() - extract_start) * 1000

        if not extracted_content and not allow_raw_fallback:
            self._log_build(
                crawler_type,
                url,
                fetch_ms,
                extract_ms,
                status,
                html_len,
                0,
                "no_extraction",
            )
            return None

        metadata: dict[str, str] = {"source": url}
        if trafilatura_metadata:
            if trafilatura_metadata.title:
                metadata["title"] = trafilatura_metadata.title
            if trafilatura_metadata.description:
                metadata["description"] = trafilatura_metadata.description
            if trafilatura_metadata.author:
                metadata["author"] = trafilatura_metadata.author
            if trafilatura_metadata.date:
                metadata["date"] = trafilatura_metadata.date
        metadata.setdefault("title", url)

        content = extracted_content if extracted_content else raw_html
        self._log_build(
            crawler_type,
            url,
            fetch_ms,
            extract_ms,
            status,
            html_len,
            len(content),
            "extracted" if extracted_content else "raw_fallback",
        )

        return {
            "content": content,
            "metadata": metadata,
            "crawler_type": crawler_type,
        }

    @staticmethod
    def _log_build(
        crawler_type: str,
        url: str,
        fetch_ms: float | None,
        extract_ms: float,
        status: int | None,
        html_len: int,
        content_len: int,
        outcome: str,
    ) -> None:
        """Emit a detailed perf line splitting fetch vs Trafilatura extraction."""
        fetch_repr = f"{fetch_ms:.1f}" if fetch_ms is not None else "n/a"
        logger.info(
            "%s tier=%s url=%s status=%s fetch_ms=%s extract_ms=%.1f "
            "html_len=%d content_len=%d outcome=%s",
            _PERF,
            crawler_type,
            url,
            status,
            fetch_repr,
            extract_ms,
            html_len,
            content_len,
            outcome,
        )

    def format_to_structured_document(
        self, crawl_result: dict[str, Any], exclude_metadata: bool = False
    ) -> str:
        """
        Format crawl result as a structured document.

        Args:
            crawl_result: Result from crawl_url method
            exclude_metadata: If True, excludes ALL metadata fields from the document.
                            This is useful for content hash generation to ensure the hash
                            only changes when actual content changes, not when metadata
                            (which often contains dynamic fields like timestamps, IDs, etc.) changes.

        Returns:
            Structured document string
        """
        metadata = crawl_result["metadata"]
        content = crawl_result["content"]

        document_parts = ["<DOCUMENT>"]

        # Include metadata section only if not excluded
        if not exclude_metadata:
            document_parts.append("<METADATA>")
            for key, value in metadata.items():
                document_parts.append(f"{key.upper()}: {value}")
            document_parts.append("</METADATA>")

        document_parts.extend(
            [
                "<CONTENT>",
                "FORMAT: markdown",
                "TEXT_START",
                content,
                "TEXT_END",
                "</CONTENT>",
                "</DOCUMENT>",
            ]
        )

        return "\n".join(document_parts)