Coverage for src/local_deep_research/web_search_engines/engines/search_engine

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

11from ...utilities.llm_utils import invoke_llm_sync

14class GuardianSearchEngine(BaseSearchEngine):

15 """Enhanced Guardian API search engine implementation with LLM query optimization"""

17 # Mark as public search engine

18 is_public = True

19 egress_sensitivity = Sensitivity.NON_SENSITIVE

20 egress_exposure = Exposure.EXPOSING

22 def __init__(

23 self,

24 max_results: int = 10,

25 api_key: Optional[str] = None,

26 from_date: Optional[str] = None,

27 to_date: Optional[str] = None,

28 section: Optional[str] = None,

29 order_by: str = "relevance",

30 llm: Optional[BaseLLM] = None,

31 max_filtered_results: Optional[int] = None,

32 optimize_queries: bool = True,

33 adaptive_search: bool = True,

34 search_snippets_only: bool = True,

35 settings_snapshot: Optional[Dict[str, Any]] = None,

36 **kwargs,

37 ):

38 """

39 Initialize The Guardian search engine with enhanced features.

41 Args:

42 max_results: Maximum number of search results

43 api_key: The Guardian API key (can also be set via LDR_SEARCH_ENGINE_WEB_GUARDIAN_API_KEY env var or in UI settings)

44 from_date: Start date for search (YYYY-MM-DD format, default 1 month ago)

45 to_date: End date for search (YYYY-MM-DD format, default today)

46 section: Filter by section (e.g., "politics", "technology", "sport")

47 order_by: Sort order ("relevance", "newest", "oldest")

48 llm: Language model for relevance filtering and query optimization

49 max_filtered_results: Maximum number of results to keep after filtering

50 optimize_queries: Whether to optimize queries using LLM

51 adaptive_search: Whether to use adaptive search (adjusting date ranges)

52 search_snippets_only: Whether to request and return previews only

53 """

54 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results

55 super().__init__(

56 llm=llm,

57 max_filtered_results=max_filtered_results,

58 max_results=max_results,

59 search_snippets_only=search_snippets_only,

60 settings_snapshot=settings_snapshot,

61 )

63 # Get API key - check params, settings, or env vars

64 self.api_key = self._resolve_api_key(

65 api_key,

66 "search.engine.web.guardian.api_key",

67 engine_name="Guardian",

68 settings_snapshot=settings_snapshot,

69 )

70 self.optimize_queries = optimize_queries

71 self.adaptive_search = adaptive_search

73 # Set date ranges if not provided

74 if not from_date:

75 # Default to one month ago

76 one_month_ago = datetime.now(UTC) - timedelta(days=30)

77 self.from_date = one_month_ago.strftime("%Y-%m-%d")

78 else:

79 self.from_date = from_date

81 if not to_date:

82 # Default to today

83 self.to_date = datetime.now(UTC).strftime("%Y-%m-%d")

84 else:

85 self.to_date = to_date

87 self.section = section

88 self.order_by = order_by

89 self._original_date_params = {

90 "from_date": self.from_date,

91 "to_date": self.to_date,

92 }

94 # API base URL

95 self.api_url = "https://content.guardianapis.com/search"

97 def _optimize_query_for_guardian(self, query: str) -> str:

98 """

99 Optimize a natural language query for Guardian search.

100 Uses LLM to transform questions into effective news search queries.

101

102 Args:

103 query: Natural language query

104

105 Returns:

106 Optimized query string for Guardian

107 """

108 # Handle extremely long queries by truncating first

109 if len(query) > 150:

110 simple_query = " ".join(query.split()[:10])

111 logger.info(

112 f"Query too long ({len(query)} chars), truncated to "

113 f"{len(simple_query)} chars"

114 )

115 query = simple_query

116

117 if not self.llm or not self.optimize_queries:

118 # Return original query if no LLM available or optimization disabled

119 return query

120

121 try:

122 # Prompt for query optimization

123 prompt = f"""Transform this natural language question into a very short Guardian news search query.

124

125Original query: "{query}"

126

127CRITICAL RULES:

1281. ONLY RETURN THE EXACT SEARCH QUERY - NO EXPLANATIONS, NO COMMENTS

1292. Keep it EXTREMELY BRIEF - MAXIMUM 3-4 words total

1303. Focus only on the main topic/person/event

1314. Include proper names when relevant

1325. Remove ALL unnecessary words

1336. DO NOT use Boolean operators (no AND/OR)

1347. DO NOT use quotes

135

136EXAMPLE CONVERSIONS:

137✓ "What's the impact of rising interest rates on UK housing market?" → "UK housing rates"

138✓ "Latest developments in the Ukraine-Russia peace negotiations" → "Ukraine Russia negotiations"

139✓ "How are tech companies responding to AI regulation?" → "tech AI regulation"

140✓ "What is Donald Trump's current political activity?" → "Trump political activity"

141

142Return ONLY the extremely brief search query.

143"""

