Coverage for src/local_deep_research/web_search_engines/engines/full_search.py: 98%
129 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1from datetime import datetime, UTC
2from typing import Any, Dict, List, Optional, Protocol, runtime_checkable
4from langchain_core.language_models import BaseLLM
6from ...config.search_config import QUALITY_CHECK_DDG_URLS
7from ...research_library.downloaders.extraction import (
8 batch_fetch_and_extract,
9)
10from ...security.egress.policy import PolicyDeniedError
11from ...security.log_sanitizer import scrub_error
12from ...security.secure_logging import logger
13from ...security.ssrf_validator import redact_url_for_log, validate_url
14from ...utilities.js_rendering import (
15 read_js_rendering_setting as _read_js_rendering_setting,
16)
17from ...utilities.json_utils import extract_json, get_llm_response_text
20@runtime_checkable
21class _Invokable(Protocol):
22 def invoke(self, query: str) -> Any: ... 22 ↛ exitline 22 didn't return from function 'invoke' because
25class FullSearchResults:
26 def __init__(
27 self,
28 llm: Optional[BaseLLM],
29 web_search: _Invokable,
30 output_format: str = "list",
31 language: str = "English",
32 max_results: int = 10,
33 region: str = "wt-wt",
34 time: Optional[str] = "y",
35 safesearch: str | int = "Moderate",
36 settings_snapshot: Optional[Dict] = None,
37 egress_context: Optional[Any] = None,
38 ):
39 self.llm = llm
40 self.output_format = output_format
41 self.language = language
42 self.max_results = max_results
43 self.region = region
44 self.time = time
45 self.safesearch = safesearch
46 self.web_search = web_search
47 self.settings_snapshot = settings_snapshot
48 # Set by the factory when the parent engine is gated against a
49 # specific scope; used to evaluate per-URL fetches below.
50 self.egress_context = egress_context
52 def check_urls(self, results: List[Dict], query: str) -> List[Dict]:
53 if not results:
54 return results
56 now = datetime.now(UTC)
57 current_time = now.strftime("%Y-%m-%d")
58 prompt = f"""ONLY Return a JSON array. The response contains no letters. Evaluate these URLs for:
59 1. Timeliness (today: {current_time})
60 2. Factual accuracy (cross-reference major claims)
61 3. Source reliability (prefer official company websites, established news outlets)
62 4. Direct relevance to query: {query}
64 URLs to evaluate:
65 {results}
67 Return a JSON array of indices (0-based) for sources that meet ALL criteria.
68 ONLY Return a JSON array of indices (0-based) and nothing else. No letters.
69 Example response: \n[0, 2, 4]\n\n"""
71 try:
72 if self.llm is None:
73 return results
74 response = self.llm.invoke(prompt)
75 response_text = get_llm_response_text(response)
76 good_indices = extract_json(response_text, expected_type=list)
78 if good_indices is None:
79 good_indices = []
81 return [r for i, r in enumerate(results) if i in good_indices]
82 except PolicyDeniedError:
83 # The URL-quality LLM was denied by egress policy (e.g. a cloud
84 # LLM under require_local / PRIVATE_ONLY). Do NOT fall through to
85 # the unfiltered-results fallback below — that would let the run
86 # proceed despite the user's policy refusing the LLM. Fail closed.
87 raise
88 except Exception as e:
89 safe_msg = scrub_error(e)
90 logger.exception(
91 f"URL filtering error ({type(e).__name__}): {safe_msg}"
92 )
93 logger.warning(
94 "URL quality filter unavailable — returning {} unfiltered "
95 "results as fallback",
96 len(results),
97 )
98 return results # Fall back to original results on LLM error
100 def run(self, query: str):
101 # Step 1: Get search results
102 search_results = self.web_search.invoke(query)
103 if not isinstance(search_results, list):
104 raise ValueError("Expected the search results in list format.")
106 # Step 2: Filter URLs using LLM
107 if QUALITY_CHECK_DDG_URLS:
108 filtered_results = self.check_urls(search_results, query)
109 else:
110 filtered_results = search_results
112 # Extract URLs from filtered results
113 urls = [
114 result.get("link")
115 for result in filtered_results
116 if result.get("link")
117 ]
119 if not urls:
120 logger.error("\n === NO VALID LINKS ===\n")
121 return []
123 # SSRF-validate + egress-scope-validate each URL. SSRF only checks
124 # IP class; scope enforcement is the separate axis that blocks
125 # public hosts under PRIVATE_ONLY / STRICT.
126 evaluate_url_fn = None
127 egress_ctx = self.egress_context
128 if egress_ctx is not None:
129 from ...security.egress.policy import evaluate_url as _ev_url
131 evaluate_url_fn = _ev_url
132 safe_urls: List[str] = []
133 for url in urls:
134 if url is None: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true
135 continue
136 if not validate_url(url):
137 logger.warning(
138 "SSRF validation blocked URL from full content fetch: "
139 f"{redact_url_for_log(url)}."
140 )
141 continue
142 if evaluate_url_fn is not None and egress_ctx is not None:
143 url_decision = evaluate_url_fn(url, egress_ctx)
144 if not url_decision.allowed:
145 logger.bind(policy_audit=True).warning(
146 "full_search URL denied by egress policy",
147 url=redact_url_for_log(url),
148 scope=egress_ctx.scope.value,
149 reason=url_decision.reason,
150 )
151 continue
152 safe_urls.append(url)
154 if not safe_urls:
155 logger.warning(
156 "All URLs were blocked by SSRF validation — returning results "
157 "without full content. This can happen when search results "
158 "point to internal/private network addresses."
159 )
160 for result in filtered_results:
161 result["full_content"] = None
162 return filtered_results
164 # Fetch and extract all pages — specialized downloaders (arXiv,
165 # PubMed, etc.) are tried first, with HTML crawling as fallback.
166 url_to_content = batch_fetch_and_extract(
167 safe_urls,
168 language=self.language,
169 enable_js_rendering=_read_js_rendering_setting(
170 self.settings_snapshot
171 ),
172 )
174 nr_full_text = sum(1 for v in url_to_content.values() if v)
175 for result in filtered_results:
176 link = result.get("link")
177 result["full_content"] = url_to_content.get(link) if link else None
179 logger.info(f"Full search: retrieved content from {nr_full_text} pages")
180 return filtered_results
182 def _get_full_content(
183 self, relevant_items: List[Dict[str, Any]]
184 ) -> List[Dict[str, Any]]:
185 """Fetch and attach full content to an existing list of items."""
186 evaluate_url_fn = None
187 egress_ctx = self.egress_context
188 if egress_ctx is not None:
189 from ...security.egress.policy import evaluate_url as _ev_url
191 evaluate_url_fn = _ev_url
192 urls: List[str] = []
193 for item in relevant_items:
194 link = item.get("link")
195 if link is None:
196 continue
197 if not validate_url(link):
198 logger.warning(
199 "SSRF validation blocked URL from full content fetch: "
200 f"{redact_url_for_log(link)}."
201 )
202 continue
203 if evaluate_url_fn is not None and egress_ctx is not None:
204 url_decision = evaluate_url_fn(link, egress_ctx)
205 if not url_decision.allowed:
206 logger.bind(policy_audit=True).warning(
207 "full_search _get_full_content URL denied by egress policy",
208 url=redact_url_for_log(link),
209 scope=egress_ctx.scope.value,
210 reason=url_decision.reason,
211 )
212 continue
213 urls.append(link)
215 if not urls:
216 for item in relevant_items:
217 item["full_content"] = None
218 return relevant_items
220 try:
221 url_to_content = batch_fetch_and_extract(
222 urls,
223 language=self.language,
224 enable_js_rendering=_read_js_rendering_setting(
225 self.settings_snapshot
226 ),
227 )
228 except Exception as e:
229 safe_msg = scrub_error(e)
230 logger.exception(
231 f"Error fetching full content ({type(e).__name__}): {safe_msg}"
232 )
233 for item in relevant_items:
234 item["full_content"] = None
235 return relevant_items
237 for item in relevant_items:
238 link = item.get("link")
239 item["full_content"] = url_to_content.get(link) if link else None
241 return relevant_items
243 def invoke(self, query: str) -> Any:
244 return self.run(query)
246 def __call__(self, query: str) -> Any:
247 return self.invoke(query)