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

1"""Best-effort ``/sitemap.xml`` lookup used as a progress-hint denominator.""" 

2 

3from __future__ import annotations 

4 

5import re 

6from http import HTTPStatus 

7from typing import TYPE_CHECKING 

8from urllib.parse import unquote, urlparse 

9from urllib.robotparser import RobotFileParser 

10 

11from lilbee.crawler.models import NO_LOGIN, CrawlLogin 

12from lilbee.crawler.url_filter import host_in_scope, parsed_url, require_valid_crawl_url 

13from lilbee.runtime.progress import CRAWL_TOTAL_UNKNOWN 

14 

15# ada-url is in the optional ``crawler`` extra, so it is imported on call. 

16if TYPE_CHECKING: 

17 import ada_url 

18 import httpx 

19 

20# Sitemap lookups are best-effort progress hints; never block the actual crawl. 

21_SITEMAP_FETCH_TIMEOUT_SECONDS = 5.0 

22_SITEMAP_MAX_URLS = 10_000 

23_SITEMAP_URL_TAG_RE = re.compile(r"<loc>\s*([^<]+?)\s*</loc>", re.IGNORECASE) 

24_USER_AGENT_HEADER = "User-Agent" 

25_SITEMAP_PATH = "/sitemap.xml" 

26_ROBOTS_PATH = "/robots.txt" 

27_ROBOTS_ABSENT_STATUSES = frozenset({HTTPStatus.NOT_FOUND, HTTPStatus.GONE}) 

28_ANY_USER_AGENT = "*" 

29 

30 

31def _basic_credentials(seed: ada_url.URL) -> tuple[str, str] | None: 

32 """The decoded username and password of *seed*, or None when it has neither.""" 

33 if not seed.username and not seed.password: 

34 return None 

35 return unquote(seed.username), unquote(seed.password) 

36 

37 

38def _get(url: str, seed: ada_url.URL, user_agent: str, login: CrawlLogin) -> httpx.Response | None: 

39 """One GET of *url* with the seed's credentials and *login*, or None on a transport failure. 

40 

41 The seed's own username and password win over *login*'s, as in crawlberg. 

42 

43 Redirects are not followed: a 3xx could steer this best-effort request to a 

44 private or metadata host (SSRF) before the body is inspected. 

45 """ 

46 import httpx 

47 

48 try: 

49 return httpx.get( 

50 url, 

51 auth=_basic_credentials(seed) or login.basic_credentials, 

52 headers=_headers(user_agent, login), 

53 timeout=_SITEMAP_FETCH_TIMEOUT_SECONDS, 

54 follow_redirects=False, 

55 ) 

56 except (httpx.HTTPError, OSError): 

57 return None 

58 

59 

60def _headers(user_agent: str, login: CrawlLogin) -> dict[str, str] | None: 

61 """The user agent, when set, and the login's authorization and cookie headers, or None.""" 

62 agent = {_USER_AGENT_HEADER: user_agent} if user_agent else {} 

63 return {**agent, **login.authorization_header, **login.cookie_header} or None 

64 

65 

66def _is_success(resp: httpx.Response) -> bool: 

67 return HTTPStatus.OK <= resp.status_code < HTTPStatus.MULTIPLE_CHOICES 

68 

69 

70def _robots_allow_sitemap( 

71 seed: ada_url.URL, sitemap_url: str, user_agent: str, login: CrawlLogin 

72) -> bool: 

73 """True when the site's robots.txt lets *user_agent* fetch *sitemap_url*. 

74 

75 A missing robots.txt (404 or 410) allows everything. A robots.txt that cannot 

76 be read, or that answers with any other status, allows nothing. 

77 """ 

78 resp = _get(f"{seed.origin}{_ROBOTS_PATH}", seed, user_agent, login) 

79 if resp is None: 

80 return False 

81 if resp.status_code in _ROBOTS_ABSENT_STATUSES: 

82 return True 

83 if not _is_success(resp): 

84 return False 

85 rules = RobotFileParser() 

86 rules.parse(resp.text.splitlines()) 

87 return rules.can_fetch(user_agent or _ANY_USER_AGENT, sitemap_url) 

88 

89 

90def _fetch_sitemap_text( 

91 start_url: str, *, user_agent: str = "", respect_robots_txt: bool, login: CrawlLogin = NO_LOGIN 

92) -> str | None: 

93 """Return sitemap.xml body or None on any fetch/status failure. 

94 

95 A blank *user_agent* leaves the user agent to the HTTP client. With 

96 *respect_robots_txt* the sitemap is requested only when robots.txt allows it. 

97 """ 

98 parsed = parsed_url(start_url) 

99 if parsed is None: 

100 return None 

101 sitemap_url = f"{parsed.origin}{_SITEMAP_PATH}" 

102 # Validate the seed before any connection. This is only a progress hint, so 

103 # an unvalidated or redirecting sitemap simply yields an unknown total. 

104 try: 

105 require_valid_crawl_url(sitemap_url) 

106 except ValueError: 

107 return None 

108 if respect_robots_txt and not _robots_allow_sitemap(parsed, sitemap_url, user_agent, login): 

109 return None 

110 resp = _get(sitemap_url, parsed, user_agent, login) 

111 if resp is None or not _is_success(resp): 

112 return None 

113 return resp.text 

114 

115 

116def _count_sitemap_urls( 

117 start_url: str, 

118 *, 

119 include_subdomains: bool, 

120 respect_robots_txt: bool, 

121 user_agent: str = "", 

122 login: CrawlLogin = NO_LOGIN, 

123) -> int: 

124 """Best-effort count of URLs in the host's /sitemap.xml that match the crawl scope. 

125 

126 Returns ``CRAWL_TOTAL_UNKNOWN`` on any failure (missing sitemap, timeout, 

127 parse error, redirect away from the starting host). This is purely a 

128 progress-hint denominator, so correctness is not load-bearing. 

129 

130 Only fetches sitemap.xml directly at the root of the starting host; does 

131 not follow robots.txt references or nested sitemap indexes. 

132 """ 

133 host = (urlparse(start_url).hostname or "").lower() 

134 if not host: 

135 return CRAWL_TOTAL_UNKNOWN 

136 text = _fetch_sitemap_text( 

137 start_url, user_agent=user_agent, respect_robots_txt=respect_robots_txt, login=login 

138 ) 

139 if text is None: 

140 return CRAWL_TOTAL_UNKNOWN 

141 

142 count = 0 

143 for match in _SITEMAP_URL_TAG_RE.finditer(text): 

144 link_host = (urlparse(match.group(1).strip()).hostname or "").lower() 

145 if host_in_scope(link_host, host, include_subdomains=include_subdomains): 

146 count += 1 

147 if count >= _SITEMAP_MAX_URLS: 

148 break 

149 return count if count > 0 else CRAWL_TOTAL_UNKNOWN