feat(web-crawler): add site_crawler spider + url_policy over the fetch engine

surfsense_backend/app/proprietary/web_crawler/__init__.py

@@ -9,5 +9,12 @@ from app.proprietary.web_crawler.connector import (
     CrawlOutcomeStatus,
     WebCrawlerConnector,
 )
+from app.proprietary.web_crawler.site_crawler import CrawlPage, crawl_site
 
-__all__ = ["CrawlOutcome", "CrawlOutcomeStatus", "WebCrawlerConnector"]
+__all__ = [
+    "CrawlOutcome",
+    "CrawlOutcomeStatus",
+    "CrawlPage",
+    "WebCrawlerConnector",
+    "crawl_site",
+]

surfsense_backend/app/proprietary/web_crawler/connector.py

@@ -34,6 +34,7 @@ from scrapling.engines.toolbelt import is_proxy_error
 from scrapling.fetchers import AsyncFetcher, DynamicFetcher, StealthyFetcher
 
 from app.proprietary.web_crawler.captcha import build_captcha_page_action
+from app.proprietary.web_crawler.url_policy import extract_links
 from app.proprietary.web_crawler.stealth import (
     build_stealthy_kwargs,
     get_stealth_config,
@@ -601,6 +602,9 @@ class WebCrawlerConnector:
             "content": content,
             "metadata": metadata,
             "crawler_type": crawler_type,
+            # Next-hop targets for ``site_crawler.crawl_site``; ignored by
+            # single-URL callers.
+            "links": extract_links(raw_html, url),
         }
 
     @staticmethod

surfsense_backend/app/proprietary/web_crawler/site_crawler.py

@@ -0,0 +1,123 @@
+# SurfSense proprietary crawler engine.
+#
+# Part of the ``app.proprietary`` package; licensed separately from the
+# Apache-2.0 project root (see ``app/proprietary/LICENSE``).
+"""Depth-bounded site crawl built on the single-URL engine (``crawl_url``).
+
+Breadth-first frontier: fetch a page, follow its same-site links one hop deeper,
+dedupe by canonical URL, stop at ``max_crawl_pages``. Every fetch runs through
+``crawl_url`` so tiered fetch, proxy, and captcha handling are reused.
+
+Sequential today; ``_fetch_page`` isolates one unit of work so a bounded worker
+pool can replace the loop later without changing the traversal.
+"""
+
+from __future__ import annotations
+
+from collections import deque
+from dataclasses import dataclass
+
+from app.proprietary.web_crawler.connector import (
+    CrawlOutcome,
+    CrawlOutcomeStatus,
+    WebCrawlerConnector,
+)
+from app.proprietary.web_crawler.url_policy import canonicalize_url, host_of, same_site
+
+
+@dataclass(frozen=True)
+class CrawlPage:
+    """A fetched page plus its crawl provenance (depth and referrer)."""
+
+    url: str
+    status: CrawlOutcomeStatus
+    depth: int
+    referrer: str | None = None
+    loaded_url: str | None = None
+    content: str | None = None
+    metadata: dict[str, str] | None = None
+    error: str | None = None
+    captcha_attempts: int = 0
+    captcha_solved: bool = False
+
+
+async def crawl_site(
+    engine: WebCrawlerConnector,
+    start_urls: list[str],
+    *,
+    max_crawl_depth: int,
+    max_crawl_pages: int,
+) -> list[CrawlPage]:
+    """Crawl ``start_urls`` up to ``max_crawl_depth`` hops / ``max_crawl_pages`` pages.
+
+    Depth 0 fetches only the start URLs. Links are followed only from successful
+    pages, under the depth cap, and only on a start URL's site. Start URLs count
+    toward ``max_crawl_pages``. Order is BFS from the seeds.
+    """
+    allowed_hosts = {host_of(url) for url in start_urls}
+    visited: set[str] = set()
+    frontier: deque[tuple[str, int, str | None]] = deque()
+    for seed in start_urls:
+        key = canonicalize_url(seed)
+        if key not in visited:
+            visited.add(key)
+            frontier.append((seed, 0, None))
+
+    pages: list[CrawlPage] = []
+    while frontier and len(pages) < max_crawl_pages:
+        url, depth, referrer = frontier.popleft()
+        page, outcome = await _fetch_page(engine, url, depth, referrer)
+        pages.append(page)
+
+        if depth >= max_crawl_depth:
+            continue
+        if outcome.status is not CrawlOutcomeStatus.SUCCESS or not outcome.result:
+            continue
+        for link in outcome.result.get("links", []):
+            key = canonicalize_url(link)
+            if key in visited or not same_site(link, allowed_hosts):
+                continue
+            visited.add(key)
+            frontier.append((link, depth + 1, url))
+    return pages
+
+
+async def _fetch_page(
+    engine: WebCrawlerConnector,
+    url: str,
+    depth: int,
+    referrer: str | None,
+) -> tuple[CrawlPage, CrawlOutcome]:
+    """Fetch one URL and map it to a ``CrawlPage`` (the future concurrency unit)."""
+    outcome = await engine.crawl_url(url)
+    return _to_page(url, outcome, depth, referrer), outcome
+
+
+def _to_page(
+    url: str,
+    outcome: CrawlOutcome,
+    depth: int,
+    referrer: str | None,
+) -> CrawlPage:
+    result = outcome.result or {}
+    if outcome.status is CrawlOutcomeStatus.SUCCESS and result:
+        return CrawlPage(
+            url=url,
+            status=outcome.status,
+            depth=depth,
+            referrer=referrer,
+            loaded_url=result.get("loaded_url") or url,
+            content=result.get("content"),
+            metadata=result.get("metadata"),
+            captcha_attempts=outcome.captcha_attempts,
+            captcha_solved=outcome.captcha_solved,
+        )
+    return CrawlPage(
+        url=url,
+        status=outcome.status,
+        depth=depth,
+        referrer=referrer,
+        error=outcome.error,
+        captcha_attempts=outcome.captcha_attempts,
+        captcha_solved=outcome.captcha_solved,
+    )

