Coverage for src/lilbee/crawler/runner.py: 100%
192 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 11:46 +0000
1"""Crawl orchestration: build specs from ``cfg``, drive a :class:`WebFetcher`.
3By default a recursive crawl is scoped to the exact starting host so a
4Wikipedia article does not wander into other language editions. Callers
5opt into subdomain scope via ``include_subdomains=True``.
6"""
8from __future__ import annotations
10import asyncio
11import logging
12import threading
13import time
14from collections.abc import Awaitable, Callable
15from dataclasses import dataclass
16from datetime import UTC, datetime
17from pathlib import Path
18from typing import Any
20from lilbee.app.services import get_services
21from lilbee.core.config import cfg
22from lilbee.core.config.enums import CrawlRenderMode
23from lilbee.crawler import bootstrap, save, sitemap
24from lilbee.crawler.bootstrap import CrawlerBrowserError
25from lilbee.crawler.crawl4ai_fetcher import Crawl4aiFetcher
26from lilbee.crawler.discovery import build_concurrency_spec, build_filter_spec
27from lilbee.crawler.events import (
28 _drain_page_stream,
29 _fetched_to_result,
30 _handle_crawl_teardown_error,
31 _pages_cap,
32)
33from lilbee.crawler.models import CRAWL_PAGES_UNLIMITED, CrawlResult
34from lilbee.crawler.save import METADATA_FLUSH_INTERVAL, CrawlMeta
35from lilbee.crawler.url_filter import validate_crawl_url
36from lilbee.runtime.progress import (
37 CrawlDoneEvent,
38 CrawlPageEvent,
39 CrawlStartEvent,
40 DetailedProgressCallback,
41 EventType,
42 SetupDoneEvent,
43 SetupStartEvent,
44)
46# Component name for the browser-warmup setup phase (distinct from the
47# Chromium download, whose component is "chromium"). The crawl emits a
48# start/done bracket around opening the crawler so the Task Center shows a
49# "preparing crawler" stage instead of a silent stall on first use.
50_BROWSER_SETUP_COMPONENT = "browser"
52log = logging.getLogger(__name__)
55def _get_crawl_semaphore() -> asyncio.Semaphore | None:
56 """Return the process-wide crawl semaphore, or None when unlimited."""
57 return get_services().crawler_semaphore
60def _resolve_depth(value: int | None, cfg_ceiling: int | None) -> int | None:
61 """Resolve a crawl depth to the value the dispatcher consumes.
63 Depth has its own contract, distinct from the page-count limit: ``0`` is a
64 valid "seed only / single page" depth, not "unbounded". (Page counts use
65 :func:`_resolve_page_limit`, where ``0`` means "no limit".)
67 None -> cfg_ceiling (itself may be None; ``None`` means unbounded)
68 n >= 0 -> n (0 = seed only; explicit caller intent overrides cfg)
69 n < 0 -> ValueError (use None for unbounded)
70 """
71 effective = value if value is not None else cfg_ceiling
72 if effective is None:
73 return None
74 if effective < 0:
75 raise ValueError("crawl depth must be 0 (seed only) or a positive int")
76 return effective
79def _resolve_page_limit(max_pages: int | None) -> int | None:
80 """Resolve the page bound the fetcher consumes (None means unbounded).
82 ``CRAWL_PAGES_UNLIMITED`` (0) is an explicit "no limit" and returns None.
83 ``None`` is unspecified: it falls back to ``cfg.crawl_max_pages`` if set,
84 else the protective default ``cfg.crawl_safety_max_pages`` so a hostile site
85 can't exhaust the disk on a crawl nobody bounded. A positive int is honored
86 as-is, even above the default.
87 """
88 if max_pages == CRAWL_PAGES_UNLIMITED:
89 return None
90 if max_pages is not None:
91 return max_pages
92 if cfg.crawl_max_pages is not None:
93 return cfg.crawl_max_pages
94 return cfg.crawl_safety_max_pages
97def _looks_like_missing_chromium(exc: BaseException) -> bool:
98 """Heuristic for the Playwright "Executable doesn't exist" launch failure."""
99 return "Executable doesn't exist" in str(exc)
102async def crawl_single(
103 url: str,
104 *,
105 quiet: bool = False,
106 on_progress: DetailedProgressCallback | None = None,
107 render_mode: CrawlRenderMode = CrawlRenderMode.BROWSER,
108) -> CrawlResult:
109 """Fetch a single URL.
111 ``render_mode`` defaults to ``BROWSER`` for direct callers; the public
112 entry point :func:`crawl_and_save` resolves it from ``cfg.crawl_render_mode``
113 and passes the canonical value down.
115 Raises :class:`CrawlerBackendError` if the crawler extra isn't installed.
116 On a "Chromium executable missing" launch failure, re-runs the
117 bootstrap once and retries -- ``chromium_installed()`` can return True
118 when the wrong revision lives in the cache root, in which case the
119 launch fails the first attempt.
121 ``on_progress`` receives a setup_start/setup_done bracket around opening
122 the crawler so the first crawl's browser warmup is visible rather than a
123 silent stall.
124 """
125 validate_crawl_url(url)
126 from lilbee.crawler import crawler_available
128 if not crawler_available():
129 raise bootstrap.CrawlerBackendError(
130 "Web crawling is not available. Run 'uv sync --extra crawler' to enable it."
131 )
132 # The setup bracket exists to surface the Chromium warmup, which only
133 # happens in browser mode; HTTP mode opens a browserless client with no
134 # warmup, so emitting a "browser" setup stage there would be misleading.
135 emit_setup = render_mode is CrawlRenderMode.BROWSER
136 if on_progress is not None and emit_setup:
137 on_progress(EventType.SETUP_START, SetupStartEvent(component=_BROWSER_SETUP_COMPONENT))
138 try:
139 async with Crawl4aiFetcher(quiet=quiet, render_mode=render_mode) as fetcher:
140 if on_progress is not None and emit_setup:
141 on_progress(
142 EventType.SETUP_DONE,
143 SetupDoneEvent(component=_BROWSER_SETUP_COMPONENT, success=True),
144 )
145 page = await fetcher.fetch_single(url, timeout=cfg.crawl_timeout)
146 return _fetched_to_result(page)
147 except CrawlerBrowserError:
148 raise
149 except Exception as exc:
150 if _looks_like_missing_chromium(exc):
151 log.warning("Chromium missing for %s; bootstrapping then retrying", url)
152 await bootstrap.bootstrap_chromium(on_progress=None)
153 try:
154 async with Crawl4aiFetcher(quiet=quiet, render_mode=render_mode) as fetcher:
155 page = await fetcher.fetch_single(url, timeout=cfg.crawl_timeout)
156 return _fetched_to_result(page)
157 except Exception as retry_exc:
158 log.warning("Crawl retry failed for %s: %s", url, retry_exc)
159 return CrawlResult(url=url, success=False, error=str(retry_exc))
160 log.warning("Failed to crawl %s: %s", url, exc)
161 return CrawlResult(url=url, success=False, error=str(exc))
164async def crawl_recursive(
165 url: str,
166 max_depth: int | None = None,
167 max_pages: int | None = None,
168 on_progress: DetailedProgressCallback | None = None,
169 cancel: threading.Event | None = None,
170 *,
171 quiet: bool = False,
172 include_subdomains: bool = False,
173 on_result: Callable[[CrawlResult], Any] | None = None,
174 render_mode: CrawlRenderMode = CrawlRenderMode.BROWSER,
175) -> list[CrawlResult]:
176 """Crawl a URL recursively using BFS, streaming per-page progress.
178 ``render_mode`` defaults to ``BROWSER`` for direct callers; the public
179 entry point :func:`crawl_and_save` resolves it from ``cfg.crawl_render_mode``
180 and passes the canonical value down.
182 ``max_depth`` of None means unbounded depth. ``max_pages`` of
183 ``CRAWL_PAGES_UNLIMITED`` (0) means no page limit; a positive int is that
184 cap; None is unspecified and falls back to ``cfg.crawl_safety_max_pages`` so
185 a hostile site can't exhaust the disk on a crawl nobody bounded.
186 ``CRAWL_PAGE`` events fire as each page completes; total is
187 ``CRAWL_TOTAL_UNKNOWN`` by default and promoted to the sitemap count
188 when available.
190 Pass ``include_subdomains=True`` to broaden scope from the exact host to the
191 host plus any subdomains. If ``on_result`` is provided, it's called for each
192 streamed ``CrawlResult`` the moment it arrives so callers can flush pages to
193 disk incrementally and keep partial output across cancellation.
194 """
195 validate_crawl_url(url)
196 # ``_run_crawl`` already resolved the depth ceiling (and routed a seed-only
197 # 0 to the single-page path), so the recursive path takes ``max_depth`` as
198 # given: None = unbounded, a positive int = the cap.
199 depth = max_depth
200 pages = _resolve_page_limit(max_pages)
202 # Fail fast when the ``crawler`` extra wasn't installed so SSE
203 # callers see ``event: error`` instead of a silent zero-results run.
204 from lilbee.crawler import crawler_available
206 if not crawler_available():
207 raise bootstrap.CrawlerBackendError(
208 "Web crawling is not available. Run 'uv sync --extra crawler' to enable it."
209 )
211 # Fail fast before pulling in backend submodules so callers get a clean
212 # CrawlerBrowserError instead of a Playwright install banner. HTTP mode
213 # needs no browser, so the guard only applies to browser-mode crawls.
214 if render_mode is CrawlRenderMode.BROWSER and not bootstrap.chromium_installed():
215 raise CrawlerBrowserError(
216 "Playwright Chromium browser not installed. "
217 "Run 'uv run playwright install chromium' to enable browser-mode crawling."
218 )
220 # Best-effort sitemap lookup so the TUI / CLI can render a real page-count
221 # denominator instead of [n/-1]. Falls back to CRAWL_TOTAL_UNKNOWN on any
222 # failure; off the hot path so a slow/missing sitemap never blocks the crawl.
223 sitemap_total = await asyncio.to_thread(
224 sitemap._count_sitemap_urls, url, include_subdomains=include_subdomains
225 )
227 concurrency = build_concurrency_spec()
228 filters = build_filter_spec(include_subdomains=include_subdomains)
230 results: list[CrawlResult] = []
231 # Browser mode launches Chromium, whose one-time warmup can take many
232 # seconds; bracket it with setup events so the Task Center shows a
233 # "preparing crawler" stage instead of a silent stall. HTTP mode has no
234 # browser warmup, so the bracket is skipped to avoid a misleading stage.
235 emit_setup = render_mode is CrawlRenderMode.BROWSER
236 if on_progress is not None and emit_setup:
237 on_progress(EventType.SETUP_START, SetupStartEvent(component=_BROWSER_SETUP_COMPONENT))
238 try:
239 async with Crawl4aiFetcher(quiet=quiet, render_mode=render_mode) as fetcher:
240 if on_progress is not None and emit_setup:
241 on_progress(
242 EventType.SETUP_DONE,
243 SetupDoneEvent(component=_BROWSER_SETUP_COMPONENT, success=True),
244 )
245 # Hold an explicit reference to the generator so we can aclose
246 # it deterministically on break. Without this, the generator's
247 # finally block (which also short-circuits the BFS strategy) only
248 # runs at gc time, which is too late for callers that expect the
249 # strategy to stop the moment we hit ``max_pages``.
250 page_stream = fetcher.fetch_recursive(
251 url,
252 depth=depth,
253 max_pages=pages,
254 timeout=cfg.crawl_timeout,
255 concurrency=concurrency,
256 filters=filters,
257 cancel=cancel,
258 )
259 try:
260 results = await _drain_page_stream(
261 page_stream,
262 on_progress=on_progress,
263 on_result=on_result,
264 sitemap_total=sitemap_total,
265 pages_cap=_pages_cap(pages),
266 cancel=cancel,
267 )
268 finally:
269 await page_stream.aclose()
270 except CrawlerBrowserError:
271 raise
272 except Exception as exc:
273 _handle_crawl_teardown_error(url, exc, cancel=cancel, results=results)
275 return results
278async def _maybe_periodic_sync(tasks: set[asyncio.Task[None]]) -> None:
279 """Fire off a background sync if the ``crawl_sync_interval`` has elapsed.
281 Skips when periodic sync is disabled (``interval=0``) or another sync
282 is already running. The spawned task is added to ``tasks`` so the
283 caller can drain it before returning.
284 """
285 interval = cfg.crawl_sync_interval
286 sync_state = get_services().crawler_sync_state
287 if interval <= 0 or not sync_state.lock.acquire(blocking=False):
288 return
290 now = time.monotonic()
291 if now - sync_state.last_run < interval:
292 sync_state.lock.release()
293 return
295 sync_state.last_run = now
297 async def _run_sync() -> None:
298 try:
299 from lilbee.data.ingest import sync
301 await sync(quiet=True)
302 except Exception as exc:
303 log.warning("Periodic sync during crawl failed: %s", exc)
304 finally:
305 sync_state.lock.release()
307 task = asyncio.create_task(_run_sync())
308 tasks.add(task)
309 task.add_done_callback(tasks.discard)
312@dataclass
313class _FlushCounter:
314 """Tracks metadata writes pending since the last sidecar flush."""
316 pending: int = 0
319def _make_flush_page(
320 meta: dict[str, CrawlMeta],
321 written_paths: list[Path],
322 counter: _FlushCounter,
323) -> Callable[[CrawlResult], Any]:
324 """Build a per-result flush closure that batches metadata writes via ``to_thread``."""
326 def _sync_flush(result: CrawlResult) -> Path | None:
327 outcome = save._save_single_result(result, meta)
328 if outcome is None:
329 return None
330 save._update_single_metadata(meta, result.url, outcome, datetime.now(UTC).isoformat())
331 counter.pending += 1
332 if counter.pending >= METADATA_FLUSH_INTERVAL:
333 save.save_crawl_metadata(meta)
334 counter.pending = 0
335 return outcome.path
337 async def flush_page(result: CrawlResult) -> Path | None:
338 path = await asyncio.to_thread(_sync_flush, result)
339 if path is not None:
340 written_paths.append(path)
341 return path
343 return flush_page
346async def _ensure_crawler_ready(
347 on_progress: DetailedProgressCallback | None,
348 render_mode: CrawlRenderMode,
349) -> None:
350 """Reject early when the extra is missing; bootstrap Chromium on first use.
352 Runs before the Chromium bootstrap so a user without [crawler] doesn't pay
353 the ~160 MB download just to hit the same error afterward. Only browser mode
354 needs Chromium; HTTP mode skips the bootstrap entirely. The bootstrap
355 short-circuits when Chromium is already installed; any progress is forwarded
356 through ``on_progress`` so downstream UIs surface a 'setup' stage.
357 """
358 from lilbee.crawler import crawler_available
360 if not crawler_available():
361 raise bootstrap.CrawlerBackendError(
362 "Web crawling is not available. Run 'uv sync --extra crawler' to enable it."
363 )
365 if render_mode is CrawlRenderMode.BROWSER and not bootstrap.chromium_installed():
366 await bootstrap.bootstrap_chromium(on_progress=on_progress)
369async def _run_crawl(
370 url: str,
371 *,
372 depth: int | None,
373 max_pages: int | None,
374 on_progress: DetailedProgressCallback | None,
375 cancel: threading.Event | None,
376 quiet: bool,
377 include_subdomains: bool,
378 flush_page: Callable[[Any], Awaitable[Path | None]],
379 render_mode: CrawlRenderMode,
380) -> int:
381 """Run the single-URL or recursive crawl. Returns ``pages_seen``.
383 Resolves the depth ceiling here (permitting the seed-only ``0``) so a
384 ``cfg.crawl_max_depth`` of 0 routes to the single-page path instead of
385 blowing up inside the recursive resolver.
387 A resolved page limit of 1 is also a single-page crawl: crawl4ai's BFS
388 under-counts tiny ``max_pages`` (``max_pages=1`` yields 0 pages), so route
389 the "at most one page" request to the reliable single-URL fetch.
390 """
391 depth = _resolve_depth(depth, cfg.crawl_max_depth)
392 pages = _resolve_page_limit(max_pages)
393 if depth == 0 or pages == 1:
394 result = await crawl_single(
395 url, quiet=quiet, on_progress=on_progress, render_mode=render_mode
396 )
397 try:
398 await flush_page(result)
399 except OSError:
400 log.exception("Flush failed for %s", result.url)
401 if on_progress:
402 on_progress(EventType.CRAWL_PAGE, CrawlPageEvent(url=url, current=1, total=1))
403 return 1
404 results = await crawl_recursive(
405 url,
406 max_depth=depth,
407 max_pages=max_pages,
408 on_progress=on_progress,
409 cancel=cancel,
410 quiet=quiet,
411 include_subdomains=include_subdomains,
412 on_result=flush_page,
413 render_mode=render_mode,
414 )
415 return len(results)
418async def crawl_and_save(
419 url: str,
420 *,
421 depth: int | None = None,
422 max_pages: int | None = None,
423 on_progress: DetailedProgressCallback | None = None,
424 cancel: threading.Event | None = None,
425 quiet: bool = False,
426 include_subdomains: bool = False,
427 render_mode: CrawlRenderMode | None = None,
428) -> list[Path]:
429 """Crawl URL(s), save as markdown, update metadata. Returns paths written.
431 ``depth``: ``None`` = whole-site unbounded recursion (default). ``0`` =
432 single URL, no recursion. ``N > 0`` = max link-follow depth. ``max_pages``:
433 ``None`` (unspecified) defers to ``cfg.crawl_max_pages``, else the protective
434 ``cfg.crawl_safety_max_pages`` cap. ``0`` (``CRAWL_PAGES_UNLIMITED``) is the
435 only truly unbounded value; a positive int is honored as-is.
436 ``cfg.crawl_max_depth`` acts as a ceiling applied only when ``depth`` is
437 ``None``.
439 ``render_mode``: ``None`` resolves to ``cfg.crawl_render_mode`` (the single
440 write-boundary for the default). ``http`` fetches without a browser;
441 ``browser`` runs a tuned Chromium with JavaScript enabled.
443 Hash-based change detection: always fetches but only saves changed or new
444 files. Pages flush to disk as they stream so a cancelled crawl preserves
445 the pages already fetched.
446 """
447 mode = render_mode if render_mode is not None else cfg.crawl_render_mode
448 await _ensure_crawler_ready(on_progress, mode)
450 sem = _get_crawl_semaphore()
451 if sem is not None:
452 await sem.acquire()
453 tasks: set[asyncio.Task[None]] = set()
454 try:
455 if on_progress:
456 start_depth = depth if depth is not None else 0
457 on_progress(EventType.CRAWL_START, CrawlStartEvent(url=url, depth=start_depth))
459 meta = save.load_crawl_metadata()
460 written_paths: list[Path] = []
461 counter = _FlushCounter()
462 flush_page = _make_flush_page(meta, written_paths, counter)
464 pages_seen = await _run_crawl(
465 url,
466 depth=depth,
467 max_pages=max_pages,
468 on_progress=on_progress,
469 cancel=cancel,
470 quiet=quiet,
471 include_subdomains=include_subdomains,
472 flush_page=flush_page,
473 render_mode=mode,
474 )
476 if counter.pending > 0:
477 try:
478 save.save_crawl_metadata(meta)
479 except OSError:
480 log.exception("Final metadata flush failed")
482 cancelled = cancel is not None and cancel.is_set()
483 if not cancelled:
484 await _maybe_periodic_sync(tasks)
486 if on_progress:
487 on_progress(
488 EventType.CRAWL_DONE,
489 CrawlDoneEvent(pages_crawled=pages_seen, files_written=len(written_paths)),
490 )
492 return written_paths
493 finally:
494 # Drain this call's periodic-sync tasks before returning so
495 # asyncio.run() doesn't close the loop with a pending sync.
496 if tasks:
497 await asyncio.gather(*tasks, return_exceptions=True)
498 if sem is not None:
499 sem.release()