Coverage for src/lilbee/crawler/events.py: 100%

1"""Per-page event emission, result translation, and cancel-teardown classification.""" 

2 

3from __future__ import annotations 

4 

5import inspect 

6import logging 

7import math 

8import threading 

9from collections.abc import Callable 

10from typing import Any 

11 

12from lilbee.crawler.models import CrawlResult, FetchedPage 

13from lilbee.runtime.progress import ( 

14 CrawlPageEvent, 

15 CrawlPageFailedEvent, 

16 DetailedProgressCallback, 

17 EventType, 

18) 

19 

20log = logging.getLogger(__name__) 

21 

22 

23def _emit_page_failure(on_progress: DetailedProgressCallback | None, result: CrawlResult) -> None: 

24 """Log and emit ``CRAWL_PAGE_FAILED`` for a page that yields nothing to save.""" 

25 reason = result.failure_reason() 

26 if reason is None: 

27 return 

28 log.warning("Crawled page yields no content: %s: %s", result.url, reason) 

29 if on_progress: 

30 on_progress( 

31 EventType.CRAWL_PAGE_FAILED, 

32 CrawlPageFailedEvent(url=result.url, reason=reason), 

33 ) 

34 

35 

36def _fetched_to_result(page: FetchedPage) -> CrawlResult: 

37 """Translate the fetcher's value type to the public ``CrawlResult`` shape.""" 

38 return CrawlResult( 

39 url=page.url, 

40 markdown=page.markdown, 

41 success=page.success, 

42 error=page.error, 

43 ) 

44 

45 

46def _pages_cap(pages: int | None) -> float: 

47 """Return the per-result counter ceiling for visible progress. 

48 

49 ``None`` (unbounded) maps to ``math.inf`` so the streaming loop's hard 

50 cap check is a pure numeric compare with no branching. 

51 """ 

52 return math.inf if pages is None else pages 

53 

54 

55async def _drain_page_stream( 

56 page_stream: Any, 

57 *, 

58 on_progress: DetailedProgressCallback | None, 

59 on_result: Callable[[CrawlResult], Any] | None, 

60 sitemap_total: int, 

61 pages_cap: float, 

62 cancel: threading.Event | None, 

63) -> list[CrawlResult]: 

64 """Consume a fetcher's page stream, emitting events and flushing per page. 

65 

66 Returns the accumulated ``CrawlResult`` list. The stream is closed 

67 deterministically by the caller; this helper only iterates. 

68 """ 

69 results: list[CrawlResult] = [] 

70 counter = 0 

71 

72 def _should_cancel() -> bool: 

73 return cancel is not None and cancel.is_set() 

74 

75 async for page in page_stream: 

76 if _should_cancel(): 

77 break 

78 counter += 1 

79 if on_progress: 

80 on_progress( 

81 EventType.CRAWL_PAGE, 

82 CrawlPageEvent(url=page.url, current=counter, total=sitemap_total), 

83 ) 

84 new_result = _fetched_to_result(page) 

85 results.append(new_result) 

86 _emit_page_failure(on_progress, new_result) 

87 if on_result is not None: 

88 try: 

89 rv = on_result(new_result) 

90 if inspect.isawaitable(rv): 

91 await rv 

92 except OSError: 

93 # A disk-side flush failure must not masquerade as a crawl 

94 # failure. Log and keep streaming; the caller still sees the 

95 # result in its returned list. 

96 log.exception("Flush callback failed for %s", new_result.url) 

97 # Hard cap on visible progress. The BFS may emit failed / redirected 

98 # pages that push the per-result counter past the cap even after the 

99 # strategy has stopped dispatching. Break explicitly so the 

100 # user-visible count never exceeds the number the caller asked for. 

101 if counter >= pages_cap: 

102 break 

103 return results 

104 

105 

106def _handle_crawl_teardown_error( 

107 url: str, 

108 exc: Exception, 

109 *, 

110 cancel: threading.Event | None, 

111 results: list[CrawlResult], 

112) -> None: 

113 """Classify a recursive-crawl exception: cancel-teardown vs real failure. 

114 

115 After cancel, crawl4ai may raise BrowserContext teardown errors as 

116 in-flight URLs bail. That's expected noise, not a failure worth 

117 surfacing. Otherwise, log and append a synthetic error result (only 

118 when nothing was produced so callers always see at least one entry). 

119 """ 

120 cancelled = cancel is not None and cancel.is_set() 

121 if cancelled: 

122 log.debug("Recursive crawl of %s ended during cancel teardown: %s", url, exc) 

123 return 

124 log.warning("Recursive crawl of %s failed: %s", url, exc) 

125 if not results: 

126 results.append(CrawlResult(url=url, success=False, error=str(exc)))