144

145 # Get response from LLM

146 response = invoke_llm_sync(self.llm, prompt)

147 optimized_query = remove_think_tags(

148 str(response.content)

149 if hasattr(response, "content")

150 else str(response)

151 ).strip()

152

153 # Clean up the query - remove any explanations

154 lines = optimized_query.split("\n")

155 for line in lines:

156 line = line.strip()

157 if line and not line.lower().startswith(

158 ("here", "i would", "the best", "this query")

159 ):

160 optimized_query = line

161 break

162

163 # Remove any quotes that wrap the entire query

164 if (

165 optimized_query.startswith('"')

166 and optimized_query.endswith('"')

167 and optimized_query.count('"') == 2

168 ):

169 optimized_query = optimized_query[1:-1]

170

171 logger.info(

172 f"Optimized query for Guardian "

173 f"({len(query)} -> {len(optimized_query)} chars)"

174 )

175

176 return optimized_query

177

178 except Exception as e:

179 safe_msg = self._scrub_error(e)

180 logger.warning(f"Error optimizing query: {safe_msg}")

181 return query # Fall back to original query on error

182

183 def _adapt_dates_for_query_type(self, query: str) -> None:

184 """

185 Adapt date range based on query type (historical vs current).

186

187 Args:

188 query: The search query

189 """

190 # Fast path - for very short queries, default to recent news

191 if len(query.split()) <= 4:

192 logger.info("Short query detected, defaulting to recent news")

193 # Default to 60 days for short queries

194 recent = (datetime.now(UTC) - timedelta(days=60)).strftime(

195 "%Y-%m-%d"

196 )

197 self.from_date = recent

198 self.order_by = "newest"

199 return

200

201 if not self.llm or not self.adaptive_search:

202 return

203

204 try:

205 prompt = f"""Is this query asking about HISTORICAL events or CURRENT events?

206

207Query: "{query}"

208

209ONE WORD ANSWER ONLY:

210- "HISTORICAL" if about past events (older than 1 year)

211- "CURRENT" if about recent events (within past year)

212- "UNCLEAR" if can't determine

213

214ONE WORD ONLY:"""

215

216 response = invoke_llm_sync(self.llm, prompt)

217 answer = (

218 remove_think_tags(

219 str(response.content)

220 if hasattr(response, "content")

221 else str(response)

222 )

223 .strip()

224 .upper()

225 )

226

227 # Reset to original parameters first

228 self.from_date = self._original_date_params["from_date"]

229 self.to_date = self._original_date_params["to_date"]

230

231 if "HISTORICAL" in answer:

232 # For historical queries, go back 10 years

233 logger.info(

234 "Query classified as HISTORICAL - extending search timeframe"

235 )

236 ten_years_ago = (

237 datetime.now(UTC) - timedelta(days=3650)

238 ).strftime("%Y-%m-%d")

239 self.from_date = ten_years_ago

240

241 elif "CURRENT" in answer:

242 # For current events, focus on recent content

243 logger.info(

244 "Query classified as CURRENT - focusing on recent content"

245 )

246 recent = (datetime.now(UTC) - timedelta(days=60)).strftime(

247 "%Y-%m-%d"

248 )

249 self.from_date = recent

250 self.order_by = "newest" # Prioritize newest for current events

251

252 except Exception as e:

253 safe_msg = self._scrub_error(e)

254 logger.warning(f"Error adapting dates for query type: {safe_msg}")

255 # Keep original date parameters on error

256

257 def _adaptive_search(self, query: str) -> Tuple[List[Dict[str, Any]], str]:

258 """

259 Perform adaptive search that progressively adjusts parameters based on results.

260

261 Args:

262 query: The search query

263

264 Returns:

265 Tuple of (list of articles, search strategy used)

266 """

267 # Try with current parameters

268 articles = self._get_all_data(query)

269 strategy = "initial"

270

271 # If no results or too few, try different strategies

272 if len(articles) < 3 and self.adaptive_search:

273 logger.info(

274 f"Initial search found only {len(articles)} results, trying alternative strategies"

275 )

276

277 # Try with expanded date range

278 original_from_date = self.from_date

279 original_order_by = self.order_by

280

281 # Strategy 1: Expand to 6 months

