Coverage for src/local_deep_research/web_search_engines/engines/search_engine_tinyfish.py: 85%
117 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1from typing import Any, Dict, List, Optional
2from urllib.parse import urlparse
4import requests
5from langchain_core.language_models import BaseLLM
7from ...security import safe_get, safe_post
8from ...security.secure_logging import logger
9from ..rate_limiting import RateLimitError
10from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
13class TinyFishSearchEngine(BaseSearchEngine):
14 """TinyFish Search + Fetch API search engine implementation."""
16 is_public = True
17 egress_sensitivity = Sensitivity.NON_SENSITIVE
18 egress_exposure = Exposure.EXPOSING
19 is_generic = True
21 SEARCH_URL = "https://api.search.tinyfish.ai"
22 FETCH_URL = "https://api.fetch.tinyfish.ai"
23 SEARCH_TIMEOUT = 10
24 FETCH_TIMEOUT = 150
25 MAX_QUERY_LEN = 400
26 MAX_FETCH_URLS = 10
28 def __init__(
29 self,
30 max_results: int = 10,
31 location: str = "US",
32 language: str = "en",
33 api_key: Optional[str] = None,
34 llm: Optional[BaseLLM] = None,
35 include_full_content: bool = True,
36 fetch_format: str = "markdown",
37 max_filtered_results: Optional[int] = None,
38 search_snippets_only: Optional[bool] = None,
39 settings_snapshot: Optional[Dict[str, Any]] = None,
40 **kwargs,
41 ):
42 """
43 Initialize the TinyFish search engine.
45 Args:
46 max_results: Maximum number of search results
47 location: Country code for geo-targeted search results
48 language: Language code for search results
49 api_key: TinyFish API key
50 llm: Language model for relevance filtering
51 include_full_content: Whether to fetch extracted page content
52 fetch_format: TinyFish Fetch output format: markdown, html, or json
53 max_filtered_results: Maximum results to keep after filtering
54 search_snippets_only: Whether to return only search snippets
55 settings_snapshot: Settings snapshot for thread context
56 **kwargs: Additional parameters ignored for compatibility
57 """
58 if search_snippets_only is None:
59 search_snippets_only = not include_full_content
61 super().__init__(
62 llm=llm,
63 max_filtered_results=max_filtered_results,
64 max_results=max_results,
65 include_full_content=include_full_content,
66 search_snippets_only=search_snippets_only,
67 settings_snapshot=settings_snapshot,
68 )
69 self.location = self._normalize_location(location)
70 self.language = self._normalize_language(language)
71 self.fetch_format = fetch_format
72 self.api_key = self._resolve_api_key(
73 api_key,
74 "search.engine.web.tinyfish.api_key",
75 engine_name="TinyFish",
76 settings_snapshot=settings_snapshot,
77 )
79 @staticmethod
80 def _normalize_location(location: str) -> str:
81 return (location or "US").strip().upper()
83 @staticmethod
84 def _normalize_language(language: str) -> str:
85 language_value = (language or "en").strip()
86 language_map = {
87 "english": "en",
88 "spanish": "es",
89 "french": "fr",
90 "german": "de",
91 "italian": "it",
92 "portuguese": "pt",
93 "japanese": "ja",
94 "chinese": "zh",
95 "korean": "ko",
96 }
97 return language_map.get(language_value.lower(), language_value)
99 def _headers(self) -> Dict[str, str]:
100 return {"X-API-Key": self.api_key}
102 def _raise_request_exception_if_rate_limited(
103 self, error: requests.exceptions.RequestException
104 ) -> None:
105 response = getattr(error, "response", None)
106 status_code = getattr(response, "status_code", None)
107 if status_code is not None:
108 self._raise_if_rate_limit(status_code)
110 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
111 """
112 Get preview information from TinyFish Search.
114 Args:
115 query: The search query
117 Returns:
118 List of preview dictionaries
119 """
120 logger.info("Getting search results from TinyFish")
122 try:
123 self._last_wait_time = self.rate_tracker.apply_rate_limit(
124 self.engine_type
125 )
127 response = safe_get(
128 self.SEARCH_URL,
129 params={
130 "query": query[: self.MAX_QUERY_LEN],
131 "location": self.location,
132 "language": self.language,
133 },
134 headers=self._headers(),
135 timeout=self.SEARCH_TIMEOUT,
136 )
137 self._raise_if_rate_limit(response.status_code)
138 response.raise_for_status()
140 data = response.json()
141 results = data.get("results", [])[: self.max_results]
143 previews = []
144 for idx, result in enumerate(results):
145 link = result.get("url", "")
146 display_link = result.get("site_name", "")
147 if not display_link and link:
148 try:
149 display_link = urlparse(link).netloc or ""
150 except Exception:
151 logger.debug("Failed to parse URL for display")
153 preview = {
154 "id": link or str(idx),
155 "title": result.get("title", ""),
156 "link": link,
157 "snippet": result.get("snippet", ""),
158 "displayed_link": display_link,
159 "position": result.get("position", idx + 1),
160 }
161 preview["_full_result"] = result
162 previews.append(preview)
164 self._search_results = previews
165 return previews
167 except RateLimitError:
168 raise
169 except requests.exceptions.RequestException as e:
170 # The search query rides in the request URL (params=query=...), so
171 # the requests exception string/traceback would leak it into logs.
172 # Log only a static message plus the HTTP status — never the
173 # URL-bearing error. (_scrub_error does not strip a plain query=.)
174 status_code = getattr(
175 getattr(e, "response", None), "status_code", None
176 )
177 logger.warning(
178 f"Error getting TinyFish search results (status={status_code})"
179 )
180 self._raise_request_exception_if_rate_limited(e)
181 return []
182 except Exception as e:
183 safe_msg = self._scrub_error(e)
184 logger.exception(
185 f"Unexpected error getting TinyFish search results ({type(e).__name__}): {safe_msg}"
186 )
187 return []
189 def _get_full_content(
190 self, relevant_items: List[Dict[str, Any]]
191 ) -> List[Dict[str, Any]]:
192 """
193 Fetch clean page content from TinyFish Fetch for selected results.
195 Args:
196 relevant_items: List of relevant preview dictionaries
198 Returns:
199 List of result dictionaries enriched with extracted content
200 """
201 results = [
202 {k: v for k, v in item.items() if k != "_full_result"}
203 for item in relevant_items
204 ]
205 if not self.include_full_content:
206 return results
208 url_to_result = {
209 result.get("link") or result.get("url"): result
210 for result in results
211 if result.get("link") or result.get("url")
212 }
213 urls = list(url_to_result.keys())[: self.MAX_FETCH_URLS]
214 if not urls: 214 ↛ 215line 214 didn't jump to line 215 because the condition on line 214 was never true
215 return results
217 try:
218 response = safe_post(
219 self.FETCH_URL,
220 json={"urls": urls, "format": self.fetch_format},
221 headers={**self._headers(), "Content-Type": "application/json"},
222 timeout=self.FETCH_TIMEOUT,
223 )
224 self._raise_if_rate_limit(response.status_code)
225 response.raise_for_status()
227 data = response.json()
228 for page in data.get("results", []):
229 url = page.get("url") or page.get("final_url")
230 result = url_to_result.get(url)
231 if result is None: 231 ↛ 232line 231 didn't jump to line 232 because the condition on line 231 was never true
232 result = url_to_result.get(page.get("final_url"))
233 if result is None: 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true
234 continue
236 content = page.get("text")
237 if content: 237 ↛ 239line 237 didn't jump to line 239 because the condition on line 237 was always true
238 result["content"] = content
239 for field in ("final_url", "description", "language"):
240 if page.get(field): 240 ↛ 239line 240 didn't jump to line 239 because the condition on line 240 was always true
241 result[field] = page[field]
242 if page.get("title") and not result.get("title"): 242 ↛ 243line 242 didn't jump to line 243 because the condition on line 242 was never true
243 result["title"] = page["title"]
245 errors = data.get("errors", [])
246 if errors: 246 ↛ 247line 246 didn't jump to line 247 because the condition on line 246 was never true
247 logger.info(
248 f"TinyFish Fetch returned {len(errors)} per-URL errors"
249 )
251 except RateLimitError:
252 raise
253 except requests.exceptions.RequestException as e:
254 # Mirror _get_previews: avoid logging the URL-bearing exception
255 # string; record a static message plus the HTTP status only.
256 status_code = getattr(
257 getattr(e, "response", None), "status_code", None
258 )
259 logger.warning(
260 f"Error fetching TinyFish page content (status={status_code})"
261 )
262 self._raise_request_exception_if_rate_limited(e)
263 except Exception as e:
264 safe_msg = self._scrub_error(e)
265 logger.exception(
266 f"Unexpected error fetching TinyFish page content ({type(e).__name__}): {safe_msg}"
267 )
269 return results