Coverage for src/local_deep_research/content_fetcher/fetcher.py: 95%
226 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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 ..security.ssrf_validator import redact_url_for_log
17from ..utilities.resource_utils import safe_close
19# Default maximum content length (500KB of text)
20DEFAULT_MAX_CONTENT_LENGTH = 500_000
22# URL types whose selected strategy owns the terminal outcome when its
23# download fails.
24_NO_HTML_FALLBACK = {
25 URLType.ARXIV,
26 URLType.HTML,
27 URLType.DOI,
28 URLType.INVALID,
29 URLType.PDF,
30}
32# URL types that stay terminal even when the specialized downloader could not
33# be imported at all. Only arXiv qualifies, and only because URLType.ARXIV is
34# now reserved for URLs that name one paper (see URLClassifier.classify): for
35# those the downloader owns the canonical HTML, PDF, and API sources, and the
36# input URL (an ar5iv mirror included) is never fetched directly, so generic
37# extraction of that URL is not a valid substitute. Other arXiv-host pages --
38# listings, author pages, info./blog./static. subdomains, the legacy /ftp/
39# PDF layout -- have no canonical source and classify as HTML or PDF, so they
40# keep the generic path. Every other type keeps the import-failure fallback.
41_NO_IMPORT_FALLBACK = {URLType.ARXIV}
44class ContentFetcher:
45 """
46 Unified content fetcher that routes to appropriate downloaders.
48 Automatically detects URL type and uses the best downloader.
49 """
51 def __init__(
52 self,
53 timeout: int = 30,
54 language: str = "English",
55 enable_js_rendering: bool = False,
56 egress_context: Any = None,
57 content_cache: Any = None,
58 url_guard: Any = None,
59 ):
60 """
61 Initialize the content fetcher.
63 Args:
64 timeout: Request timeout in seconds
65 language: Language for justext stoplist (passed to HTML downloader)
66 enable_js_rendering: When True, the HTML/DOI downloader falls back
67 to a headless browser (Crawl4AI/Playwright) for pages that need
68 JavaScript to render. Defaults to False because the default
69 Docker production image ships without Chromium and the fallback
70 otherwise wastes work on every fetch. In limited (mostly
71 accidental) internal benchmark comparisons between dev
72 instances that happened to have Chromium installed and routine
73 Docker runs that did not, JS rendering did not measurably
74 improve research quality, and most regular benchmark runs are
75 on Docker without Chromium anyway — so disabling by default
76 does not regress observed quality. The user-facing toggle is
77 the ``web.enable_javascript_rendering`` setting.
78 egress_context: Optional ``EgressContext`` from the active
79 policy. When present and the run's scope is
80 ``PRIVATE_ONLY``, SSRF validation permits private IPs so
81 local lab deployments (Ollama at 127.0.0.1, SearXNG on
82 192.168.x) can actually be reached without forcing the
83 operator to set SSRF_ALLOW_PRIVATE_IPS=1 globally.
84 content_cache: Optional shared ``ResearchContentCache`` for the
85 whole research run. When None, the cache bound to the
86 current research (if any) is used, so the quick-report
87 phase, detailed subsections, lead agent and subagents
88 share previously fetched pages instead of re-fetching.
89 url_guard: Optional extra URL boundary checked before each
90 static request, including redirects. Guarded documents use
91 static extraction without browser or source enrichment.
92 They never read or store page content in the shared cache;
93 only its per-run SSRF validation memo is shared.
94 """
95 self.timeout = timeout
96 self.language = language
97 self.enable_js_rendering = enable_js_rendering
98 self.egress_context = egress_context
99 self.content_cache = content_cache
100 self.url_guard = url_guard
101 self._downloaders: Dict[URLType, Any] = {}
103 def _resolve_cache(self) -> Any:
104 """Return the effective shared cache (explicit or research-bound)."""
105 if self.content_cache is not None:
106 return self.content_cache
107 try:
108 from ..utilities.research_content_cache import get_current_cache
109 except Exception: # pragma: no cover - defensive
110 return None
111 try:
112 return get_current_cache()
113 except Exception: # pragma: no cover - defensive
114 return None
116 def _get_downloader(self, url_type: URLType):
117 """Get or create the appropriate downloader for a URL type."""
118 if self.url_guard is not None:
119 # A replay URL is saved evidence, even when its embedded target
120 # resembles a specialized source. Avoid live metadata enrichment.
121 url_type = URLType.HTML
122 if url_type in self._downloaders:
123 return self._downloaders[url_type]
125 downloader: Any = None
127 if url_type == URLType.ARXIV:
128 try:
129 from ..research_library.downloaders.arxiv import ArxivDownloader
131 downloader = ArxivDownloader(timeout=self.timeout)
132 except ImportError:
133 logger.warning("content_fetcher.arxiv_downloader_unavailable")
135 elif url_type in (URLType.PUBMED, URLType.PMC):
136 try:
137 from ..research_library.downloaders.pubmed import (
138 PubMedDownloader,
139 )
141 downloader = PubMedDownloader(timeout=self.timeout)
142 except ImportError:
143 logger.warning("content_fetcher.pubmed_downloader_unavailable")
145 elif url_type == URLType.SEMANTIC_SCHOLAR:
146 try:
147 from ..research_library.downloaders.semantic_scholar import (
148 SemanticScholarDownloader,
149 )
151 downloader = SemanticScholarDownloader(timeout=self.timeout)
152 except ImportError:
153 logger.warning(
154 "content_fetcher.semantic_scholar_downloader_unavailable"
155 )
157 elif url_type in (URLType.BIORXIV, URLType.MEDRXIV):
158 try:
159 from ..research_library.downloaders.biorxiv import (
160 BioRxivDownloader,
161 )
163 downloader = BioRxivDownloader(timeout=self.timeout)
164 except ImportError:
165 logger.warning("content_fetcher.biorxiv_downloader_unavailable")
167 elif url_type == URLType.PDF:
168 try:
169 from ..research_library.downloaders.direct_pdf import (
170 DirectPDFDownloader,
171 )
173 downloader = DirectPDFDownloader(timeout=self.timeout)
174 except ImportError:
175 logger.warning(
176 "content_fetcher.direct_pdf_downloader_unavailable"
177 )
179 elif url_type == URLType.HTML:
180 try:
181 if self.url_guard is not None:
182 # Archived evidence must stay within its capture. Static
183 # extraction avoids browser navigation and live resources.
184 from ..research_library.downloaders.html import (
185 HTMLDownloader as StaticHTMLDownloader,
186 )
188 downloader = StaticHTMLDownloader(
189 timeout=self.timeout, language=self.language
190 )
191 else:
192 from ..research_library.downloaders.playwright_html import (
193 AutoHTMLDownloader as HTMLDownloader,
194 )
196 downloader = HTMLDownloader(
197 timeout=self.timeout,
198 language=self.language,
199 enable_js_rendering=self.enable_js_rendering,
200 )
201 except ImportError:
202 logger.warning("content_fetcher.html_downloader_unavailable")
204 elif url_type == URLType.DOI:
205 # DOI URLs typically redirect to publisher pages
206 # Use HTML downloader as fallback
207 try:
208 from ..research_library.downloaders.playwright_html import (
209 AutoHTMLDownloader as HTMLDownloader,
210 )
212 downloader = HTMLDownloader(
213 timeout=self.timeout,
214 language=self.language,
215 enable_js_rendering=self.enable_js_rendering,
216 )
217 except ImportError:
218 logger.warning("content_fetcher.html_downloader_unavailable")
220 # Cache the downloader
221 if downloader:
222 self._apply_egress_policy_to_downloader(downloader)
223 self._downloaders[url_type] = downloader
225 return downloader
227 def _apply_egress_policy_to_downloader(self, downloader: Any) -> None:
228 """Carry the run policy into transport and relax private IPs only for
229 PRIVATE_ONLY, mirroring ``policy_aware_validate_url``.
231 Without this, a private/lab URL that ContentFetcher already approved
232 (policy_aware_validate_url + evaluate_url both allow private hosts
233 under PRIVATE_ONLY) is then rejected by the downloader's OWN strict
234 SafeSession SSRF re-validation (allow_private_ips defaults to False),
235 breaking PRIVATE_ONLY's documented "reach your local services" use
236 case. ``SafeSession.request`` reads ``allow_private_ips`` per-request,
237 so setting it post-construction takes effect. Cloud-metadata IPs stay
238 blocked regardless (``is_ip_blocked`` always rejects them).
239 """
240 if self.url_guard is not None and downloader is not None:
241 from ..security.safe_requests import SafeSession
243 session = getattr(downloader, "session", None)
244 if not isinstance(session, SafeSession): 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 raise ValueError(
246 "Content downloader cannot enforce the URL boundary"
247 )
248 session.url_guard = self.url_guard
249 if downloader is None or self.egress_context is None:
250 return
251 session = getattr(downloader, "session", None)
252 if session is not None:
253 session.egress_context = self.egress_context
254 # Narrow except (not bare Exception): the only things that can throw
255 # here are the policy import and the ctx.scope attribute access. The
256 # failure direction is fail-SAFE — if we don't set allow_private_ips it
257 # stays False (strict SSRF), i.e. over-restrictive, never a bypass — but
258 # catching narrowly avoids masking an unrelated bug (e.g. a refactor
259 # that breaks the import) while still never breaking a fetch on the
260 # expected misconfiguration cases.
261 try:
262 from ..security.egress.policy import EgressScope
264 scope = self.egress_context.scope
265 except (ImportError, AttributeError): # pragma: no cover - defensive
266 logger.debug("content_fetcher.egress_scope_unavailable")
267 return
268 if scope == EgressScope.PRIVATE_ONLY:
269 if session is not None and hasattr(session, "allow_private_ips"):
270 session.allow_private_ips = True
271 # The browser path is independently refused for restricted scopes;
272 # this flag is retained for the pinned static HTTP path.
273 if hasattr(downloader, "allow_private_ips"):
274 downloader.allow_private_ips = True
276 def fetch(
277 self,
278 url: str,
279 max_length: Optional[int] = None,
280 prefer_text: bool = True,
281 ) -> Dict[str, Any]:
282 """
283 Fetch content from a URL.
285 Automatically detects the URL type and uses the appropriate downloader.
287 Args:
288 url: The URL to fetch content from
289 max_length: Maximum content length to return (chars). Defaults to 500KB.
290 prefer_text: If True, prefer text extraction over PDF download
292 Returns:
293 Dict with:
294 - status: "success" or "error"
295 - content: Extracted text content
296 - url: Original URL
297 - source_type: Type of source (arxiv, pubmed, html, etc.)
298 - title: Title if available
299 - error: Error message if failed
300 """
301 # Apply default max_length if not specified
302 if max_length is None:
303 max_length = DEFAULT_MAX_CONTENT_LENGTH
305 # Classify the URL
306 url_type = URLClassifier.classify(url)
307 source_name = URLClassifier.get_source_name(url_type)
309 # Reject invalid/dangerous URLs
310 if url_type == URLType.INVALID:
311 return {
312 "status": "error",
313 "url": url,
314 "source_type": source_name,
315 "error": "Invalid or unsupported URL scheme (only http/https allowed)",
316 }
318 # SSRF validation: reject private/internal IPs before reaching downloaders.
319 # Policy-aware so PRIVATE_ONLY egress scope can actually reach
320 # private hosts (lab deployments) without disabling SSRF globally.
321 # Per-run memo in the shared cache removes the per-fetch getaddrinfo
322 # when the fetch-tool gate already validated the same (url, grant):
323 # an uncached fetch otherwise resolves 3x (tool read gate, here,
324 # tool put gate). Only True is pinned — a False (including a
325 # transient DNS failure, which validate_url maps to False rather
326 # than raising) is recomputed so recovery stays open.
327 def _ssrf_ok() -> bool:
328 _cache_for_gate = self._resolve_cache()
329 _memo = getattr(_cache_for_gate, "validated", None)
330 if callable(_memo):
331 try:
332 from ..security.egress.policy import EgressScope
333 except Exception: # pragma: no cover - defensive
334 EgressScope = None # type: ignore[assignment, misc]
335 _allow_private = bool(
336 EgressScope is not None
337 and self.egress_context is not None
338 and getattr(self.egress_context, "scope", None)
339 == EgressScope.PRIVATE_ONLY
340 )
341 # Mirror the exact booleans policy_aware_validate_url
342 # computes with: (True, True) under PRIVATE_ONLY, strict
343 # (False, False) otherwise.
344 _block_link_local = bool(_allow_private)
345 try:
346 return bool(
347 _memo(
348 url,
349 _allow_private,
350 _block_link_local,
351 lambda: policy_aware_validate_url(
352 url, self.egress_context
353 ),
354 )
355 )
356 except Exception: # pragma: no cover - defensive
357 return False
358 try:
359 return bool(policy_aware_validate_url(url, self.egress_context))
360 except Exception: # pragma: no cover - defensive
361 return False
363 if not _ssrf_ok():
364 logger.warning(
365 "content_fetcher.ssrf_denied url={url}",
366 url=redact_url_for_log(url),
367 )
368 return {
369 "status": "error",
370 "url": url,
371 "source_type": source_name,
372 "error": "URL failed security validation (blocked by SSRF protection)",
373 }
375 # Egress policy: reject URLs that are SSRF-OK but scope-incompatible.
376 # SSRF only checks the IP class; scope enforcement is a separate axis
377 # (PRIVATE_ONLY blocks public hosts, PUBLIC_ONLY blocks private hosts).
378 # Centralized here so all callers (LangGraph fetch tool, MCP
379 # download_content, future tools) get uniform enforcement instead of
380 # each remembering to wrap the call site.
381 if self.egress_context is not None:
382 from ..security.egress.policy import evaluate_url
384 url_decision = evaluate_url(url, self.egress_context)
385 if not url_decision.allowed:
386 logger.bind(policy_audit=True).warning(
387 "content_fetcher.egress_denied url={url} scope={scope} reason={reason}",
388 url=redact_url_for_log(url),
389 scope=self.egress_context.scope.value,
390 reason=url_decision.reason,
391 )
392 # Return a structured error rather than raise — callers in
393 # the fetch path already handle dict-shaped errors.
394 return {
395 "status": "error",
396 "url": url,
397 "source_type": source_name,
398 "error": (
399 f"URL refused by egress policy ({url_decision.reason})"
400 ),
401 }
403 logger.info(
404 "content_fetcher.fetch_started url={url} source_type={source_type}",
405 url=redact_url_for_log(url),
406 source_type=source_name,
407 )
409 # Shared cache with request coalescing: repeat URLs (quick phase vs
410 # subagents, lead vs subagents, web_search vs fetch_content) are
411 # served from the single research-scoped cache without a second
412 # download, and concurrent same-URL fetches collapse to one network
413 # call via ``get_or_fetch``. Only successful (full pre-truncation)
414 # text is cached, so transient errors still retry. The cache sits
415 # AFTER the egress gate above — no policy bypass.
416 # A guarded (archive) read neither reads nor stores cached page
417 # content; only the SSRF gate above shares the run cache's
418 # validation memo. An entry under the same replay URL can come from
419 # an unguarded fetch whose redirects left the capture, and the
420 # guarded caller needs this read's own transport failure, not a
421 # cached or coalesced result.
422 _cache = None if self.url_guard is not None else self._resolve_cache()
423 content_type = ContentType.TEXT if prefer_text else ContentType.PDF
425 def _download_once() -> Dict[str, Any] | None:
426 """Download and decode *url* once (no cache access).
428 Returns a cache-shaped success dict with the FULL pre-truncation
429 text on success, or an error-shaped dict (no ``content``) on
430 failure so ``get_or_fetch`` passes it through without caching
431 while waiters retry with their own fetch. A guarded read
432 re-raises download failures instead.
433 """
434 downloader = self._get_downloader(url_type)
435 if not downloader and url_type not in _NO_IMPORT_FALLBACK:
436 downloader = self._get_downloader(URLType.HTML)
437 if not downloader:
438 return {
439 "status": "error",
440 "url": url,
441 "source_type": source_name,
442 "error": "No suitable downloader available",
443 }
444 try:
445 result = self._download_content(downloader, url, content_type)
446 if (
447 not result.is_success
448 and self.url_guard is None
449 and url_type not in _NO_HTML_FALLBACK
450 ):
451 logger.debug(
452 "content_fetcher.html_fallback_started url={url}",
453 url=redact_url_for_log(url),
454 )
455 html_downloader = self._get_downloader(URLType.HTML)
456 if html_downloader: 456 ↛ 461line 456 didn't jump to line 461 because the condition on line 456 was always true
457 result = html_downloader.download_with_result(
458 url, content_type
459 )
460 downloader = html_downloader
461 if result.is_success and result.content:
462 if result.content[:4] == b"%PDF":
463 from ..research_library.downloaders.base import (
464 BaseDownloader,
465 )
467 content = BaseDownloader.extract_text_from_pdf(
468 result.content
469 )
470 if not content:
471 return {
472 "status": "error",
473 "url": url,
474 "source_type": source_name,
475 "error": "Could not extract text from PDF",
476 }
477 else:
478 try:
479 content = result.content.decode("utf-8")
480 except UnicodeDecodeError:
481 return {
482 "status": "error",
483 "url": url,
484 "source_type": source_name,
485 "error": (
486 "Content is not valid UTF-8 and not a PDF"
487 ),
488 }
489 metadata: Dict[str, Any] = {}
490 if hasattr(downloader, "get_metadata"):
491 try:
492 metadata = downloader.get_metadata(url)
493 except Exception:
494 # Best-effort/optional metadata; expected to fail
495 # routinely so keep at DEBUG without exception
496 # attachment to keep raw URLs out of log sinks.
497 logger.opt(exception=False).debug(
498 "content_fetcher.metadata_failed url={url}",
499 url=redact_url_for_log(url),
500 )
501 return {
502 "content": content,
503 "content_is_complete": getattr(
504 result, "content_is_complete", True
505 )
506 is not False,
507 "title": metadata.get("title"),
508 "url": url,
509 "author": metadata.get("author"),
510 "published_date": metadata.get("published_date"),
511 }
512 return {
513 "status": "error",
514 "url": url,
515 "source_type": source_name,
516 "error": result.skip_reason or "Download failed",
517 }
518 except Exception as e:
519 if self.url_guard is not None:
520 # The guarded fetch tool classifies this failure and emits
521 # its safe message. Ordinary callers retain dict errors.
522 raise
523 logger.opt(exception=False).error(
524 "content_fetcher.download_error url={url}",
525 url=redact_url_for_log(url),
526 )
527 return {
528 "status": "error",
529 "url": url,
530 "source_type": source_name,
531 "error": str(e),
532 }
534 if _cache is None:
535 outcome = _download_once()
536 if outcome is None or not isinstance(outcome.get("content"), str):
537 if ( 537 ↛ 542line 537 didn't jump to line 542 because the condition on line 537 was always true
538 isinstance(outcome, dict)
539 and outcome.get("status") == "error"
540 ):
541 return outcome
542 return {
543 "status": "error",
544 "url": url,
545 "source_type": source_name,
546 "error": "Download failed",
547 }
548 full_content = outcome["content"] or ""
549 display = full_content
550 if max_length and len(display) > max_length:
551 display = (
552 display[:max_length] + "\n\n[... content truncated ...]"
553 )
554 return {
555 "status": "success",
556 "content": display,
557 "content_is_complete": outcome.get("content_is_complete", True),
558 "url": url,
559 "source_type": source_name,
560 "title": outcome.get("title"),
561 "author": outcome.get("author"),
562 "published_date": outcome.get("published_date"),
563 }
565 try:
566 cached = _cache.get_or_fetch(url, _download_once)
567 except Exception as e:
568 logger.opt(exception=False).error(
569 "content_fetcher.download_error url={url}",
570 url=redact_url_for_log(url),
571 )
572 return {
573 "status": "error",
574 "url": url,
575 "source_type": source_name,
576 "error": str(e),
577 }
578 if cached is None or not isinstance(cached.get("content"), str):
579 if isinstance(cached, dict) and cached.get("status") == "error": 579 ↛ 581line 579 didn't jump to line 581 because the condition on line 579 was always true
580 return cached
581 return {
582 "status": "error",
583 "url": url,
584 "source_type": source_name,
585 "error": "Download failed",
586 }
587 full_cached = cached.get("content") or ""
588 display_cached = full_cached
589 if max_length and len(display_cached) > max_length:
590 display_cached = (
591 display_cached[:max_length] + "\n\n[... content truncated ...]"
592 )
593 result_out: Dict[str, Any] = {
594 "status": "success",
595 "content": display_cached,
596 "content_is_complete": cached.get("content_is_complete", True),
597 "url": url,
598 "source_type": source_name,
599 "title": cached.get("title"),
600 }
601 # get_or_fetch returns the cached entry (content/title/url only);
602 # when this caller was the fetcher, the full download also produced
603 # author metadata — include it when available.
604 if isinstance(cached.get("author"), str): 604 ↛ 605line 604 didn't jump to line 605 because the condition on line 604 was never true
605 result_out["author"] = cached.get("author")
606 if cached.get("published_date") is not None: 606 ↛ 607line 606 didn't jump to line 607 because the condition on line 606 was never true
607 result_out["published_date"] = cached.get("published_date")
608 return result_out
610 def _download_content(self, downloader, url, content_type):
611 if self.url_guard is None:
612 return downloader.download_with_result(url, content_type)
613 # Guarded archive reads need the actual terminal failure (including
614 # Retry-After), rather than a sanitized skip reason. State is local
615 # to the call, so concurrent fetches cannot overwrite each other.
616 transport_errors: list[BaseException] = []
617 result = downloader.download_with_result(
618 url,
619 content_type,
620 on_transport_error=transport_errors.append,
621 )
622 if transport_errors:
623 raise transport_errors[-1]
624 return result
626 def fetch_text(
627 self, url: str, max_length: Optional[int] = None
628 ) -> Optional[str]:
629 """
630 Convenience method to fetch just the text content.
632 Args:
633 url: The URL to fetch
634 max_length: Maximum content length
636 Returns:
637 Text content or None if failed
638 """
639 result = self.fetch(url, max_length=max_length, prefer_text=True)
640 if result.get("status") == "success":
641 return result.get("content")
642 return None
644 def fetch_batch(self, urls: List[str]) -> Dict[str, Optional[str]]:
645 """Fetch multiple URLs, routing each to the best downloader.
647 Specialized downloaders are tried first; generic HTML extraction is
648 used only for fallback-eligible URL types. arXiv handles its own
649 canonical fallbacks. Downloaders are cached by URL type, so a single
650 Playwright browser is shared across all HTML URLs.
652 Returns:
653 Dict mapping URL → extracted text (or None if failed).
654 """
655 return {url: self.fetch_text(url) for url in urls}
657 def get_url_info(self, url: str) -> Dict[str, Any]:
658 """
659 Get information about a URL without downloading.
661 Args:
662 url: The URL to analyze
664 Returns:
665 Dict with url_type, source_name, and extracted_id
666 """
667 url_type = URLClassifier.classify(url)
668 return {
669 "url": url,
670 "url_type": url_type.value,
671 "source_name": URLClassifier.get_source_name(url_type),
672 "extracted_id": URLClassifier.extract_id(url, url_type),
673 }
675 def close(self):
676 """Close all cached downloaders and their HTTP sessions."""
677 for url_type, downloader in self._downloaders.items():
678 safe_close(downloader, f"downloader-{url_type.value}")
679 self._downloaders.clear()
681 def __enter__(self):
682 return self
684 def __exit__(self, exc_type, exc_val, exc_tb):
685 self.close()
686 return False