Coverage for src/lilbee/crawler/runner.py: 100%
176 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +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 time
13from collections.abc import Awaitable, Callable
14from pathlib import Path
15from typing import Any
17from lilbee.app.ingest import remove_sources_and_forget_in_wiki
18from lilbee.app.services import get_services
19from lilbee.core.config import cfg
20from lilbee.core.config.enums import CrawlRenderMode
21from lilbee.core.crawl_patterns import exclude_patterns_refusal
22from lilbee.crawler import availability, bootstrap, save, sitemap
23from lilbee.crawler.availability import CRAWLER_MISSING_MESSAGE
24from lilbee.crawler.backends.crawlberg import fetcher_from_config
25from lilbee.crawler.bootstrap import (
26 ChromiumMissingError,
27 CrawlEngineRefusedError,
28 CrawlerBrowserError,
29)
30from lilbee.crawler.discovery import build_concurrency_spec, build_filter_spec
31from lilbee.crawler.events import (
32 _browser_warm_up,
33 _drain_page_stream,
34 _emit_page_failure,
35 _fetched_to_result,
36 _handle_crawl_teardown_error,
37 _pages_cap,
38)
39from lilbee.crawler.messages import CHROMIUM_MISSING_MESSAGE
40from lilbee.crawler.models import (
41 CRAWL_PAGES_UNLIMITED,
42 NO_LOGIN,
43 CancelToken,
44 CrawlLogin,
45 CrawlResult,
46)
47from lilbee.crawler.save import CrawlRecord
48from lilbee.crawler.url_filter import validate_crawl_url, without_userinfo
49from lilbee.runtime.progress import (
50 CrawlDoneEvent,
51 CrawlPageEvent,
52 CrawlStartEvent,
53 DetailedProgressCallback,
54 EventType,
55)
57log = logging.getLogger(__name__)
59# A crawl limited to this many pages fetches only its seed.
60_ONE_PAGE = 1
63def _get_crawl_semaphore() -> asyncio.Semaphore | None:
64 """Return the process-wide crawl semaphore, or None when unlimited."""
65 return get_services().crawler_semaphore
68def _resolve_depth(value: int | None, cfg_ceiling: int | None) -> int | None:
69 """Resolve a crawl depth to the value the dispatcher consumes.
71 Depth has its own contract, distinct from the page-count limit: ``0`` is a
72 valid "seed only / single page" depth, not "unbounded". (Page counts use
73 :func:`_resolve_page_limit`, where ``0`` means "no limit".)
75 None -> cfg_ceiling (itself may be None; ``None`` means unbounded)
76 n >= 0 -> n (0 = seed only; explicit caller intent overrides cfg)
77 n < 0 -> ValueError (use None for unbounded)
78 """
79 effective = value if value is not None else cfg_ceiling
80 if effective is None:
81 return None
82 if effective < 0:
83 raise ValueError("crawl depth must be 0 (seed only) or a positive int")
84 return effective
87def _resolve_page_limit(max_pages: int | None) -> int | None:
88 """Resolve the page bound the fetcher consumes (None means unbounded).
90 ``CRAWL_PAGES_UNLIMITED`` (0) is an explicit "no limit" and returns None.
91 ``None`` is unspecified: it falls back to ``cfg.crawl_max_pages`` if set,
92 else the protective default ``cfg.crawl_safety_max_pages`` so a hostile site
93 can't exhaust the disk on a crawl nobody bounded. A positive int is honored
94 as-is, even above the default.
95 """
96 if max_pages == CRAWL_PAGES_UNLIMITED:
97 return None
98 if max_pages is not None:
99 return max_pages
100 if cfg.crawl_max_pages is not None:
101 return cfg.crawl_max_pages
102 return cfg.crawl_safety_max_pages
105def _require_crawler_extra() -> None:
106 """Raise :class:`CrawlerBackendError` when the crawler extra is not installed."""
107 if not availability.crawler_available():
108 raise bootstrap.CrawlerBackendError(CRAWLER_MISSING_MESSAGE)
111async def _fetch_single_page(
112 url: str,
113 render_mode: CrawlRenderMode,
114 on_progress: DetailedProgressCallback | None,
115 login: CrawlLogin,
116) -> CrawlResult:
117 """One single-URL fetch; a failure to fetch the page becomes a failed result."""
118 try:
119 async with fetcher_from_config(render_mode, login) as fetcher:
120 warm_up_done = _browser_warm_up(on_progress, render_mode)
121 page = await fetcher.fetch_single(
122 url, timeout=cfg.crawl_timeout, concurrency=build_concurrency_spec()
123 )
124 if warm_up_done is not None:
125 warm_up_done()
126 return _fetched_to_result(page)
127 except (CrawlerBrowserError, CrawlEngineRefusedError):
128 raise
129 except Exception as exc:
130 shown = without_userinfo(url)
131 log.warning("Failed to crawl %s: %s", shown, exc)
132 return CrawlResult(url=shown, success=False, error=str(exc))
135async def crawl_single(
136 url: str,
137 *,
138 render_mode: CrawlRenderMode = CrawlRenderMode.BROWSER,
139 on_progress: DetailedProgressCallback | None = None,
140 login: CrawlLogin = NO_LOGIN,
141) -> CrawlResult:
142 """Fetch a single URL, sending *login* to its host.
144 ``render_mode`` defaults to ``BROWSER`` for direct callers; the public
145 entry point :func:`crawl_and_save` resolves it from ``cfg.crawl_render_mode``
146 and passes the canonical value down. ``on_progress`` receives the setup
147 bracket of a browser crawl's warm-up.
149 Raises :class:`CrawlerBackendError` if the crawler extra isn't installed.
150 When the headless Chromium shell is missing at launch, runs the bootstrap
151 once and retries.
152 """
153 validate_crawl_url(url)
154 _require_crawler_extra()
155 try:
156 return await _fetch_single_page(url, render_mode, on_progress, login)
157 except ChromiumMissingError:
158 log.warning("Chromium missing for %s; bootstrapping then retrying", without_userinfo(url))
159 await bootstrap.bootstrap_chromium(on_progress=None)
160 return await _fetch_single_page(url, render_mode, on_progress, login)
163async def crawl_recursive(
164 url: str,
165 max_depth: int | None = None,
166 max_pages: int | None = None,
167 on_progress: DetailedProgressCallback | None = None,
168 cancel: CancelToken | None = None,
169 *,
170 include_subdomains: bool = False,
171 on_result: Callable[[CrawlResult], Any] | None = None,
172 render_mode: CrawlRenderMode = CrawlRenderMode.BROWSER,
173 login: CrawlLogin = NO_LOGIN,
174) -> int:
175 """Crawl a URL recursively using BFS, streaming per-page progress; the pages it yielded.
177 The *login* goes to the seed host only, in the crawl and in the sitemap lookup.
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. The crawl keeps
195 no page after its ``on_result`` call returns.
196 """
197 validate_crawl_url(url)
198 # ``_run_crawl`` already resolved the depth ceiling (and routed a seed-only
199 # 0 to the single-page path), so the recursive path takes ``max_depth`` as
200 # given: None = unbounded, a positive int = the cap.
201 depth = max_depth
202 pages = _resolve_page_limit(max_pages)
204 # Fail fast when the ``crawler`` extra wasn't installed so SSE
205 # callers see ``event: error`` instead of a silent zero-results run.
206 _require_crawler_extra()
208 # HTTP mode needs no browser, so the guard only applies to browser-mode crawls.
209 if render_mode is CrawlRenderMode.BROWSER and not bootstrap.chromium_installed():
210 raise ChromiumMissingError(CHROMIUM_MISSING_MESSAGE)
212 # Best-effort sitemap lookup so the TUI / CLI can render a real page-count
213 # denominator instead of [n/-1]. Falls back to CRAWL_TOTAL_UNKNOWN on any
214 # failure; off the hot path so a slow/missing sitemap never blocks the crawl.
215 sitemap_total = await asyncio.to_thread(
216 sitemap._count_sitemap_urls,
217 url,
218 include_subdomains=include_subdomains,
219 respect_robots_txt=cfg.crawl_respect_robots_txt,
220 user_agent=cfg.crawl_user_agent,
221 login=login,
222 )
224 concurrency = build_concurrency_spec()
225 filters = build_filter_spec(include_subdomains=include_subdomains)
227 pages_seen = 0
228 try:
229 async with fetcher_from_config(render_mode, login) as fetcher:
230 # The explicit reference lets the stream close, and the crawl stop,
231 # the moment the drain breaks on max_pages or cancel.
232 page_stream = fetcher.fetch_recursive(
233 url,
234 depth=depth,
235 max_pages=pages,
236 timeout=cfg.crawl_timeout,
237 concurrency=concurrency,
238 filters=filters,
239 cancel=cancel,
240 )
241 try:
242 pages_seen = await _drain_page_stream(
243 page_stream,
244 on_progress=on_progress,
245 on_result=on_result,
246 sitemap_total=sitemap_total,
247 pages_cap=_pages_cap(pages),
248 cancel=cancel,
249 render_mode=render_mode,
250 )
251 finally:
252 await page_stream.aclose()
253 except (CrawlerBrowserError, CrawlEngineRefusedError):
254 raise
255 except Exception as exc:
256 _handle_crawl_teardown_error(without_userinfo(url), exc, cancel=cancel)
258 return pages_seen
261async def _maybe_periodic_sync(tasks: set[asyncio.Task[None]], cancel: CancelToken | None) -> None:
262 """Fire off a background sync if the ``crawl_sync_interval`` has elapsed.
264 Skips when periodic sync is disabled (``interval=0``) or another sync
265 is already running. The spawned task is added to ``tasks`` so the
266 caller can drain it before returning. The sync stops on the crawl's *cancel*.
267 """
268 interval = cfg.crawl_sync_interval
269 sync_state = get_services().crawler_sync_state
270 if interval <= 0 or not sync_state.lock.acquire(blocking=False):
271 return
273 now = time.monotonic()
274 if now - sync_state.last_run < interval:
275 sync_state.lock.release()
276 return
278 sync_state.last_run = now
280 async def _run_sync() -> None:
281 try:
282 from lilbee.data.ingest import sync
284 await sync(quiet=True, cancel=cancel, started_by="a crawl")
285 except Exception as exc:
286 log.warning("Periodic sync during crawl failed: %s", exc)
287 finally:
288 sync_state.lock.release()
290 task = asyncio.create_task(_run_sync())
291 tasks.add(task)
292 task.add_done_callback(tasks.discard)
295def _make_flush_page(
296 record: CrawlRecord, written_paths: list[Path]
297) -> Callable[[CrawlResult], Any]:
298 """A flush closure that saves a page at its name and records it in *record*."""
300 def _sync_flush(result: CrawlResult) -> list[Path]:
301 path = save.save_page(result, record, remove_sources_and_forget_in_wiki)
302 return [] if path is None else [path]
304 async def flush_page(result: CrawlResult) -> list[Path]:
305 paths = await asyncio.to_thread(_sync_flush, result)
306 written_paths.extend(paths)
307 return paths
309 return flush_page
312def _compact_pages(record: CrawlRecord) -> None:
313 """Fold the crawl's commits to the page table into one version; a failure only warns."""
314 try:
315 save.finish_crawl(record)
316 except Exception:
317 log.warning("Compacting the crawl page table failed", exc_info=True)
320def _require_valid_exclude_patterns() -> None:
321 """Raise :class:`CrawlEngineRefusedError` naming each exclude pattern crawlberg refuses."""
322 refusal = exclude_patterns_refusal(cfg.crawl_exclude_patterns)
323 if refusal is not None:
324 raise CrawlEngineRefusedError(refusal)
327async def _ensure_crawler_ready(
328 on_progress: DetailedProgressCallback | None,
329 render_mode: CrawlRenderMode,
330) -> None:
331 """Reject early when the extra is missing or a setting is invalid; bootstrap Chromium.
333 Runs before the Chromium bootstrap so a user without [crawler] doesn't pay
334 the ~160 MB download just to hit the same error afterward. Only browser mode
335 needs Chromium; HTTP mode skips the bootstrap entirely. The bootstrap
336 short-circuits when Chromium is already installed; any progress is forwarded
337 through ``on_progress`` so downstream UIs surface a 'setup' stage. An exclude pattern
338 crawlberg refuses fails both a single-page and a recursive crawl before any setup.
339 """
340 _require_crawler_extra()
341 await asyncio.to_thread(_require_valid_exclude_patterns)
343 if render_mode is CrawlRenderMode.BROWSER and not bootstrap.chromium_installed():
344 await bootstrap.bootstrap_chromium(on_progress=on_progress)
347async def _run_crawl(
348 url: str,
349 *,
350 depth: int | None,
351 max_pages: int | None,
352 on_progress: DetailedProgressCallback | None,
353 cancel: CancelToken | None,
354 include_subdomains: bool,
355 flush_page: Callable[[Any], Awaitable[list[Path]]],
356 render_mode: CrawlRenderMode,
357 login: CrawlLogin,
358) -> int:
359 """Run the single-URL or recursive crawl. Returns ``pages_seen``.
361 Resolves the depth ceiling here (permitting the seed-only ``0``) so a
362 ``cfg.crawl_max_depth`` of 0 routes to the single-page path instead of
363 blowing up inside the recursive resolver. A page limit of one also takes
364 the single-page path, so the crawl reads no sitemap. A crawl whose
365 ``cancel`` is already set fetches nothing and builds no engine, at any depth.
366 """
367 depth = _resolve_depth(depth, cfg.crawl_max_depth)
368 if cancel is not None and cancel.is_set():
369 return 0
370 if depth == 0 or _resolve_page_limit(max_pages) == _ONE_PAGE:
371 result = await crawl_single(
372 url, render_mode=render_mode, on_progress=on_progress, login=login
373 )
374 try:
375 await flush_page(result)
376 except OSError:
377 log.exception("Flush failed for %s", result.url)
378 if on_progress:
379 shown = without_userinfo(url)
380 on_progress(EventType.CRAWL_PAGE, CrawlPageEvent(url=shown, current=1, total=1))
381 _emit_page_failure(on_progress, result)
382 return 1
383 return await crawl_recursive(
384 url,
385 max_depth=depth,
386 max_pages=max_pages,
387 on_progress=on_progress,
388 cancel=cancel,
389 include_subdomains=include_subdomains,
390 on_result=flush_page,
391 render_mode=render_mode,
392 login=login,
393 )
396async def crawl_and_save(
397 url: str,
398 *,
399 depth: int | None = None,
400 max_pages: int | None = None,
401 on_progress: DetailedProgressCallback | None = None,
402 cancel: CancelToken | None = None,
403 include_subdomains: bool = False,
404 render_mode: CrawlRenderMode | None = None,
405 login: CrawlLogin = NO_LOGIN,
406) -> list[Path]:
407 """Crawl URL(s), save as markdown, record each saved page. Returns paths written.
409 ``login`` goes to the seed host only and is never saved or logged.
411 ``depth``: ``None`` = whole-site unbounded recursion (default). ``0`` =
412 single URL, no recursion. ``N > 0`` = max link-follow depth. ``max_pages``:
413 ``None`` (unspecified) defers to ``cfg.crawl_max_pages``, else the protective
414 ``cfg.crawl_safety_max_pages`` cap. ``0`` (``CRAWL_PAGES_UNLIMITED``) is the
415 only truly unbounded value; a positive int is honored as-is.
416 ``cfg.crawl_max_depth`` acts as a ceiling applied only when ``depth`` is
417 ``None``.
419 ``render_mode``: ``None`` resolves to ``cfg.crawl_render_mode`` (the single
420 write-boundary for the default). ``http`` fetches without a browser;
421 ``browser`` runs a tuned Chromium with JavaScript enabled.
423 Hash-based change detection: always fetches but only saves changed or new
424 files. Pages flush to disk as they stream so a cancelled crawl preserves
425 the pages already fetched.
426 """
427 mode = render_mode if render_mode is not None else cfg.crawl_render_mode
428 await _ensure_crawler_ready(on_progress, mode)
430 sem = _get_crawl_semaphore()
431 if sem is not None:
432 await sem.acquire()
433 tasks: set[asyncio.Task[None]] = set()
434 try:
435 if on_progress:
436 start_depth = depth if depth is not None else 0
437 start = CrawlStartEvent(url=without_userinfo(url), depth=start_depth)
438 on_progress(EventType.CRAWL_START, start)
440 record = await asyncio.to_thread(save.load_crawl_record, get_services().store)
441 written_paths: list[Path] = []
442 flush_page = _make_flush_page(record, written_paths)
444 pages_seen = await _run_crawl(
445 url,
446 depth=depth,
447 max_pages=max_pages,
448 on_progress=on_progress,
449 cancel=cancel,
450 include_subdomains=include_subdomains,
451 flush_page=flush_page,
452 render_mode=mode,
453 login=login,
454 )
455 await asyncio.to_thread(_compact_pages, record)
457 cancelled = cancel is not None and cancel.is_set()
458 if not cancelled:
459 await _maybe_periodic_sync(tasks, cancel)
461 if on_progress:
462 on_progress(
463 EventType.CRAWL_DONE,
464 CrawlDoneEvent(pages_crawled=pages_seen, files_written=len(written_paths)),
465 )
467 return written_paths
468 finally:
469 # Drain this call's periodic-sync tasks before returning so
470 # asyncio.run() doesn't close the loop with a pending sync.
471 if tasks:
472 await asyncio.gather(*tasks, return_exceptions=True)
473 if sem is not None:
474 sem.release()