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

1"""Backend-agnostic value types crossing the runner/fetcher seam.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from typing import TypeAlias 

7 

8from lilbee.core.config.defaults import MAX_CRAWL_DEPTH, MAX_CRAWL_PAGES 

9from lilbee.core.config.model import CONTROL_CHARACTER 

10from lilbee.runtime.cancellation import CancelSignal 

11 

12# Explicit "no page limit" for a crawl. Distinct from None, which means 

13# "unspecified, use the protective default cfg.crawl_safety_max_pages". 

14CRAWL_PAGES_UNLIMITED = 0 

15 

16AUTHORIZATION_HEADER = "Authorization" 

17COOKIE_HEADER = "Cookie" 

18# The login fields sent as raw header values; crawlberg and httpx send ASCII only. 

19_HEADER_FIELDS = ("authorization", "cookie") 

20 

21 

22def crawl_limit_error(depth: int | None, max_pages: int | None) -> str | None: 

23 """The message for a per-crawl depth or page count outside the shared bounds, else None.""" 

24 if depth is not None and not 0 <= depth <= MAX_CRAWL_DEPTH: 

25 return f"depth must be between 0 and {MAX_CRAWL_DEPTH} (omit it to crawl the whole site)" 

26 if max_pages is not None and not 0 <= max_pages <= MAX_CRAWL_PAGES: 

27 return f"max_pages must be between 0 and {MAX_CRAWL_PAGES} (0 = unlimited)" 

28 return None 

29 

30 

31@dataclass 

32class CrawlResult: 

33 """Outcome of crawling a single URL. 

34 

35 This is the high-level result surfaced to lilbee callers 

36 (CLI, MCP, HTTP, TUI). The adapter produces ``FetchedPage`` 

37 and the orchestration layer converts it to ``CrawlResult`` 

38 when returning up to the caller. 

39 """ 

40 

41 url: str 

42 markdown: str = "" 

43 success: bool = True 

44 error: str | None = None 

45 title: str = "" 

46 

47 def failure_reason(self) -> str | None: 

48 """Why this page has nothing to save, or None when it does. 

49 

50 The save layer skips exactly these pages, so the event layer reports 

51 failures from the same predicate: the two cannot drift apart. 

52 """ 

53 if not self.success: 

54 return self.error or "fetch failed" 

55 if not self.markdown.strip(): 

56 return "page produced empty markdown" 

57 return None 

58 

59 

60@dataclass 

61class FetchedPage: 

62 """Single page produced by a ``WebFetcher`` backend. 

63 

64 Distinct from :class:`CrawlResult` so the adapter surface 

65 stays narrow and neutral: just the bytes we needed out of 

66 the underlying SDK's response object. 

67 """ 

68 

69 url: str 

70 markdown: str = "" 

71 success: bool = True 

72 error: str | None = None 

73 links: list[str] = field(default_factory=list) 

74 title: str = "" 

75 

76 

77@dataclass(frozen=True) 

78class CrawlLogin: 

79 """The login one crawl sends to its seed host; no field shows in a repr. 

80 

81 An empty field is unset. Raises ``ValueError`` naming the field, never its value. 

82 """ 

83 

84 authorization: str = field(default="", repr=False) 

85 username: str = field(default="", repr=False) 

86 password: str = field(default="", repr=False) 

87 cookie: str = field(default="", repr=False) 

88 

89 def __post_init__(self) -> None: 

90 values = { 

91 "authorization": self.authorization, 

92 "username": self.username, 

93 "password": self.password, 

94 "cookie": self.cookie, 

95 } 

96 for name, value in values.items(): 

97 if CONTROL_CHARACTER.search(value): 

98 raise ValueError(f"the crawl login {name} holds a control character") 

99 for name in _HEADER_FIELDS: 

100 if not values[name].isascii(): 

101 raise ValueError( 

102 f"the crawl login {name} holds a character outside ASCII, " 

103 "which an HTTP header cannot carry" 

104 ) 

105 if self.password and not self.username: 

106 raise ValueError("the crawl login password needs a username") 

107 if self.authorization and self.username: 

108 raise ValueError("give the crawl login an authorization header or a username, not both") 

109 

110 @property 

111 def basic_credentials(self) -> tuple[str, str] | None: 

112 """The username and password, or None when no username is set.""" 

113 return (self.username, self.password) if self.username else None 

114 

115 @property 

116 def cookie_header(self) -> dict[str, str]: 

117 """The Cookie request header, or no header when no cookie is set.""" 

118 return {COOKIE_HEADER: self.cookie} if self.cookie else {} 

119 

120 @property 

121 def authorization_header(self) -> dict[str, str]: 

122 """The Authorization request header, or no header when none is set.""" 

123 return {AUTHORIZATION_HEADER: self.authorization} if self.authorization else {} 

124 

125 

126NO_LOGIN = CrawlLogin() 

127 

128 

129@dataclass 

130class ConcurrencySpec: 

131 """Backend-agnostic concurrency, pacing and retry settings for one crawl.""" 

132 

133 semaphore_count: int = 1 

134 mean_delay: float = 0.0 

135 max_delay_range: float = 0.0 

136 retry_on_rate_limit: bool = False 

137 retry_base_delay_min: float = 0.0 

138 retry_base_delay_max: float = 0.0 

139 retry_max_backoff: float = 0.0 

140 retry_max_attempts: int = 0 

141 

142 

143@dataclass 

144class FilterSpec: 

145 """Backend-agnostic filter settings applied to discovered links. 

146 

147 Pure Python data; each adapter decides how to plug the settings 

148 into its own filter pipeline. 

149 """ 

150 

151 exclude_patterns: list[str] = field(default_factory=list) 

152 include_subdomains: bool = False 

153 

154 

155CancelToken: TypeAlias = CancelSignal 

156"""Cancellation handle the orchestration layer passes to a fetcher. 

157 

158A token whose ``is_set()`` is True means "stop as soon as you can". A 

159fetcher checks it between pages. 

160"""