Coverage for src/local_deep_research/web_search_engines/engines/search_engine_guardian.py: 95%
246 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1from datetime import datetime, timedelta, UTC
2from typing import Any, Dict, List, Optional, Tuple
4from langchain_core.language_models import BaseLLM
5from ...security.secure_logging import logger
7from ...utilities.search_utilities import remove_think_tags
8from ...security.safe_requests import safe_get
9from ..rate_limiting import RateLimitError
10from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
13class GuardianSearchEngine(BaseSearchEngine):
14 """Enhanced Guardian API search engine implementation with LLM query optimization"""
16 # Mark as public search engine
17 is_public = True
18 egress_sensitivity = Sensitivity.NON_SENSITIVE
19 egress_exposure = Exposure.EXPOSING
21 def __init__(
22 self,
23 max_results: int = 10,
24 api_key: Optional[str] = None,
25 from_date: Optional[str] = None,
26 to_date: Optional[str] = None,
27 section: Optional[str] = None,
28 order_by: str = "relevance",
29 llm: Optional[BaseLLM] = None,
30 max_filtered_results: Optional[int] = None,
31 optimize_queries: bool = True,
32 adaptive_search: bool = True,
33 settings_snapshot: Optional[Dict[str, Any]] = None,
34 **kwargs,
35 ):
36 """
37 Initialize The Guardian search engine with enhanced features.
39 Args:
40 max_results: Maximum number of search results
41 api_key: The Guardian API key (can also be set via LDR_SEARCH_ENGINE_WEB_GUARDIAN_API_KEY env var or in UI settings)
42 from_date: Start date for search (YYYY-MM-DD format, default 1 month ago)
43 to_date: End date for search (YYYY-MM-DD format, default today)
44 section: Filter by section (e.g., "politics", "technology", "sport")
45 order_by: Sort order ("relevance", "newest", "oldest")
46 llm: Language model for relevance filtering and query optimization
47 max_filtered_results: Maximum number of results to keep after filtering
48 optimize_queries: Whether to optimize queries using LLM
49 adaptive_search: Whether to use adaptive search (adjusting date ranges)
50 """
51 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results
52 super().__init__(
53 llm=llm,
54 max_filtered_results=max_filtered_results,
55 max_results=max_results,
56 settings_snapshot=settings_snapshot,
57 )
59 # Get API key - check params, settings, or env vars
60 self.api_key = self._resolve_api_key(
61 api_key,
62 "search.engine.web.guardian.api_key",
63 engine_name="Guardian",
64 settings_snapshot=settings_snapshot,
65 )
66 self.optimize_queries = optimize_queries
67 self.adaptive_search = adaptive_search
69 # Set date ranges if not provided
70 if not from_date:
71 # Default to one month ago
72 one_month_ago = datetime.now(UTC) - timedelta(days=30)
73 self.from_date = one_month_ago.strftime("%Y-%m-%d")
74 else:
75 self.from_date = from_date
77 if not to_date:
78 # Default to today
79 self.to_date = datetime.now(UTC).strftime("%Y-%m-%d")
80 else:
81 self.to_date = to_date
83 self.section = section
84 self.order_by = order_by
85 self._original_date_params = {
86 "from_date": self.from_date,
87 "to_date": self.to_date,
88 }
90 # API base URL
91 self.api_url = "https://content.guardianapis.com/search"
93 def _optimize_query_for_guardian(self, query: str) -> str:
94 """
95 Optimize a natural language query for Guardian search.
96 Uses LLM to transform questions into effective news search queries.
98 Args:
99 query: Natural language query
101 Returns:
102 Optimized query string for Guardian
103 """
104 # Handle extremely long queries by truncating first
105 if len(query) > 150:
106 simple_query = " ".join(query.split()[:10])
107 logger.info(
108 f"Query too long ({len(query)} chars), truncating to: {simple_query}"
109 )
110 query = simple_query
112 if not self.llm or not self.optimize_queries:
113 # Return original query if no LLM available or optimization disabled
114 return query
116 try:
117 # Prompt for query optimization
118 prompt = f"""Transform this natural language question into a very short Guardian news search query.
120Original query: "{query}"
122CRITICAL RULES:
1231. ONLY RETURN THE EXACT SEARCH QUERY - NO EXPLANATIONS, NO COMMENTS
1242. Keep it EXTREMELY BRIEF - MAXIMUM 3-4 words total
1253. Focus only on the main topic/person/event
1264. Include proper names when relevant
1275. Remove ALL unnecessary words
1286. DO NOT use Boolean operators (no AND/OR)
1297. DO NOT use quotes
131EXAMPLE CONVERSIONS:
132✓ "What's the impact of rising interest rates on UK housing market?" → "UK housing rates"
133✓ "Latest developments in the Ukraine-Russia peace negotiations" → "Ukraine Russia negotiations"
134✓ "How are tech companies responding to AI regulation?" → "tech AI regulation"
135✓ "What is Donald Trump's current political activity?" → "Trump political activity"
137Return ONLY the extremely brief search query.
138"""
140 # Get response from LLM
141 response = self.llm.invoke(prompt)
142 optimized_query = remove_think_tags(
143 str(response.content)
144 if hasattr(response, "content")
145 else str(response)
146 ).strip()
148 # Clean up the query - remove any explanations
149 lines = optimized_query.split("\n")
150 for line in lines:
151 line = line.strip()
152 if line and not line.lower().startswith(
153 ("here", "i would", "the best", "this query")
154 ):
155 optimized_query = line
156 break
158 # Remove any quotes that wrap the entire query
159 if (
160 optimized_query.startswith('"')
161 and optimized_query.endswith('"')
162 and optimized_query.count('"') == 2
163 ):
164 optimized_query = optimized_query[1:-1]
166 logger.info(f"Original query: '{query}'")
167 logger.info(f"Optimized for Guardian: '{optimized_query}'")
169 return optimized_query
171 except Exception as e:
172 safe_msg = self._scrub_error(e)
173 logger.warning(f"Error optimizing query: {safe_msg}")
174 return query # Fall back to original query on error
176 def _adapt_dates_for_query_type(self, query: str) -> None:
177 """
178 Adapt date range based on query type (historical vs current).
180 Args:
181 query: The search query
182 """
183 # Fast path - for very short queries, default to recent news
184 if len(query.split()) <= 4:
185 logger.info("Short query detected, defaulting to recent news")
186 # Default to 60 days for short queries
187 recent = (datetime.now(UTC) - timedelta(days=60)).strftime(
188 "%Y-%m-%d"
189 )
190 self.from_date = recent
191 self.order_by = "newest"
192 return
194 if not self.llm or not self.adaptive_search:
195 return
197 try:
198 prompt = f"""Is this query asking about HISTORICAL events or CURRENT events?
200Query: "{query}"
202ONE WORD ANSWER ONLY:
203- "HISTORICAL" if about past events (older than 1 year)
204- "CURRENT" if about recent events (within past year)
205- "UNCLEAR" if can't determine
207ONE WORD ONLY:"""
209 response = self.llm.invoke(prompt)
210 answer = (
211 remove_think_tags(
212 str(response.content)
213 if hasattr(response, "content")
214 else str(response)
215 )
216 .strip()
217 .upper()
218 )
220 # Reset to original parameters first
221 self.from_date = self._original_date_params["from_date"]
222 self.to_date = self._original_date_params["to_date"]
224 if "HISTORICAL" in answer:
225 # For historical queries, go back 10 years
226 logger.info(
227 "Query classified as HISTORICAL - extending search timeframe"
228 )
229 ten_years_ago = (
230 datetime.now(UTC) - timedelta(days=3650)
231 ).strftime("%Y-%m-%d")
232 self.from_date = ten_years_ago
234 elif "CURRENT" in answer:
235 # For current events, focus on recent content
236 logger.info(
237 "Query classified as CURRENT - focusing on recent content"
238 )
239 recent = (datetime.now(UTC) - timedelta(days=60)).strftime(
240 "%Y-%m-%d"
241 )
242 self.from_date = recent
243 self.order_by = "newest" # Prioritize newest for current events
245 except Exception as e:
246 safe_msg = self._scrub_error(e)
247 logger.warning(f"Error adapting dates for query type: {safe_msg}")
248 # Keep original date parameters on error
250 def _adaptive_search(self, query: str) -> Tuple[List[Dict[str, Any]], str]:
251 """
252 Perform adaptive search that progressively adjusts parameters based on results.
254 Args:
255 query: The search query
257 Returns:
258 Tuple of (list of articles, search strategy used)
259 """
260 # Try with current parameters
261 articles = self._get_all_data(query)
262 strategy = "initial"
264 # If no results or too few, try different strategies
265 if len(articles) < 3 and self.adaptive_search:
266 logger.info(
267 f"Initial search found only {len(articles)} results, trying alternative strategies"
268 )
270 # Try with expanded date range
271 original_from_date = self.from_date
272 original_order_by = self.order_by
274 # Strategy 1: Expand to 6 months
275 logger.info("Strategy 1: Expanding time range to 6 months")
276 six_months_ago = (datetime.now(UTC) - timedelta(days=180)).strftime(
277 "%Y-%m-%d"
278 )
279 self.from_date = six_months_ago
281 articles1 = self._get_all_data(query)
282 if len(articles1) > len(articles):
283 articles = articles1
284 strategy = "expanded_6mo"
286 # Strategy 2: Expand to all time and try relevance order
287 if len(articles) < 3:
288 logger.info(
289 "Strategy 2: Expanding to all time with relevance ordering"
290 )
291 self.from_date = "2000-01-01" # Effectively "all time"
292 self.order_by = "relevance"
294 articles2 = self._get_all_data(query)
295 if len(articles2) > len(articles):
296 articles = articles2
297 strategy = "all_time_relevance"
299 # Strategy 3: Try removing section constraints
300 if len(articles) < 3 and self.section:
301 logger.info("Strategy 3: Removing section constraint")
302 original_section = self.section
303 self.section = None
305 articles3 = self._get_all_data(query)
306 if len(articles3) > len(articles): 306 ↛ 311line 306 didn't jump to line 311 because the condition on line 306 was always true
307 articles = articles3
308 strategy = "no_section"
310 # Restore section setting
311 self.section = original_section
313 # Restore original settings
314 self.from_date = original_from_date
315 self.order_by = original_order_by
317 logger.info(
318 f"Adaptive search using strategy '{strategy}' found {len(articles)} results"
319 )
320 return articles, strategy
322 def _get_all_data(self, query: str) -> List[Dict[str, Any]]:
323 """
324 Get all article data from The Guardian API in a single call.
325 Always requests all fields for simplicity.
327 Args:
328 query: The search query
330 Returns:
331 List of articles with all data
332 """
333 try:
334 # Ensure query is not empty
335 if not query or query.strip() == "":
336 query = "news"
337 logger.warning("Empty query provided, using 'news' as default")
339 # Ensure query is not too long for API
340 if len(query) > 100:
341 logger.warning(
342 f"Query too long for Guardian API ({len(query)} chars), truncating"
343 )
344 query = query[:100]
346 # Always request all fields for simplicity
347 # Ensure max_results is an integer to avoid comparison errors
348 page_size = min(
349 int(self.max_results) if self.max_results is not None else 10,
350 50,
351 )
353 # Log full parameters for debugging
354 logger.info(f"Guardian API search query: '{query}'")
355 logger.info(
356 f"Guardian API date range: {self.from_date} to {self.to_date}"
357 )
359 params = {
360 "q": query,
361 "api-key": self.api_key,
362 "from-date": self.from_date,
363 "to-date": self.to_date,
364 "order-by": self.order_by,
365 "page-size": page_size, # API maximum is 50
366 "show-fields": "headline,trailText,byline,body,publication",
367 "show-tags": "keyword",
368 }
370 # Add section filter if specified
371 if self.section:
372 params["section"] = self.section
374 # Log the complete request parameters (except API key)
375 log_params = params.copy()
376 log_params["api-key"] = "REDACTED"
377 logger.info(f"Guardian API request parameters: {log_params}")
379 # Apply rate limiting before request
380 self._last_wait_time = self.rate_tracker.apply_rate_limit(
381 self.engine_type
382 )
384 # Execute the API request
385 response = safe_get(self.api_url, params=params)
386 response.raise_for_status()
388 data = response.json()
390 # Extract results from the response
391 articles = data.get("response", {}).get("results", [])
392 logger.info(f"Guardian API returned {len(articles)} articles")
394 # Format results to include all data
395 formatted_articles = []
396 for i, article in enumerate(articles):
397 if i >= self.max_results:
398 break
400 fields = article.get("fields", {})
402 # Format the article with all fields
403 result = {
404 "id": article.get("id", ""),
405 "title": fields.get(
406 "headline", article.get("webTitle", "")
407 ),
408 "link": article.get("webUrl", ""),
409 "snippet": fields.get("trailText", ""),
410 "publication_date": article.get("webPublicationDate", ""),
411 "section": article.get("sectionName", ""),
412 "author": fields.get("byline", ""),
413 "content": fields.get("body", ""),
414 "full_content": fields.get("body", ""),
415 }
417 # Extract tags/keywords
418 tags = article.get("tags", [])
419 result["keywords"] = [
420 tag.get("webTitle", "")
421 for tag in tags
422 if tag.get("type") == "keyword"
423 ]
425 formatted_articles.append(result)
427 return formatted_articles
429 except RateLimitError:
430 raise
431 except Exception as e:
432 # Chain both scrubbers: explicit-value redaction catches the
433 # known api_key, regex-based _sanitize_error_message catches
434 # unknown credential shapes (Bearer tokens, ?key= URLs) that
435 # the upstream might echo back. Use logger.warning to drop
436 # the traceback chain.
437 safe_msg = self._scrub_error(e)
438 logger.warning(
439 f"Error getting data from The Guardian API: {safe_msg}"
440 )
441 self._raise_if_rate_limit(e)
442 return []
444 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
445 """
446 Get preview information for Guardian articles with enhanced optimization.
448 Args:
449 query: The search query
451 Returns:
452 List of preview dictionaries
453 """
454 logger.info(
455 f"Getting articles from The Guardian API for query: {query}"
456 )
458 # Step 1: Optimize the query using LLM
459 optimized_query = self._optimize_query_for_guardian(query)
461 # Step 2: Adapt date parameters based on query type
462 self._adapt_dates_for_query_type(optimized_query)
464 # Step 3: Perform adaptive search
465 articles, strategy = self._adaptive_search(optimized_query)
467 # Store search metadata for debugging
468 self._search_metadata = {
469 "original_query": query,
470 "optimized_query": optimized_query,
471 "strategy": strategy,
472 "from_date": self.from_date,
473 "to_date": self.to_date,
474 "section": self.section,
475 "order_by": self.order_by,
476 }
478 # Store full articles for later use
479 self._full_articles = {a["id"]: a for a in articles}
481 # Return only preview fields for each article
482 previews = []
483 for article in articles:
484 preview = {
485 "id": article["id"],
486 "title": article["title"],
487 "link": article["link"],
488 "snippet": article["snippet"],
489 "publication_date": article["publication_date"],
490 "section": article["section"],
491 "author": article["author"],
492 "keywords": article.get("keywords", []),
493 }
494 previews.append(preview)
496 return previews
498 def _get_full_content(
499 self, relevant_items: List[Dict[str, Any]]
500 ) -> List[Dict[str, Any]]:
501 """
502 Get full content for the relevant Guardian articles.
503 Restores full content from the cached data.
505 Args:
506 relevant_items: List of relevant preview dictionaries
508 Returns:
509 List of result dictionaries with full content
510 """
511 logger.info(
512 f"Adding full content to {len(relevant_items)} relevant Guardian articles"
513 )
515 # Get full articles for relevant items
516 results = []
517 for item in relevant_items:
518 article_id = item.get("id", "")
520 # Get the full article from our cache
521 if (
522 hasattr(self, "_full_articles")
523 and article_id in self._full_articles
524 ):
525 results.append(self._full_articles[article_id])
526 else:
527 # If not found (shouldn't happen), just use the preview
528 results.append(item)
530 return results
532 def run(
533 self, query: str, research_context: Dict[str, Any] | None = None
534 ) -> List[Dict[str, Any]]:
535 """
536 Execute a search using The Guardian API with the enhanced approach.
538 Args:
539 query: The search query
540 research_context: Context from previous research to use.
542 Returns:
543 List of search results
544 """
545 logger.info("---Execute a search using The Guardian (enhanced)---")
547 # Additional safety check for None query
548 if query is None:
549 logger.error("None query passed to Guardian search engine")
550 query = "news"
552 try:
553 # Get previews with our enhanced method
554 previews = self._get_previews(query)
556 # If no results, try one more time with a simplified query
557 if not previews:
558 simple_query = " ".join(
559 [w for w in query.split() if len(w) > 3][:3]
560 )
561 logger.warning(
562 f"No Guardian articles found, trying simplified query: {simple_query}"
563 )
564 previews = self._get_previews(simple_query)
566 # If still no results, try with a very generic query as last resort
567 if not previews and "trump" in query.lower(): 567 ↛ 568line 567 didn't jump to line 568 because the condition on line 567 was never true
568 logger.warning("Trying last resort query: 'Donald Trump'")
569 previews = self._get_previews("Donald Trump")
570 elif not previews: 570 ↛ 575line 570 didn't jump to line 575 because the condition on line 570 was always true
571 logger.warning("Trying last resort query: 'news'")
572 previews = self._get_previews("news")
574 # If still no results after all attempts, return empty list
575 if not previews:
576 logger.warning(
577 "No Guardian articles found after multiple attempts"
578 )
579 return []
581 # Filter for relevance if we have an LLM
582 if ( 582 ↛ 587line 582 didn't jump to line 587 because the condition on line 582 was never true
583 self.llm
584 and hasattr(self, "max_filtered_results")
585 and self.max_filtered_results
586 ):
587 filtered_items = self._filter_for_relevance(previews, query)
588 if not filtered_items:
589 # Fall back to unfiltered results if everything was filtered out
590 logger.warning(
591 "All articles filtered out, using unfiltered results"
592 )
593 filtered_items = previews[: self.max_filtered_results]
594 else:
595 filtered_items = previews
597 # Get full content for relevant items
598 results = self._get_full_content(filtered_items)
600 # Add source information to make it clear these are from The Guardian
601 for result in results:
602 if "source" not in result:
603 result["source"] = "The Guardian"
605 # Clean up the cache after use
606 if hasattr(self, "_full_articles"):
607 del self._full_articles
609 # Restore original date parameters
610 self.from_date = self._original_date_params["from_date"]
611 self.to_date = self._original_date_params["to_date"]
613 # Log search metadata if available
614 if hasattr(self, "_search_metadata"):
615 logger.info(f"Search metadata: {self._search_metadata}")
616 del self._search_metadata
618 return results
620 except RateLimitError:
621 raise
622 except Exception as e:
623 safe_msg = self._scrub_error(e)
624 logger.warning(f"Error in Guardian search: {safe_msg}")
626 # Restore original date parameters on error
627 self.from_date = self._original_date_params["from_date"]
628 self.to_date = self._original_date_params["to_date"]
630 return []
632 def search_by_section(
633 self, section: str, max_results: Optional[int] = None
634 ) -> List[Dict[str, Any]]:
635 """
636 Search for articles in a specific section.
638 Args:
639 section: The Guardian section name (e.g., "politics", "technology")
640 max_results: Maximum number of results (defaults to self.max_results)
642 Returns:
643 List of articles in the section
644 """
645 original_section = self.section
646 original_max_results = self.max_results
648 try:
649 # Set section and max_results for this search
650 self.section = section
651 if max_results: 651 ↛ 652line 651 didn't jump to line 652 because the condition on line 651 was never true
652 self.max_results = max_results
654 # Use empty query to get all articles in the section
655 return self.run("")
657 finally:
658 # Restore original values
659 self.section = original_section
660 self.max_results = original_max_results
662 def get_recent_articles(
663 self, days: int = 7, max_results: Optional[int] = None
664 ) -> List[Dict[str, Any]]:
665 """
666 Get recent articles from The Guardian.
668 Args:
669 days: Number of days to look back
670 max_results: Maximum number of results (defaults to self.max_results)
672 Returns:
673 List of recent articles
674 """
675 original_from_date = self.from_date
676 original_order_by = self.order_by
677 original_max_results = self.max_results
679 try:
680 # Set parameters for this search
681 self.from_date = (
682 datetime.now(UTC) - timedelta(days=days)
683 ).strftime("%Y-%m-%d")
684 self.order_by = "newest"
685 if max_results: 685 ↛ 686line 685 didn't jump to line 686 because the condition on line 685 was never true
686 self.max_results = max_results
688 # Use empty query to get all recent articles
689 return self.run("")
691 finally:
692 # Restore original values
693 self.from_date = original_from_date
694 self.order_by = original_order_by
695 self.max_results = original_max_results