Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/loaders/web.py: 18%

1"""Web document loader for RAG.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from pathlib import Path 

7from typing import Any 

8import urllib.parse 

9 

10from lexigram.ai.rag.chunking.types import Chunk 

11from lexigram.ai.rag.types import RAGError 

12from lexigram.contracts.security import is_safe_url_for_request 

13from lexigram.logging import ( 

14 get_logger, 

15) 

16 

17logger = get_logger(__name__) 

18 

19 

20class WebScraperLoader: 

21 """Scrape a web page and extract its text content. 

22 

23 Uses ``ResilientHTTPClient`` (from ``lexigram-ai-llm``) for HTTP and 

24 BeautifulSoup for HTML parsing. 

25 

26 Requires: beautifulsoup4 

27 Install: pip install lexigram-ai-rag[web] 

28 """ 

29 

30 def __init__( 

31 self, 

32 *, 

33 timeout: float = 30.0, 

34 follow_links: bool = False, 

35 max_links: int = 5, 

36 user_agent: str = "Lexigram-RAG/1.0", 

37 ) -> None: 

38 """Initialize web scraper loader. 

39 

40 Args: 

41 timeout: HTTP request timeout in seconds. 

42 follow_links: Whether to also scrape links found on the page. 

43 max_links: Maximum number of links to follow when 

44 ``follow_links=True``. 

45 user_agent: User-Agent header sent with requests. 

46 """ 

47 self.timeout = timeout 

48 self.follow_links = follow_links 

49 self.max_links = max_links 

50 self.user_agent = user_agent 

51 

52 async def load(self, source: str | Path) -> list[Chunk]: 

53 """Scrape a URL and return its text content as chunks. 

54 

55 Args: 

56 source: URL to scrape (must start with ``http://`` or 

57 ``https://``). 

58 

59 Returns: 

60 List of chunks (one per page when following links). 

61 

62 Raises: 

63 ImportError: If beautifulsoup4 is not installed. 

64 RAGError: If the page cannot be fetched or parsed. 

65 """ 

66 try: 

67 try: 

68 from bs4 import BeautifulSoup # type: ignore[import-not-found] 

69 except ImportError as e: 

70 msg = "WebScraperLoader requires 'beautifulsoup4'. Install with: pip install beautifulsoup4" 

71 raise ImportError(msg) from e 

72 

73 url = str(source) 

74 if not await asyncio.to_thread(is_safe_url_for_request, url): 

75 raise RAGError(f"Source URL is not publicly reachable: {url!r}") 

76 headers = {"User-Agent": self.user_agent} 

77 

78 async def _fetch_and_parse(page_url: str) -> tuple[str, list[str]]: 

79 """Return (text_content, list_of_links).""" 

80 try: 

81 import aiohttp 

82 except ImportError as _e: 

83 msg = "WebScraperLoader requires 'aiohttp'. Install with: pip install aiohttp" 

84 raise ImportError(msg) from _e 

85 async with aiohttp.ClientSession( 

86 timeout=aiohttp.ClientTimeout(total=self.timeout), 

87 ) as _session: 

88 current_url = page_url 

89 for _hop in range(6): 

90 async with _session.get( 

91 current_url, headers=headers, allow_redirects=False 

92 ) as _resp: 

93 if _resp.status in (301, 302, 303, 307, 308): 

94 location = _resp.headers.get("Location", "") 

95 if not location: 

96 raise aiohttp.ClientError( 

97 "Redirect response without Location header" 

98 ) 

99 current_url = urllib.parse.urljoin( 

100 current_url, location 

101 ) 

102 if not await asyncio.to_thread( 

103 is_safe_url_for_request, current_url 

104 ): 

105 raise RAGError( 

106 f"Redirect target is not publicly reachable: {current_url!r}" 

107 ) 

108 continue 

109 html = await _resp.text() 

110 break 

111 

112 def _parse() -> Any: 

113 soup = BeautifulSoup(html, "html.parser") 

114 for tag in soup(["script", "style", "nav", "footer"]): 

115 tag.decompose() 

116 text = soup.get_text(separator="\n", strip=True) 

117 links: list[str] = [] 

118 if self.follow_links: 

119 for a_tag in soup.find_all("a", href=True): 

120 href = str(a_tag["href"]) 

121 if href.startswith("http"): 

122 links.append(href) 

123 return text, links[: self.max_links] 

124 

125 return await asyncio.to_thread(_parse) 

126 

127 chunks: list[Chunk] = [] 

128 text, links = await _fetch_and_parse(url) 

129 chunks.append( 

130 Chunk( 

131 text=text, 

132 source=url, 

133 chunk_index=0, 

134 metadata={"source": url, "type": "web"}, 

135 ) 

136 ) 

137 

138 if self.follow_links: 

139 for link_idx, link_url in enumerate(links, start=1): 

140 if not await asyncio.to_thread(is_safe_url_for_request, link_url): 

141 logger.warning( 

142 "link_blocked_unsafe_url", 

143 url=link_url, 

144 parent=url, 

145 ) 

146 continue 

147 try: 

148 link_text, _ = await _fetch_and_parse(link_url) 

149 chunks.append( 

150 Chunk( 

151 text=link_text, 

152 source=link_url, 

153 chunk_index=link_idx, 

154 metadata={ 

155 "source": link_url, 

156 "type": "web", 

157 "parent": url, 

158 }, 

159 ) 

160 ) 

161 except Exception as e: # noqa: BLE001 — broadened intentionally; individual link failures are non-fatal 

162 logger.warning("link_fetch_failed", url=link_url, error=str(e)) 

163 

164 return chunks 

165 

166 except (ImportError, RAGError): 

167 raise 

168 except Exception as e: 

169 msg = f"Failed to scrape {source}: {e}" 

170 raise RAGError(msg) from e 

171 

172 

173__all__ = ["WebScraperLoader"]