surfsense_backend/app/proprietary/web_crawler/url_policy.py

@@ -0,0 +1,59 @@
+# SurfSense proprietary crawler engine.
+#
+# Part of the ``app.proprietary`` package; licensed separately from the
+# Apache-2.0 project root (see ``app/proprietary/LICENSE``).
+"""URL helpers for the site crawler: link extraction, canonical key, same-site scope.
+
+Pure functions (no I/O) so the crawl frontier stays deterministic and testable.
+"""
+
+from __future__ import annotations
+
+from urllib.parse import urldefrag, urljoin, urlsplit
+
+from lxml import html as lxml_html
+from lxml.etree import ParserError
+from w3lib.url import canonicalize_url as _w3lib_canonicalize_url
+
+
+def extract_links(page_html: str | None, base_url: str) -> list[str]:
+    """Absolute, http(s), fragment-free, de-duplicated ``<a href>`` targets.
+
+    Relative hrefs resolve against ``base_url``; the page's own URL is dropped.
+    First-seen order is preserved to keep the frontier stable.
+    """
+    if not page_html or not page_html.strip():
+        return []
+    try:
+        root = lxml_html.fromstring(page_html)
+    except (ParserError, ValueError):
+        return []
+
+    self_url, _ = urldefrag(base_url)
+    seen: set[str] = set()
+    links: list[str] = []
+    for href in root.xpath("//a/@href"):
+        target, _ = urldefrag(urljoin(base_url, href.strip()))
+        if urlsplit(target).scheme not in ("http", "https"):
+            continue
+        if target == self_url or target in seen:
+            continue
+        seen.add(target)
+        links.append(target)
+    return links
+
+
+def canonicalize_url(url: str) -> str:
+    """Stable visited-set key: sorts query, normalizes encoding, drops fragment."""
+    return _w3lib_canonicalize_url(url, keep_fragments=False)
+
+
+def host_of(url: str) -> str:
+    """Lowercased host with a leading ``www.`` removed, for same-site matching."""
+    host = (urlsplit(url).hostname or "").lower()
+    return host[4:] if host.startswith("www.") else host
+
+
+def same_site(url: str, allowed_hosts: set[str]) -> bool:
+    """Whether ``url``'s host (``www.``-normalized) is in ``allowed_hosts``."""
+    return host_of(url) in allowed_hosts

surfsense_backend/tests/unit/proprietary/web_crawler/test_connector_links.py

@@ -0,0 +1,29 @@
+"""``_build_result`` surfaces absolute page links for the spider to enqueue."""
+
+from __future__ import annotations
+
+import pytest
+
+from app.proprietary.web_crawler import WebCrawlerConnector
+
+pytestmark = pytest.mark.unit
+
+
+def test_build_result_includes_absolute_links() -> None:
+    html = (
+        "<html><body>"
+        '<a href="/a">A</a>'
+        '<a href="https://example.com/b">B</a>'
+        "</body></html>"
+    )
+
+    result = WebCrawlerConnector()._build_result(
+        html,
+        "https://example.com/",
+        "scrapling-static",
+        allow_raw_fallback=True,
+    )
+
+    assert result is not None
+    assert "https://example.com/a" in result["links"]
+    assert "https://example.com/b" in result["links"]

surfsense_backend/tests/unit/proprietary/web_crawler/test_site_crawler.py

