Coverage for src/local_deep_research/content_fetcher/fetcher.py: 98%
151 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Unified Content Fetcher.
4Provides a single interface to fetch content from various sources:
5- Academic papers (arXiv, PubMed, Semantic Scholar)
6- Web pages (HTML)
7- Direct PDF links
8"""
10from typing import Any, Dict, List, Optional
11from loguru import logger
13from .url_classifier import URLClassifier, URLType
14from ..research_library.downloaders.base import ContentType
15from ..security.egress.fetch import policy_aware_validate_url
16from ..utilities.resource_utils import safe_close
18# Default maximum content length (500KB of text)
19DEFAULT_MAX_CONTENT_LENGTH = 500_000
21# URL types where HTML fallback is pointless when the specialized downloader fails
22_NO_HTML_FALLBACK = {URLType.HTML, URLType.DOI, URLType.INVALID, URLType.PDF}
25class ContentFetcher:
26 """
27 Unified content fetcher that routes to appropriate downloaders.
29 Automatically detects URL type and uses the best downloader.
30 """
32 def __init__(
33 self,
34 timeout: int = 30,
35 language: str = "English",
36 enable_js_rendering: bool = False,
37 egress_context: Any = None,
38 ):
39 """
40 Initialize the content fetcher.
42 Args:
43 timeout: Request timeout in seconds
44 language: Language for justext stoplist (passed to HTML downloader)
45 enable_js_rendering: When True, the HTML/DOI downloader falls back
46 to a headless browser (Crawl4AI/Playwright) for pages that need
47 JavaScript to render. Defaults to False because the default
48 Docker production image ships without Chromium and the fallback
49 otherwise wastes work on every fetch. In limited (mostly
50 accidental) internal benchmark comparisons between dev
51 instances that happened to have Chromium installed and routine
52 Docker runs that did not, JS rendering did not measurably
53 improve research quality, and most regular benchmark runs are
54 on Docker without Chromium anyway — so disabling by default
55 does not regress observed quality. The user-facing toggle is
56 the ``web.enable_javascript_rendering`` setting.
57 egress_context: Optional ``EgressContext`` from the active
58 policy. When present and the run's scope is
59 ``PRIVATE_ONLY``, SSRF validation permits private IPs so
60 local lab deployments (Ollama at 127.0.0.1, SearXNG on
61 192.168.x) can actually be reached without forcing the
62 operator to set SSRF_ALLOW_PRIVATE_IPS=1 globally.
63 """
64 self.timeout = timeout
65 self.language = language
66 self.enable_js_rendering = enable_js_rendering
67 self.egress_context = egress_context
68 self._downloaders: Dict[URLType, Any] = {}
70 def _get_downloader(self, url_type: URLType):
71 """Get or create the appropriate downloader for a URL type."""
72 if url_type in self._downloaders:
73 return self._downloaders[url_type]
75 downloader: Any = None
77 if url_type == URLType.ARXIV:
78 try:
79 from ..research_library.downloaders.arxiv import ArxivDownloader
81 downloader = ArxivDownloader(timeout=self.timeout)
82 except ImportError:
83 logger.warning("ArxivDownloader not available")
85 elif url_type in (URLType.PUBMED, URLType.PMC):
86 try:
87 from ..research_library.downloaders.pubmed import (
88 PubMedDownloader,
89 )
91 downloader = PubMedDownloader(timeout=self.timeout)
92 except ImportError:
93 logger.warning("PubMedDownloader not available")
95 elif url_type == URLType.SEMANTIC_SCHOLAR:
96 try:
97 from ..research_library.downloaders.semantic_scholar import (
98 SemanticScholarDownloader,
99 )
101 downloader = SemanticScholarDownloader(timeout=self.timeout)
102 except ImportError:
103 logger.warning("SemanticScholarDownloader not available")
105 elif url_type in (URLType.BIORXIV, URLType.MEDRXIV):
106 try:
107 from ..research_library.downloaders.biorxiv import (
108 BioRxivDownloader,
109 )
111 downloader = BioRxivDownloader(timeout=self.timeout)
112 except ImportError:
113 logger.warning("BioRxivDownloader not available")
115 elif url_type == URLType.PDF:
116 try:
117 from ..research_library.downloaders.direct_pdf import (
118 DirectPDFDownloader,
119 )
121 downloader = DirectPDFDownloader(timeout=self.timeout)
122 except ImportError:
123 logger.warning("DirectPDFDownloader not available")
125 elif url_type == URLType.HTML:
126 try:
127 from ..research_library.downloaders.playwright_html import (
128 AutoHTMLDownloader as HTMLDownloader,
129 )
131 downloader = HTMLDownloader(
132 timeout=self.timeout,
133 language=self.language,
134 enable_js_rendering=self.enable_js_rendering,
135 )
136 except ImportError:
137 logger.warning("HTMLDownloader not available")
139 elif url_type == URLType.DOI:
140 # DOI URLs typically redirect to publisher pages
141 # Use HTML downloader as fallback
142 try:
143 from ..research_library.downloaders.playwright_html import (
144 AutoHTMLDownloader as HTMLDownloader,
145 )
147 downloader = HTMLDownloader(
148 timeout=self.timeout,
149 language=self.language,
150 enable_js_rendering=self.enable_js_rendering,
151 )
152 except ImportError:
153 logger.warning("HTMLDownloader not available")
155 # Cache the downloader
156 if downloader:
157 self._apply_egress_policy_to_downloader(downloader)
158 self._downloaders[url_type] = downloader
160 return downloader
162 def _apply_egress_policy_to_downloader(self, downloader: Any) -> None:
163 """Relax a downloader's SafeSession to allow private IPs when the
164 active scope is PRIVATE_ONLY, mirroring ``policy_aware_validate_url``.
166 Without this, a private/lab URL that ContentFetcher already approved
167 (policy_aware_validate_url + evaluate_url both allow private hosts
168 under PRIVATE_ONLY) is then rejected by the downloader's OWN strict
169 SafeSession SSRF re-validation (allow_private_ips defaults to False),
170 breaking PRIVATE_ONLY's documented "reach your local services" use
171 case. ``SafeSession.request`` reads ``allow_private_ips`` per-request,
172 so setting it post-construction takes effect. Cloud-metadata IPs stay
173 blocked regardless (``is_ip_blocked`` always rejects them).
174 """
175 if downloader is None or self.egress_context is None:
176 return
177 # Narrow except (not bare Exception): the only things that can throw
178 # here are the policy import and the ctx.scope attribute access. The
179 # failure direction is fail-SAFE — if we don't set allow_private_ips it
180 # stays False (strict SSRF), i.e. over-restrictive, never a bypass — but
181 # catching narrowly avoids masking an unrelated bug (e.g. a refactor
182 # that breaks the import) while still never breaking a fetch on the
183 # expected misconfiguration cases.
184 try:
185 from ..security.egress.policy import EgressScope
187 scope = self.egress_context.scope
188 except (ImportError, AttributeError): # pragma: no cover - defensive
189 logger.debug(
190 "could not resolve egress scope for downloader session",
191 exc_info=True,
192 )
193 return
194 if scope == EgressScope.PRIVATE_ONLY:
195 session = getattr(downloader, "session", None)
196 if session is not None and hasattr(session, "allow_private_ips"):
197 session.allow_private_ips = True
198 # Also thread the relaxation to the JS-render browser path. The
199 # AutoHTMLDownloader lazily builds a PlaywrightHTMLDownloader
200 # child that SSRF-validates every browser request (initial nav +
201 # each redirect hop + subresources) via its own guard, reading
202 # ``allow_private_ips`` at build time — so setting it here, before
203 # the child is built, threads PRIVATE_ONLY through to the browser.
204 # Cloud-metadata IPs stay blocked regardless (validate_url always
205 # rejects them).
206 if hasattr(downloader, "allow_private_ips"):
207 downloader.allow_private_ips = True
209 def fetch(
210 self,
211 url: str,
212 max_length: Optional[int] = None,
213 prefer_text: bool = True,
214 ) -> Dict[str, Any]:
215 """
216 Fetch content from a URL.
218 Automatically detects the URL type and uses the appropriate downloader.
220 Args:
221 url: The URL to fetch content from
222 max_length: Maximum content length to return (chars). Defaults to 500KB.
223 prefer_text: If True, prefer text extraction over PDF download
225 Returns:
226 Dict with:
227 - status: "success" or "error"
228 - content: Extracted text content
229 - url: Original URL
230 - source_type: Type of source (arxiv, pubmed, html, etc.)
231 - title: Title if available
232 - error: Error message if failed
233 """
234 # Apply default max_length if not specified
235 if max_length is None:
236 max_length = DEFAULT_MAX_CONTENT_LENGTH
238 # Classify the URL
239 url_type = URLClassifier.classify(url)
240 source_name = URLClassifier.get_source_name(url_type)
242 # Reject invalid/dangerous URLs
243 if url_type == URLType.INVALID:
244 return {
245 "status": "error",
246 "url": url,
247 "source_type": source_name,
248 "error": "Invalid or unsupported URL scheme (only http/https allowed)",
249 }
251 # SSRF validation: reject private/internal IPs before reaching downloaders.
252 # Policy-aware so PRIVATE_ONLY egress scope can actually reach
253 # private hosts (lab deployments) without disabling SSRF globally.
254 if not policy_aware_validate_url(url, self.egress_context):
255 logger.warning(f"URL failed SSRF validation: {url}")
256 return {
257 "status": "error",
258 "url": url,
259 "source_type": source_name,
260 "error": "URL failed security validation (blocked by SSRF protection)",
261 }
263 # Egress policy: reject URLs that are SSRF-OK but scope-incompatible.
264 # SSRF only checks the IP class; scope enforcement is a separate axis
265 # (PRIVATE_ONLY blocks public hosts, PUBLIC_ONLY blocks private hosts).
266 # Centralized here so all callers (LangGraph fetch tool, MCP
267 # download_content, future tools) get uniform enforcement instead of
268 # each remembering to wrap the call site.
269 if self.egress_context is not None:
270 from ..security.egress.policy import evaluate_url
271 from ..security.ssrf_validator import redact_url_for_log
273 url_decision = evaluate_url(url, self.egress_context)
274 if not url_decision.allowed:
275 logger.bind(policy_audit=True).warning(
276 "fetch URL denied by egress policy",
277 # Redact: a denied URL may carry userinfo creds / API-key
278 # query params; log only scheme://host:port.
279 url=redact_url_for_log(url),
280 scope=self.egress_context.scope.value,
281 reason=url_decision.reason,
282 )
283 # Return a structured error rather than raise — callers in
284 # the fetch path already handle dict-shaped errors.
285 return {
286 "status": "error",
287 "url": url,
288 "source_type": source_name,
289 "error": (
290 f"URL refused by egress policy ({url_decision.reason})"
291 ),
292 }
294 logger.info(f"Fetching content from {url} (detected: {source_name})")
296 # Get the appropriate downloader
297 downloader = self._get_downloader(url_type)
299 if not downloader:
300 # Fall back to generic HTML downloader. This triggers when a
301 # specialized downloader (ArXiv, SemanticScholar, etc.) failed
302 # to import — playwright_html may still be available.
303 # Use _get_downloader so the instance is cached and cleaned up
304 # by close().
305 downloader = self._get_downloader(URLType.HTML)
306 if not downloader:
307 return {
308 "status": "error",
309 "url": url,
310 "source_type": source_name,
311 "error": "No suitable downloader available",
312 }
314 # Determine content type
315 content_type = ContentType.TEXT if prefer_text else ContentType.PDF
317 # Download content
318 try:
319 result = downloader.download_with_result(url, content_type)
321 # HTML fallback: when a specialized downloader fails (e.g.
322 # arXiv PDF unavailable, PubMed paywalled), try generic HTML
323 # extraction — the abstract/landing page often has useful content.
324 if not result.is_success and url_type not in _NO_HTML_FALLBACK:
325 logger.debug(
326 f"Specialized downloader failed for {url}, "
327 "trying HTML fallback"
328 )
329 html_downloader = self._get_downloader(URLType.HTML)
330 if html_downloader: 330 ↛ 339line 330 didn't jump to line 339 because the condition on line 330 was always true
331 result = html_downloader.download_with_result(
332 url, content_type
333 )
334 # Use the HTML downloader for metadata too, so we
335 # don't call the failed specialized downloader's
336 # get_metadata (which would re-fetch or return wrong data).
337 downloader = html_downloader
339 if result.is_success and result.content:
340 # Decode content — check PDF magic bytes first, then try
341 # UTF-8, and reject anything that is neither.
342 if result.content[:4] == b"%PDF":
343 from ..research_library.downloaders.base import (
344 BaseDownloader,
345 )
347 content = BaseDownloader.extract_text_from_pdf(
348 result.content
349 )
350 if not content:
351 return {
352 "status": "error",
353 "url": url,
354 "source_type": source_name,
355 "error": "Could not extract text from PDF",
356 }
357 else:
358 try:
359 content = result.content.decode("utf-8")
360 except UnicodeDecodeError:
361 return {
362 "status": "error",
363 "url": url,
364 "source_type": source_name,
365 "error": "Content is not valid UTF-8 and not a PDF",
366 }
368 # Truncate if needed
369 if max_length and len(content) > max_length:
370 content = (
371 content[:max_length] + "\n\n[... content truncated ...]"
372 )
374 # Try to get metadata
375 metadata = {}
376 if hasattr(downloader, "get_metadata"):
377 try:
378 metadata = downloader.get_metadata(url)
379 except Exception:
380 logger.debug(
381 "Failed to fetch metadata for {}",
382 url,
383 exc_info=True,
384 )
386 return {
387 "status": "success",
388 "content": content,
389 "url": url,
390 "source_type": source_name,
391 "title": metadata.get("title"),
392 "author": metadata.get("author"),
393 "published_date": metadata.get("published_date"),
394 }
396 return {
397 "status": "error",
398 "url": url,
399 "source_type": source_name,
400 "error": result.skip_reason or "Download failed",
401 }
403 except Exception as e:
404 logger.exception(f"Error fetching content from {url}")
405 return {
406 "status": "error",
407 "url": url,
408 "source_type": source_name,
409 "error": str(e),
410 }
412 def fetch_text(
413 self, url: str, max_length: Optional[int] = None
414 ) -> Optional[str]:
415 """
416 Convenience method to fetch just the text content.
418 Args:
419 url: The URL to fetch
420 max_length: Maximum content length
422 Returns:
423 Text content or None if failed
424 """
425 result = self.fetch(url, max_length=max_length, prefer_text=True)
426 if result.get("status") == "success":
427 return result.get("content")
428 return None
430 def fetch_batch(self, urls: List[str]) -> Dict[str, Optional[str]]:
431 """Fetch multiple URLs, routing each to the best downloader.
433 Specialized downloaders (arXiv, PubMed, etc.) are tried first;
434 generic HTML extraction is used as fallback. Downloaders are
435 cached by URL type, so a single Playwright browser is shared
436 across all HTML URLs.
438 Returns:
439 Dict mapping URL → extracted text (or None if failed).
440 """
441 return {url: self.fetch_text(url) for url in urls}
443 def get_url_info(self, url: str) -> Dict[str, Any]:
444 """
445 Get information about a URL without downloading.
447 Args:
448 url: The URL to analyze
450 Returns:
451 Dict with url_type, source_name, and extracted_id
452 """
453 url_type = URLClassifier.classify(url)
454 return {
455 "url": url,
456 "url_type": url_type.value,
457 "source_name": URLClassifier.get_source_name(url_type),
458 "extracted_id": URLClassifier.extract_id(url, url_type),
459 }
461 def close(self):
462 """Close all cached downloaders and their HTTP sessions."""
463 for url_type, downloader in self._downloaders.items():
464 safe_close(downloader, f"downloader-{url_type.value}")
465 self._downloaders.clear()
467 def __enter__(self):
468 return self
470 def __exit__(self, exc_type, exc_val, exc_tb):
471 self.close()
472 return False