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 

8from collections.abc import Callable 

9from typing import Any 

10 

11from lilbee.core.config.enums import CrawlRenderMode 

12from lilbee.core.text import collapse_whitespace 

13from lilbee.crawler.models import CancelToken, CrawlResult, FetchedPage 

14from lilbee.runtime.progress import ( 

15 CrawlPageEvent, 

16 CrawlPageFailedEvent, 

17 DetailedProgressCallback, 

18 EventType, 

19 SetupDoneEvent, 

20 SetupStartEvent, 

21) 

22 

23log = logging.getLogger(__name__) 

24 

25# The setup component of a browser crawl's warm-up: the wait for the browser's first page. 

26BROWSER_SETUP_COMPONENT = "browser" 

27# The most characters of a page title lilbee keeps; a page controls its title. 

28_MAX_TITLE_CHARS = 512 

29 

30 

31def _browser_warm_up( 

32 on_progress: DetailedProgressCallback | None, render_mode: CrawlRenderMode 

33) -> Callable[[], None] | None: 

34 """Report the start of a browser crawl's warm-up; the call that reports its end, or None.""" 

35 if on_progress is None or render_mode is not CrawlRenderMode.BROWSER: 

36 return None 

37 on_progress(EventType.SETUP_START, SetupStartEvent(component=BROWSER_SETUP_COMPONENT)) 

38 done = SetupDoneEvent(component=BROWSER_SETUP_COMPONENT, success=True) 

39 return lambda: on_progress(EventType.SETUP_DONE, done) 

40 

41 

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

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

44 reason = result.failure_reason() 

45 if reason is None: 

46 return 

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

48 if on_progress: 

49 on_progress( 

50 EventType.CRAWL_PAGE_FAILED, 

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

52 ) 

53 

54 

55def _page_title(title: str) -> str: 

56 """A page's ``<title>`` on one line, cut at ``_MAX_TITLE_CHARS``.""" 

57 return collapse_whitespace(title)[:_MAX_TITLE_CHARS] 

58 

59 

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

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

62 return CrawlResult( 

63 url=page.url, 

64 markdown=page.markdown, 

65 success=page.success, 

66 error=page.error, 

67 title=_page_title(page.title), 

68 ) 

69 

70 

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

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

73 

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

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

76 """ 

77 return math.inf if pages is None else pages 

78 

79 

80async def _flush(on_result: Callable[[CrawlResult], Any] | None, result: CrawlResult) -> None: 

81 """Hand *result* to *on_result*; a disk-side flush failure is logged, not a crawl failure.""" 

82 if on_result is None: 

83 return 

84 try: 

85 rv = on_result(result) 

86 if inspect.isawaitable(rv): 

87 await rv 

88 except OSError: 

89 log.exception("Flush callback failed for %s", result.url) 

90 

91 

92async def _drain_page_stream( 

93 page_stream: Any, 

94 *, 

95 on_progress: DetailedProgressCallback | None, 

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

97 sitemap_total: int, 

98 pages_cap: float, 

99 cancel: CancelToken | None, 

100 render_mode: CrawlRenderMode, 

101) -> int: 

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

103 

104 Returns the number of pages the stream yielded; no page outlives its flush. The 

105 stream is closed deterministically by the caller; this helper only iterates. A 

106 browser crawl's wait for its first page is bracketed by setup events. 

107 """ 

108 counter = 0 

109 warm_up_done = _browser_warm_up(on_progress, render_mode) 

110 

111 def _should_cancel() -> bool: 

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

113 

114 async for page in page_stream: 

115 if warm_up_done is not None: 

116 warm_up_done() 

117 warm_up_done = None 

118 if _should_cancel(): 

119 break 

120 counter += 1 

121 if on_progress: 

122 on_progress( 

123 EventType.CRAWL_PAGE, 

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

125 ) 

126 new_result = _fetched_to_result(page) 

127 _emit_page_failure(on_progress, new_result) 

128 await _flush(on_result, new_result) 

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

130 if counter >= pages_cap: 

131 break 

132 if warm_up_done is not None: 

133 warm_up_done() 

134 return counter 

135 

136 

137def _handle_crawl_teardown_error(url: str, exc: Exception, *, cancel: CancelToken | None) -> None: 

138 """Log a recursive-crawl exception: at debug during a cancel's teardown, else a warning.""" 

139 if cancel is not None and cancel.is_set(): 

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

141 return 

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