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