282 logger.info("Strategy 1: Expanding time range to 6 months")

283 six_months_ago = (datetime.now(UTC) - timedelta(days=180)).strftime(

284 "%Y-%m-%d"

285 )

286 self.from_date = six_months_ago

287

288 articles1 = self._get_all_data(query)

289 if len(articles1) > len(articles):

290 articles = articles1

291 strategy = "expanded_6mo"

292

293 # Strategy 2: Expand to all time and try relevance order

294 if len(articles) < 3:

295 logger.info(

296 "Strategy 2: Expanding to all time with relevance ordering"

297 )

298 self.from_date = "2000-01-01" # Effectively "all time"

299 self.order_by = "relevance"

300

301 articles2 = self._get_all_data(query)

302 if len(articles2) > len(articles):

303 articles = articles2

304 strategy = "all_time_relevance"

305

306 # Strategy 3: Try removing section constraints

307 if len(articles) < 3 and self.section:

308 logger.info("Strategy 3: Removing section constraint")

309 original_section = self.section

310 self.section = None

311

312 articles3 = self._get_all_data(query)

313 if len(articles3) > len(articles): 313 ↛ 318line 313 didn't jump to line 318 because the condition on line 313 was always true

314 articles = articles3

315 strategy = "no_section"

316

317 # Restore section setting

318 self.section = original_section

319

320 # Restore original settings

321 self.from_date = original_from_date

322 self.order_by = original_order_by

323

324 logger.info(

325 f"Adaptive search using strategy '{strategy}' found {len(articles)} results"

326 )

327 return articles, strategy

328

329 def _get_all_data(self, query: str) -> List[Dict[str, Any]]:

330 """

331 Get article data from The Guardian API in a single call.

332 Request article bodies only when full content is enabled.

333

334 Args:

335 query: The search query

336

337 Returns:

338 List of articles with all data

339 """

340 try:

341 # Ensure query is not empty

342 if not query or query.strip() == "":

343 query = "news"

344 logger.warning("Empty query provided, using 'news' as default")

345

346 # Ensure query is not too long for API

347 if len(query) > 100:

348 logger.warning(

349 f"Query too long for Guardian API ({len(query)} chars), truncating"

350 )

351 query = query[:100]

352

353 # Ensure max_results is an integer to avoid comparison errors

354 page_size = min(

355 int(self.max_results) if self.max_results is not None else 10,

356 50,

357 )

358

359 # Log full parameters for debugging

360 logger.info(f"Guardian API search request ({len(query)} chars)")

361 logger.info(

362 f"Guardian API date range: {self.from_date} to {self.to_date}"

363 )

364

365 params = {

366 "q": query,

367 "api-key": self.api_key,

368 "from-date": self.from_date,

369 "to-date": self.to_date,

370 "order-by": self.order_by,

371 "page-size": page_size, # API maximum is 50

372 "show-fields": (

373 "headline,trailText,byline,publication"

374 if self.search_snippets_only

375 else "headline,trailText,byline,body,publication"

376 ),

377 "show-tags": "keyword",

378 }

379

380 # Add section filter if specified

381 if self.section:

382 params["section"] = self.section

383

384 # Log the complete request parameters (except API key)

385 log_params = params.copy()

386 log_params["api-key"] = "REDACTED"

387 # The user's search text is the "q" param — keep it out of the log

388 # the same way the API key is (#5646). The engine name and the

389 # other request parameters are all that's useful here.

390 log_params["q"] = "REDACTED"

391 logger.info(f"Guardian API request parameters: {log_params}")

392

393 # Apply rate limiting before request

394 self._last_wait_time = self.rate_tracker.apply_rate_limit(

395 self.engine_type

396 )

397

398 # Execute the API request

399 response = safe_get(self.api_url, params=params)

400 response.raise_for_status()

401

402 data = response.json()

403

404 # Extract results from the response

405 articles = data.get("response", {}).get("results", [])

406 logger.info(f"Guardian API returned {len(articles)} articles")

407

408 # Format results for the selected content mode

409 formatted_articles = []

410 for i, article in enumerate(articles):

411 if i >= self.max_results:

412 break

413

414 fields = article.get("fields", {})

415

416 result = {

417 "id": article.get("id", ""),

418 "title": fields.get(

419 "headline", article.get("webTitle", "")

420 ),

421 "link": article.get("webUrl", ""),

422 "snippet": fields.get("trailText", ""),

423 "publication_date": article.get("webPublicationDate", ""),

424 "section": article.get("sectionName", ""),

425 "author": fields.get("byline", ""),

426 }

427

428 if not self.search_snippets_only:

