Coverage for src/lilbee/crawler/crawl4ai_fetcher.py: 100%
195 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 11:46 +0000
1"""crawl4ai-backed implementation of :class:`lilbee.crawler.fetcher.WebFetcher`."""
3from __future__ import annotations
5import contextlib
6import functools
7import inspect
8import io
9import logging
10import math
11from collections.abc import AsyncGenerator, AsyncIterator
12from dataclasses import dataclass
13from typing import TYPE_CHECKING, Any
14from urllib.parse import urlparse
16import anyio.to_thread
17from anyio import CapacityLimiter
19from lilbee.core.config import cfg
20from lilbee.core.config.enums import CrawlRenderMode
21from lilbee.crawler import bootstrap
22from lilbee.crawler.bootstrap import CrawlerBrowserError
23from lilbee.crawler.markdown import base_url_for, html_to_markdown
24from lilbee.crawler.models import (
25 CancelToken,
26 ConcurrencySpec,
27 FetchedPage,
28 FilterSpec,
29)
30from lilbee.crawler.url_filter import host_in_scope, validate_crawl_url
32if TYPE_CHECKING:
33 from lilbee.crawler.fetcher import WebFetcher
35log = logging.getLogger(__name__)
38def _build_inner_crawler(*, verbose: bool, render_mode: CrawlRenderMode) -> Any:
39 """Construct a crawl4ai ``AsyncWebCrawler`` for the requested render mode.
41 HTTP mode swaps in the browserless HTTP strategy; browser mode tunes
42 Chromium for memory (light/text/memory-saving + periodic process recycle),
43 reading the recycle threshold and launch flags from config.
44 """
45 from crawl4ai import AsyncWebCrawler
47 if render_mode is CrawlRenderMode.HTTP:
48 from crawl4ai.async_crawler_strategy import AsyncHTTPCrawlerStrategy
50 return AsyncWebCrawler(crawler_strategy=AsyncHTTPCrawlerStrategy(), verbose=verbose)
52 from crawl4ai import BrowserConfig
54 config = BrowserConfig(
55 light_mode=True,
56 text_mode=True,
57 memory_saving_mode=True,
58 max_pages_before_recycle=cfg.crawl_browser_recycle_pages,
59 extra_args=list(cfg.crawl_browser_extra_args),
60 verbose=verbose,
61 )
62 return AsyncWebCrawler(config=config, verbose=verbose)
65def _conversion_limiter() -> CapacityLimiter | None:
66 """How many pages may convert off the event loop at once, or ``None`` for on it.
68 Read once per crawl, so a crawl keeps the setting it started with.
69 """
70 workers = cfg.crawl_convert_workers
71 return CapacityLimiter(workers) if workers >= 1 else None
74def _silent_markdown_generator() -> Any | None:
75 """A generator that produces nothing, or ``None`` when crawl4ai's base is absent.
77 crawl4ai converts every page inside its own async call stack, with no setting
78 to skip it. Handing it a generator that returns immediately leaves the HTML
79 for :func:`html_to_markdown` to convert where lilbee can await it.
80 """
81 try:
82 from crawl4ai.markdown_generation_strategy import MarkdownGenerationStrategy
83 from crawl4ai.models import MarkdownGenerationResult
84 except (ImportError, AttributeError):
85 return None
87 class _SilentMarkdownGenerator(MarkdownGenerationStrategy): # type: ignore[misc]
88 def generate_markdown(self, *args: Any, **kwargs: Any) -> Any:
89 return MarkdownGenerationResult(
90 raw_markdown="",
91 markdown_with_citations="",
92 references_markdown="",
93 fit_markdown="",
94 fit_html="",
95 )
97 return _SilentMarkdownGenerator()
100@dataclass(frozen=True)
101class _Conversion:
102 """The markdown seam for one crawl: crawl4ai's own generator silenced so lilbee re-converts.
104 ``generator`` is the silent generator handed to crawl4ai, or ``None`` when its markdown
105 base class is unavailable and the crawl converts itself. ``limiter`` bounds how many pages
106 convert on the thread pool at once, or ``None`` to convert inline on the loop.
107 """
109 generator: Any | None
110 limiter: CapacityLimiter | None
112 @property
113 def config_kwargs(self) -> dict[str, Any]:
114 """``CrawlerRunConfig`` kwargs that install the silent generator, if there is one."""
115 return {} if self.generator is None else {"markdown_generator": self.generator}
117 async def markdown_for(self, result: Any) -> str:
118 """The page's markdown, re-converted off the loop only when the backend was silenced.
120 An un-silenced backend already converted the page, so re-converting would
121 duplicate the work this exists to move. Converts ``cleaned_html`` only, the
122 same source crawl4ai's own generator uses, so an empty cleaned page stays
123 empty rather than turning raw nav/boilerplate into content.
124 """
125 html = result.cleaned_html or ""
126 if self.generator is None or not html:
127 return str(result.markdown or "")
128 base_url = base_url_for(result.html or "", result.url, result.redirected_url)
129 if self.limiter is None:
130 return html_to_markdown(html, base_url)
131 return await anyio.to_thread.run_sync(
132 html_to_markdown, html, base_url, limiter=self.limiter
133 )
136def _new_conversion() -> _Conversion:
137 """Build the conversion seam for one crawl, reading ``crawl_convert_workers`` once."""
138 generator = _silent_markdown_generator()
139 limiter = _conversion_limiter() if generator is not None else None
140 return _Conversion(generator, limiter)
143def _build_rate_limited_dispatcher(
144 concurrency: ConcurrencySpec, render_mode: CrawlRenderMode
145) -> Any:
146 """Build the recursive-crawl dispatcher from a ConcurrencySpec, or None.
148 BFSDeepCrawlStrategy calls ``crawler.arun_many()`` without a dispatcher
149 kwarg, so per-domain rate limiting is only reachable by threading a
150 dispatcher through AsyncWebCrawler itself. Browser mode uses a
151 MemoryAdaptiveDispatcher so a crawl backs off when system memory is tight
152 rather than steamrolling the machine; HTTP mode is light enough to stay on
153 the plain semaphore path.
154 """
155 if not concurrency.retry_on_rate_limit:
156 return None
157 from crawl4ai.async_dispatcher import RateLimiter
159 rate_limiter = RateLimiter(
160 base_delay=(concurrency.retry_base_delay_min, concurrency.retry_base_delay_max),
161 max_delay=concurrency.retry_max_backoff,
162 max_retries=concurrency.retry_max_attempts,
163 )
164 if render_mode is CrawlRenderMode.BROWSER:
165 from crawl4ai.async_dispatcher import MemoryAdaptiveDispatcher
167 return MemoryAdaptiveDispatcher(
168 max_session_permit=concurrency.semaphore_count,
169 rate_limiter=rate_limiter,
170 )
171 from crawl4ai.async_dispatcher import SemaphoreDispatcher
173 return SemaphoreDispatcher(
174 semaphore_count=concurrency.semaphore_count,
175 rate_limiter=rate_limiter,
176 )
179class _LilbeeAsyncCrawler:
180 """AsyncWebCrawler wrapper that injects a default dispatcher on ``arun_many``.
182 BFSDeepCrawlStrategy calls ``arun_many`` without a dispatcher kwarg, so the
183 wrapper supplies one to make rate limiting and 429/503 retries reachable.
184 An explicit ``dispatcher=`` on the call still wins.
185 """
187 def __init__(self, inner: Any, *, dispatcher: Any) -> None:
188 self._inner = inner
189 self._dispatcher = dispatcher
191 async def __aenter__(self) -> _LilbeeAsyncCrawler:
192 await self._inner.__aenter__()
193 return self
195 async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> Any:
196 return await self._inner.__aexit__(exc_type, exc, tb)
198 async def arun(self, *args: Any, **kwargs: Any) -> Any:
199 return await self._inner.arun(*args, **kwargs)
201 async def arun_many(
202 self, urls: Any, config: Any = None, dispatcher: Any = None, **kwargs: Any
203 ) -> Any:
204 return await self._inner.arun_many(
205 urls,
206 config=config,
207 dispatcher=dispatcher if dispatcher is not None else self._dispatcher,
208 **kwargs,
209 )
212@contextlib.asynccontextmanager
213async def _open_crawler(
214 *, quiet: bool = False, render_mode: CrawlRenderMode, dispatcher: Any = None
215) -> AsyncIterator[Any]:
216 """Open an AsyncWebCrawler for ``render_mode``, wrapping with the dispatcher.
218 Browser mode requires the Chromium binary and raises
219 :class:`CrawlerBrowserError` if it is missing, so Playwright's ASCII install
220 banner does not leak into the TUI. HTTP mode needs no browser at all.
221 """
222 if render_mode is CrawlRenderMode.BROWSER and not bootstrap.chromium_installed():
223 raise CrawlerBrowserError(
224 "Playwright Chromium browser not installed. "
225 "Run 'uv run playwright install chromium' to enable browser-mode crawling."
226 )
228 inner = _build_inner_crawler(verbose=not quiet, render_mode=render_mode)
230 stdout_ctx = contextlib.redirect_stdout(io.StringIO()) if quiet else contextlib.nullcontext()
231 stderr_ctx = contextlib.redirect_stderr(io.StringIO()) if quiet else contextlib.nullcontext()
232 with stdout_ctx, stderr_ctx:
233 if dispatcher is not None:
234 async with _LilbeeAsyncCrawler(inner, dispatcher=dispatcher) as crawler:
235 yield crawler
236 else:
237 async with inner as crawler:
238 yield crawler
241def _safe_strategy_cancel(strategy: Any) -> None:
242 """Call ``strategy.cancel()`` if available; swallow only the known SDK shapes.
244 Narrow catch: ``AttributeError`` covers a missing nested attribute mid-call;
245 ``RuntimeError`` covers cancel-on-closed-strategy. Anything else propagates.
246 """
247 cancel_method = getattr(strategy, "cancel", None)
248 if callable(cancel_method):
249 try:
250 cancel_method()
251 except (AttributeError, RuntimeError) as exc:
252 log.debug("strategy.cancel() raised: %s", exc)
255async def _safe_aclose(stream: Any) -> None:
256 """Close an async generator stream; no-op for list / single-result shapes.
258 aclose() runs the generator's own cleanup, which can surface arbitrary
259 downstream errors. A teardown failure must not mask the crawl's real result,
260 so it is logged at debug rather than propagated or silently dropped.
261 """
262 if stream is None:
263 return
264 if inspect.isasyncgen(stream):
265 try:
266 await stream.aclose()
267 except Exception as exc:
268 log.debug("crawl stream aclose() raised during teardown: %s", exc)
271async def _iter_crawl_stream(stream: Any) -> AsyncIterator[Any]:
272 """Normalize crawl4ai's ``arun()`` return (async generator, list, or single result)."""
273 if inspect.isasyncgen(stream):
274 async for item in stream:
275 yield item
276 return
277 # A list is the batch-mode shape; iterate and yield each item.
278 if isinstance(stream, list):
279 for item in stream:
280 yield item
281 return
282 yield stream
285def _link_passes_ssrf(url: str) -> bool:
286 """Return True when a discovered link resolves to a public, http(s) target.
288 Re-validates every followed link against the IP blocklist so a discovered
289 link to a private/metadata host is dropped before fetch. This is a
290 best-effort check at filter time, not DNS-rebinding protection: the fetcher
291 resolves the host again when it connects, so a record that rebinds between
292 this check and the fetch is a TOCTOU window this does not close.
293 """
294 try:
295 validate_crawl_url(url)
296 except ValueError:
297 return False
298 return True
301def _host_scope_filter(start_url: str, *, include_subdomains: bool) -> Any:
302 """Build a URLFilter that scopes a crawl to the starting URL's host.
304 Default behavior (``include_subdomains=False``) restricts link-following to
305 the exact host of *start_url*. When ``include_subdomains=True`` the host
306 plus any subdomain is in scope. Either way every followed link is also
307 re-validated against the SSRF blocklist, since the host scope check alone
308 would let a same-host link that resolves to a private IP through.
309 """
310 from crawl4ai.deep_crawling.filters import URLFilter
312 host = (urlparse(start_url).hostname or "").lower()
313 if not host:
314 return None
316 class _ScopedSsrfFilter(URLFilter): # type: ignore[misc]
317 def apply(self, url: str) -> bool:
318 link_host = (urlparse(url).hostname or "").lower()
319 ok = host_in_scope(
320 link_host, host, include_subdomains=include_subdomains
321 ) and _link_passes_ssrf(url)
322 self._update_stats(ok)
323 return ok
325 return _ScopedSsrfFilter()
328class Crawl4aiFetcher:
329 """:class:`WebFetcher` implementation backed by crawl4ai."""
331 def __init__(self, *, quiet: bool = False, render_mode: CrawlRenderMode) -> None:
332 self._quiet = quiet
333 self._render_mode = render_mode
335 async def __aenter__(self) -> Crawl4aiFetcher:
336 # Crawl4ai opens a fresh ``AsyncWebCrawler`` per operation because
337 # ``fetch_recursive`` needs a per-call dispatcher (which depends on
338 # the :class:`ConcurrencySpec` for that call). Nothing to set up here.
339 return self
341 async def __aexit__(
342 self,
343 exc_type: type[BaseException] | None,
344 exc: BaseException | None,
345 tb: Any,
346 ) -> None:
347 return None
349 async def fetch_single(self, url: str, *, timeout: float) -> FetchedPage:
350 """Fetch a single URL via crawl4ai's ``arun``."""
351 from crawl4ai import CrawlerRunConfig
353 conversion = _new_conversion()
354 config = CrawlerRunConfig(page_timeout=int(timeout * 1000), **conversion.config_kwargs)
355 async with _open_crawler(quiet=self._quiet, render_mode=self._render_mode) as crawler:
356 result = await crawler.arun(url=url, config=config)
357 markdown = (await conversion.markdown_for(result)).strip()
358 if markdown:
359 return FetchedPage(url=url, markdown=markdown, success=True)
360 return FetchedPage(
361 url=url,
362 success=False,
363 error=result.error_message or "No content extracted",
364 )
366 async def fetch_recursive(
367 self,
368 seed_url: str,
369 *,
370 depth: int | None,
371 max_pages: int | None,
372 timeout: float,
373 concurrency: ConcurrencySpec,
374 filters: FilterSpec,
375 cancel: CancelToken | None = None,
376 ) -> AsyncGenerator[FetchedPage, None]:
377 """Stream pages discovered by crawl4ai's native BFS.
379 ``depth`` / ``max_pages`` of ``None`` mean unbounded; the adapter
380 translates to ``math.inf`` for crawl4ai's BFSDeepCrawlStrategy, which
381 is the sentinel it understands.
382 """
384 def _should_cancel() -> bool:
385 return cancel is not None and cancel.is_set()
387 from crawl4ai import CrawlerRunConfig
388 from crawl4ai.deep_crawling import BFSDeepCrawlStrategy
389 from crawl4ai.deep_crawling.filters import FilterChain, URLPatternFilter
391 filter_chain_items: list[Any] = []
392 host_filter = _host_scope_filter(seed_url, include_subdomains=filters.include_subdomains)
393 if host_filter is not None:
394 filter_chain_items.append(host_filter)
395 if filters.exclude_patterns:
396 filter_chain_items.append(
397 URLPatternFilter(filters.exclude_patterns, use_glob=False, reverse=True)
398 )
399 filter_chain = FilterChain(filter_chain_items) if filter_chain_items else FilterChain()
401 strategy = BFSDeepCrawlStrategy(
402 max_depth=math.inf if depth is None else depth,
403 max_pages=math.inf if max_pages is None else max_pages,
404 should_cancel=_should_cancel,
405 filter_chain=filter_chain,
406 )
407 conversion = _new_conversion()
408 config = CrawlerRunConfig(
409 deep_crawl_strategy=strategy,
410 page_timeout=int(timeout * 1000),
411 mean_delay=concurrency.mean_delay,
412 max_range=concurrency.max_delay_range,
413 semaphore_count=concurrency.semaphore_count,
414 stream=True,
415 **conversion.config_kwargs,
416 )
418 dispatcher = _build_rate_limited_dispatcher(concurrency, self._render_mode)
419 stream: Any = None
420 strategy_cancelled = False
421 # Exceptions propagate to the orchestration layer, which decides
422 # whether to log cancel-teardown noise at debug vs surface a real
423 # failure. The adapter's only housekeeping is stream close + BFS
424 # strategy cancel so Playwright tears down in order.
425 async with _open_crawler(
426 quiet=self._quiet, render_mode=self._render_mode, dispatcher=dispatcher
427 ) as crawler:
428 stream = await crawler.arun(url=seed_url, config=config)
429 try:
430 async for cr in _iter_crawl_stream(stream):
431 if _should_cancel():
432 _safe_strategy_cancel(strategy)
433 strategy_cancelled = True
434 break
435 if cr.success:
436 yield FetchedPage(url=cr.url, markdown=await conversion.markdown_for(cr))
437 else:
438 yield FetchedPage(
439 url=cr.url,
440 success=False,
441 error=cr.error_message or "Unknown error",
442 )
443 finally:
444 # If the consumer breaks out before we saw a cancel, still
445 # short-circuit the BFS strategy so any in-flight arun_many
446 # batch stops dispatching. Mirrors the orchestrator's
447 # previous "hard cap on visible counter" behavior now that
448 # the strategy object lives inside the adapter.
449 if not strategy_cancelled:
450 _safe_strategy_cancel(strategy)
451 # Close the async generator (if it is one) before the
452 # crawler context exits, so Playwright tears down
453 # in-flight URLs in order. Skipping this is what produced
454 # the "BrowserContext.new_page: Connection closed" spam
455 # on cancel.
456 await _safe_aclose(stream)
459# Protocol conformance check: Crawl4aiFetcher is structurally a WebFetcher.
460# We don't instantiate at import time so the check stays purely structural.
461if TYPE_CHECKING:
462 _: WebFetcher = Crawl4aiFetcher(render_mode=CrawlRenderMode.HTTP)
465@functools.cache
466def crawler_available() -> bool:
467 """Check if the crawl4ai backend is importable (i.e. the extra is installed).
469 Uses ``importlib.util.find_spec`` rather than ``import crawl4ai`` so the
470 check stays fast on the UI thread. ``crawl4ai`` is in AGENTS.md's
471 known-heavy-imports list; executing it on Windows with Defender
472 real-time scanning takes seconds, and the Settings screen's feature-
473 gate call (``_FEATURE_GATED_GROUPS``) hits it synchronously during
474 ``compose``. ``find_spec`` just walks ``sys.path`` to locate the
475 package; the actual import runs later from the crawler bootstrap
476 where the cost is expected.
477 """
478 import importlib.util
480 return importlib.util.find_spec("crawl4ai") is not None