Coverage for src/local_deep_research/web_search_engines/engines/search_engine_tinyfish.py: 85%
117 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +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(
152 f"Failed to parse URL for display: {link[:50]}"
153 )
155 preview = {
156 "id": link or str(idx),
157 "title": result.get("title", ""),
158 "link": link,
159 "snippet": result.get("snippet", ""),
160 "displayed_link": display_link,
161 "position": result.get("position", idx + 1),
162 }
163 preview["_full_result"] = result
164 previews.append(preview)
166 self._search_results = previews
167 return previews
169 except RateLimitError:
170 raise
171 except requests.exceptions.RequestException as e:
172 # The search query rides in the request URL (params=query=...), so
173 # the requests exception string/traceback would leak it into logs.
174 # Log only a static message plus the HTTP status — never the
175 # URL-bearing error. (_scrub_error does not strip a plain query=.)
176 status_code = getattr(
177 getattr(e, "response", None), "status_code", None
178 )
179 logger.warning(
180 f"Error getting TinyFish search results (status={status_code})"
181 )
182 self._raise_request_exception_if_rate_limited(e)
183 return []
184 except Exception as e:
185 safe_msg = self._scrub_error(e)
186 logger.exception(
187 f"Unexpected error getting TinyFish search results ({type(e).__name__}): {safe_msg}"
188 )
189 return []
191 def _get_full_content(
192 self, relevant_items: List[Dict[str, Any]]
193 ) -> List[Dict[str, Any]]:
194 """
195 Fetch clean page content from TinyFish Fetch for selected results.
197 Args:
198 relevant_items: List of relevant preview dictionaries
200 Returns:
201 List of result dictionaries enriched with extracted content
202 """
203 results = [
204 {k: v for k, v in item.items() if k != "_full_result"}
205 for item in relevant_items
206 ]
207 if not self.include_full_content:
208 return results
210 url_to_result = {
211 result.get("link") or result.get("url"): result
212 for result in results
213 if result.get("link") or result.get("url")
214 }
215 urls = list(url_to_result.keys())[: self.MAX_FETCH_URLS]
216 if not urls: 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true
217 return results
219 try:
220 response = safe_post(
221 self.FETCH_URL,
222 json={"urls": urls, "format": self.fetch_format},
223 headers={**self._headers(), "Content-Type": "application/json"},
224 timeout=self.FETCH_TIMEOUT,
225 )
226 self._raise_if_rate_limit(response.status_code)
227 response.raise_for_status()
229 data = response.json()
230 for page in data.get("results", []):
231 url = page.get("url") or page.get("final_url")
232 result = url_to_result.get(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 result = url_to_result.get(page.get("final_url"))
235 if result is None: 235 ↛ 236line 235 didn't jump to line 236 because the condition on line 235 was never true
236 continue
238 content = page.get("text")
239 if content: 239 ↛ 241line 239 didn't jump to line 241 because the condition on line 239 was always true
240 result["content"] = content
241 for field in ("final_url", "description", "language"):
242 if page.get(field): 242 ↛ 241line 242 didn't jump to line 241 because the condition on line 242 was always true
243 result[field] = page[field]
244 if page.get("title") and not result.get("title"): 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 result["title"] = page["title"]
247 errors = data.get("errors", [])
248 if errors: 248 ↛ 249line 248 didn't jump to line 249 because the condition on line 248 was never true
249 logger.info(
250 f"TinyFish Fetch returned {len(errors)} per-URL errors"
251 )
253 except RateLimitError:
254 raise
255 except requests.exceptions.RequestException as e:
256 # Mirror _get_previews: avoid logging the URL-bearing exception
257 # string; record a static message plus the HTTP status only.
258 status_code = getattr(
259 getattr(e, "response", None), "status_code", None
260 )
261 logger.warning(
262 f"Error fetching TinyFish page content (status={status_code})"
263 )
264 self._raise_request_exception_if_rate_limited(e)
265 except Exception as e:
266 safe_msg = self._scrub_error(e)
267 logger.exception(
268 f"Unexpected error fetching TinyFish page content ({type(e).__name__}): {safe_msg}"
269 )
271 return results