429 result["content"] = fields.get("body", "")

430 result["full_content"] = fields.get("body", "")

431 result["full_content_is_complete"] = bool(

432 fields.get("body")

433 )

434

435 # Extract tags/keywords

436 tags = article.get("tags", [])

437 result["keywords"] = [

438 tag.get("webTitle", "")

439 for tag in tags

440 if tag.get("type") == "keyword"

441 ]

442

443 formatted_articles.append(result)

444

445 return formatted_articles

446

447 except RateLimitError:

448 raise

449 except Exception as e:

450 # Chain both scrubbers: explicit-value redaction catches the

451 # known api_key, regex-based _sanitize_error_message catches

452 # unknown credential shapes (Bearer tokens, ?key= URLs) that

453 # the upstream might echo back. Use logger.warning to drop

454 # the traceback chain.

455 safe_msg = self._scrub_error(e)

456 logger.warning(

457 f"Error getting data from The Guardian API: {safe_msg}"

458 )

459 self._raise_if_rate_limit(e)

460 return []

461

462 def _get_previews(self, query: str) -> List[Dict[str, Any]]:

463 """

464 Get preview information for Guardian articles with enhanced optimization.

465

466 Args:

467 query: The search query

468

469 Returns:

470 List of preview dictionaries

471 """

472 logger.info("Getting articles from The Guardian API")

473

474 # Step 1: Optimize the query using LLM

475 optimized_query = self._optimize_query_for_guardian(query)

476

477 # Step 2: Adapt date parameters based on query type

478 self._adapt_dates_for_query_type(optimized_query)

479

480 # Step 3: Perform adaptive search

481 articles, strategy = self._adaptive_search(optimized_query)

482

483 # Store search metadata for debugging. The two query strings this

484 # used to carry reached logger.info() in run() below, which is the

485 # success path of every Guardian search (#5646) — keep the lengths,

486 # which answer "was the query rewritten", and drop the text itself.

487 self._search_metadata = {

488 "original_query_length": len(query or ""),

489 "optimized_query_length": len(optimized_query or ""),

490 "strategy": strategy,

491 "from_date": self.from_date,

492 "to_date": self.to_date,

493 "section": self.section,

494 "order_by": self.order_by,

495 }

496

497 # Store full articles for later use

498 self._full_articles = {a["id"]: a for a in articles}

499

500 # Return only preview fields for each article

501 previews = []

502 for article in articles:

503 preview = {

504 "id": article["id"],

505 "title": article["title"],

506 "link": article["link"],

507 "snippet": article["snippet"],

508 "publication_date": article["publication_date"],

509 "section": article["section"],

510 "author": article["author"],

511 "keywords": article.get("keywords", []),

512 }

513 previews.append(preview)

514

515 return previews

516

517 def _get_full_content(

518 self, relevant_items: List[Dict[str, Any]]

519 ) -> List[Dict[str, Any]]:

520 """

521 Get full content for the relevant Guardian articles.

522 Restores full content from the cached data.

523

524 Args:

525 relevant_items: List of relevant preview dictionaries

526

527 Returns:

528 List of result dictionaries with full content

529 """

530 logger.info(

531 f"Adding full content to {len(relevant_items)} relevant Guardian articles"

532 )

533

534 # Get full articles for relevant items

535 results = []

536 for item in relevant_items:

537 article_id = item.get("id", "")

538

539 # Get the full article from our cache

540 if (

541 hasattr(self, "_full_articles")

542 and article_id in self._full_articles

543 ):

544 results.append(self._full_articles[article_id])

545 else:

546 # If not found (shouldn't happen), just use the preview

547 results.append(item)

548

549 return results

550

551 def run(

552 self, query: str, research_context: Dict[str, Any] | None = None

553 ) -> List[Dict[str, Any]]:

554 """

555 Execute a search using The Guardian API with the enhanced approach.

556

557 Args:

558 query: The search query

559 research_context: Context from previous research to use.

560

561 Returns:

562 List of search results

563 """

564 # Recheck cached engines before provider work; policy denials must

565 # escape the search-error handler below.

566 self._verify_egress_scope()

567 logger.info("---Execute a search using The Guardian (enhanced)---")

568

569 # Additional safety check for None query

570 if query is None:

571 logger.error("None query passed to Guardian search engine")

572 query = "news"

573

574 try:

575 # Get previews with our enhanced method

576 previews = self._get_previews(query)

577

578 # If no results, try one more time with a simplified query

579 if not previews:

580 simple_query = " ".join(

581 [w for w in query.split() if len(w) > 3][:3]

582 )