@@ -0,0 +1,165 @@
+"""``crawl_site`` BFS behavior: depth, same-site scope, dedupe, and page caps.
+
+Boundary mocked: the engine (a fake ``crawl_url`` serving a canned link graph).
+NOT mocked: the frontier/depth/dedupe/scope logic under test.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.proprietary.web_crawler import CrawlOutcome, CrawlOutcomeStatus
+from app.proprietary.web_crawler.site_crawler import crawl_site
+
+pytestmark = pytest.mark.unit
+
+_SUCCESS = CrawlOutcomeStatus.SUCCESS
+_FAILED = CrawlOutcomeStatus.FAILED
+
+
+class _FakeEngine:
+    """Serves a canned ``(status, links)`` per URL and records fetch order."""
+
+    def __init__(self, graph: dict[str, tuple[CrawlOutcomeStatus, list[str]]]):
+        self._graph = graph
+        self.calls: list[str] = []
+
+    async def crawl_url(self, url: str) -> CrawlOutcome:
+        self.calls.append(url)
+        status, links = self._graph.get(url, (_FAILED, []))
+        if status is _SUCCESS:
+            return CrawlOutcome(
+                status=_SUCCESS,
+                result={
+                    "content": f"C:{url}",
+                    "metadata": {"title": url},
+                    "links": links,
+                },
+            )
+        return CrawlOutcome(status=status, error="boom")
+
+
+async def test_depth_zero_fetches_only_the_seed() -> None:
+    engine = _FakeEngine({"https://e.com/": (_SUCCESS, ["https://e.com/a"])})
+
+    pages = await crawl_site(
+        engine, ["https://e.com/"], max_crawl_depth=0, max_crawl_pages=10
+    )
+
+    assert engine.calls == ["https://e.com/"]
+    assert len(pages) == 1
+    assert pages[0].depth == 0
+    assert pages[0].referrer is None
+    assert pages[0].content == "C:https://e.com/"
+
+
+async def test_depth_one_follows_same_site_links_only() -> None:
+    engine = _FakeEngine(
+        {
+            "https://e.com/": (
+                _SUCCESS,
+                ["https://e.com/a", "https://e.com/b", "https://other.com/x"],
+            ),
+            "https://e.com/a": (_SUCCESS, []),
+            "https://e.com/b": (_SUCCESS, []),
+        }
+    )
+
+    pages = await crawl_site(
+        engine, ["https://e.com/"], max_crawl_depth=1, max_crawl_pages=10
+    )
+
+    assert set(engine.calls) == {
+        "https://e.com/",
+        "https://e.com/a",
+        "https://e.com/b",
+    }
+    depths = {page.url: page.depth for page in pages}
+    assert depths["https://e.com/a"] == 1
+    referrers = {page.url: page.referrer for page in pages}
+    assert referrers["https://e.com/a"] == "https://e.com/"
+
+
+async def test_depth_caps_further_recursion() -> None:
+    engine = _FakeEngine(
+        {
+            "https://e.com/": (_SUCCESS, ["https://e.com/a"]),
+            "https://e.com/a": (_SUCCESS, ["https://e.com/b"]),
+            "https://e.com/b": (_SUCCESS, []),
+        }
+    )
+
+    pages = await crawl_site(
+        engine, ["https://e.com/"], max_crawl_depth=1, max_crawl_pages=10
+    )
+
+    # depth 1 reaches /a but must NOT descend to /b (depth 2).
+    assert set(engine.calls) == {"https://e.com/", "https://e.com/a"}
+    assert all(page.url != "https://e.com/b" for page in pages)
+
+
+async def test_max_pages_caps_total_fetches() -> None:
+    graph: dict[str, tuple[CrawlOutcomeStatus, list[str]]] = {
+        "https://e.com/": (_SUCCESS, [f"https://e.com/{i}" for i in range(10)])
+    }
+    for i in range(10):
+        graph[f"https://e.com/{i}"] = (_SUCCESS, [])
+    engine = _FakeEngine(graph)
+
+    pages = await crawl_site(
+        engine, ["https://e.com/"], max_crawl_depth=1, max_crawl_pages=3
+    )
+
+    assert len(pages) == 3
+    assert len(engine.calls) == 3
+
+
+async def test_dedupes_on_canonical_url() -> None:
+    engine = _FakeEngine(
+        {
+            "https://e.com/": (
+                _SUCCESS,
+                ["https://e.com/a", "https://e.com/a#frag", "https://e.com/a?"],
+            ),
+            "https://e.com/a": (_SUCCESS, ["https://e.com/"]),  # links back to seed
+        }
+    )
+
+    await crawl_site(
+        engine, ["https://e.com/"], max_crawl_depth=3, max_crawl_pages=10
+    )
+
+    assert engine.calls.count("https://e.com/a") == 1
+    assert engine.calls.count("https://e.com/") == 1
+
+
+async def test_failed_page_is_recorded_and_not_expanded() -> None:
+    engine = _FakeEngine({"https://e.com/": (_FAILED, [])})
+
+    pages = await crawl_site(
+        engine, ["https://e.com/"], max_crawl_depth=2, max_crawl_pages=10
+    )
+
+    assert len(pages) == 1
+    assert pages[0].status is _FAILED
+    assert pages[0].content is None
+    assert pages[0].error == "boom"
+
+
+async def test_multiple_seeds_are_all_entry_points() -> None:
+    engine = _FakeEngine(
+        {
+            "https://a.com/": (_SUCCESS, []),
+            "https://b.com/": (_SUCCESS, []),
+        }
+    )
+
+    pages = await crawl_site(
+        engine,
+        ["https://a.com/", "https://b.com/"],
+        max_crawl_depth=0,
+        max_crawl_pages=10,
+    )
+
+    assert {page.url for page in pages} == {"https://a.com/", "https://b.com/"}
+    assert all(page.depth == 0 for page in pages)

surfsense_backend/tests/unit/proprietary/web_crawler/test_url_policy.py

@@ -0,0 +1,89 @@
+"""Pure URL-helper behavior for the site spider (link surfacing + dedupe scope)."""
+
+from __future__ import annotations
+
+import pytest
+
+from app.proprietary.web_crawler.url_policy import (
+    canonicalize_url,
+    extract_links,
+    host_of,
+    same_site,
+)
+
+pytestmark = pytest.mark.unit
+
+_HTML = """
+<html><body>
+  <a href="/about">About (relative)</a>
+  <a href="https://example.com/contact">Contact (absolute)</a>
+  <a href="https://other.com/x">External</a>
+  <a href="#top">Anchor only</a>
+  <a href="mailto:a@b.com">Mail</a>
+  <a href="javascript:void(0)">JS</a>
+  <a href="/about">Duplicate about</a>
+</body></html>
+"""
+
+
+def test_extract_links_absolutizes_relative_hrefs() -> None:
+    links = extract_links(_HTML, "https://example.com/home")
+    assert "https://example.com/about" in links
+    assert "https://example.com/contact" in links
+
+
+def test_extract_links_keeps_only_http_schemes() -> None:
+    links = extract_links(_HTML, "https://example.com/home")
+    assert all(link.startswith(("http://", "https://")) for link in links)
+    assert not any("mailto" in link or "javascript" in link for link in links)
+
+
+def test_extract_links_drops_pure_anchor() -> None:
+    links = extract_links(_HTML, "https://example.com/home")
+    assert "https://example.com/home" not in links  # bare "#top" resolves to self
+
+
+def test_extract_links_dedupes_preserving_first_occurrence() -> None:
+    links = extract_links(_HTML, "https://example.com/home")
+    assert links.count("https://example.com/about") == 1
+
+
+def test_extract_links_strips_fragments() -> None:
+    assert extract_links('<a href="/p#sec">x</a>', "https://e.com") == [
+        "https://e.com/p"
+    ]
+
+
+def test_extract_links_on_empty_or_blank_html_is_empty() -> None:
+    assert extract_links("", "https://e.com") == []
+    assert extract_links("   ", "https://e.com") == []
+    assert extract_links(None, "https://e.com") == []
+
+
+def test_canonicalize_lowercases_host_sorts_query_and_drops_fragment() -> None:
+    assert (
+        canonicalize_url("https://E.com/a?b=2&a=1#frag") == "https://e.com/a?a=1&b=2"
+    )
+
+
+def test_canonicalize_collapses_fragment_and_empty_query_to_one_key() -> None:
+    # The three forms must dedupe to the same visited-set key.
+    canonical = canonicalize_url("https://e.com/a")
+    assert canonicalize_url("https://e.com/a#frag") == canonical
+    assert canonicalize_url("https://e.com/a?") == canonical
+
+
+def test_canonicalize_keeps_nondefault_port() -> None:
+    assert canonicalize_url("https://e.com:8443/x") == "https://e.com:8443/x"
+
+
+def test_host_of_strips_www_and_lowercases() -> None:
+    assert host_of("https://www.Example.com/x") == "example.com"
+    assert host_of("https://Example.com/x") == "example.com"
+
+
+def test_same_site_matches_on_normalized_host() -> None:
+    allowed = {"example.com"}
+    assert same_site("https://www.example.com/a", allowed) is True
+    assert same_site("https://example.com/b", allowed) is True
+    assert same_site("https://other.com/c", allowed) is False