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 urllib.parse import urlparse 

8 

9from lilbee.crawler.url_filter import host_in_scope, require_valid_crawl_url 

10from lilbee.runtime.progress import CRAWL_TOTAL_UNKNOWN 

11 

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

13_SITEMAP_FETCH_TIMEOUT_SECONDS = 5.0 

14_SITEMAP_MAX_URLS = 10_000 

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

16 

17 

18def _fetch_sitemap_text(start_url: str) -> str | None: 

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

20 import httpx 

21 

22 parsed = urlparse(start_url) 

23 sitemap_url = f"{parsed.scheme}://{parsed.netloc}/sitemap.xml" 

24 # Validate the seed before any connection, and do not follow redirects: a 

25 # 3xx could otherwise steer this best-effort fetch to a private/metadata 

26 # host (SSRF) before the body is inspected. This is only a progress hint, 

27 # so a redirecting or unvalidated sitemap simply yields an unknown total. 

28 try: 

29 require_valid_crawl_url(sitemap_url) 

30 except ValueError: 

31 return None 

32 try: 

33 resp = httpx.get( 

34 sitemap_url, timeout=_SITEMAP_FETCH_TIMEOUT_SECONDS, follow_redirects=False 

35 ) 

36 except (httpx.HTTPError, OSError): 

37 return None 

38 # Accept only a direct 2xx; an unfollowed 3xx (or any error status) yields 

39 # no usable sitemap and is treated as a miss. 

40 if not (HTTPStatus.OK <= resp.status_code < HTTPStatus.MULTIPLE_CHOICES): 

41 return None 

42 return resp.text 

43 

44 

45def _count_sitemap_urls(start_url: str, *, include_subdomains: bool) -> int: 

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

47 

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

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

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

51 

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

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

54 """ 

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

56 if not host: 

57 return CRAWL_TOTAL_UNKNOWN 

58 text = _fetch_sitemap_text(start_url) 

59 if text is None: 

60 return CRAWL_TOTAL_UNKNOWN 

61 

62 count = 0 

63 for match in _SITEMAP_URL_TAG_RE.finditer(text): 

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

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

66 count += 1 

67 if count >= _SITEMAP_MAX_URLS: 

68 break 

69 return count if count > 0 else CRAWL_TOTAL_UNKNOWN