583 logger.warning(

584 "No Guardian articles found, retrying with a simplified query"

585 )

586 previews = self._get_previews(simple_query)

587

588 # If still no results, try with a very generic query as last resort

589 if not previews and "trump" in query.lower(): 589 ↛ 590line 589 didn't jump to line 590 because the condition on line 589 was never true

590 logger.warning("Trying last resort query: 'Donald Trump'")

591 previews = self._get_previews("Donald Trump")

592 elif not previews: 592 ↛ 597line 592 didn't jump to line 597 because the condition on line 592 was always true

593 logger.warning("Trying last resort query: 'news'")

594 previews = self._get_previews("news")

595

596 # If still no results after all attempts, return empty list

597 if not previews:

598 logger.warning(

599 "No Guardian articles found after multiple attempts"

600 )

601 return []

602

603 # Filter for relevance if we have an LLM

604 if (

605 self.llm

606 and hasattr(self, "max_filtered_results")

607 and self.max_filtered_results

608 ):

609 filtered_items = self._filter_for_relevance(previews, query)

610 if not filtered_items: 610 ↛ 612line 610 didn't jump to line 612 because the condition on line 610 was never true

611 # Fall back to unfiltered results if everything was filtered out

612 logger.warning(

613 "All articles filtered out, using unfiltered results"

614 )

615 filtered_items = previews[: self.max_filtered_results]

616 else:

617 filtered_items = previews

618

619 # Relevance filtering operates on previews in both modes. Only

620 # restore cached bodies when full content was requested.

621 results = (

622 filtered_items

623 if self.search_snippets_only

624 else self._get_full_content_cached(filtered_items)

625 )

626

627 # Add source information to make it clear these are from The Guardian

628 for result in results:

629 if "source" not in result:

630 result["source"] = "The Guardian"

631

632 # Clean up the cache after use

633 if hasattr(self, "_full_articles"):

634 del self._full_articles

635

636 # Restore original date parameters

637 self.from_date = self._original_date_params["from_date"]

638 self.to_date = self._original_date_params["to_date"]

639

640 # Log search metadata if available

641 if hasattr(self, "_search_metadata"):

642 logger.info(f"Search metadata: {self._search_metadata}")

643 del self._search_metadata

644

645 return results

646

647 except RateLimitError:

648 raise

649 except Exception as e:

650 safe_msg = self._scrub_error(e)

651 logger.warning(f"Error in Guardian search: {safe_msg}")

652

653 # Restore original date parameters on error

654 self.from_date = self._original_date_params["from_date"]

655 self.to_date = self._original_date_params["to_date"]

656

657 return []

658

659 def search_by_section(

660 self, section: str, max_results: Optional[int] = None

661 ) -> List[Dict[str, Any]]:

662 """

663 Search for articles in a specific section.

664

665 Args:

666 section: The Guardian section name (e.g., "politics", "technology")

667 max_results: Maximum number of results (defaults to self.max_results)

668

669 Returns:

670 List of articles in the section

671 """

672 original_section = self.section

673 original_max_results = self.max_results

674

675 try:

676 # Set section and max_results for this search

677 self.section = section

678 if max_results: 678 ↛ 679line 678 didn't jump to line 679 because the condition on line 678 was never true

679 self.max_results = max_results

680

681 # Use empty query to get all articles in the section

682 return self.run("")

683

684 finally:

685 # Restore original values

686 self.section = original_section

687 self.max_results = original_max_results

688

689 def get_recent_articles(

690 self, days: int = 7, max_results: Optional[int] = None

691 ) -> List[Dict[str, Any]]:

692 """

693 Get recent articles from The Guardian.

694

695 Args:

696 days: Number of days to look back

697 max_results: Maximum number of results (defaults to self.max_results)

698

699 Returns:

700 List of recent articles

701 """

702 original_from_date = self.from_date

703 original_order_by = self.order_by

704 original_max_results = self.max_results

705

706 try:

707 # Set parameters for this search

708 self.from_date = (

709 datetime.now(UTC) - timedelta(days=days)

710 ).strftime("%Y-%m-%d")

711 self.order_by = "newest"

712 if max_results: 712 ↛ 713line 712 didn't jump to line 713 because the condition on line 712 was never true

713 self.max_results = max_results

714

715 # Use empty query to get all recent articles

716 return self.run("")

717

718 finally:

719 # Restore original values

720 self.from_date = original_from_date

721 self.order_by = original_order_by

722 self.max_results = original_max_results

Coverage for src/local_deep_research/web_search_engines/engines/search_engine_guardian.py: 96%

252 statements