Coverage for src/local_deep_research/web_search_engines/engines/search_engine_wikinews.py: 95%
163 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1from datetime import datetime, timedelta, UTC
2from typing import Any, Dict, List, Optional, Tuple
4import json
5import html
6import re
7import requests
8from langchain_core.language_models import BaseLLM
10from ...constants import USER_AGENT
11from ...security.secure_logging import logger
12from ...utilities.json_utils import extract_json, get_llm_response_text
13from ...utilities.search_utilities import remove_think_tags, LANGUAGE_CODE_MAP
14from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
15from ...security import safe_get
17HEADERS = {"User-Agent": USER_AGENT}
18WIKINEWS_LANGUAGES = [
19 "ru",
20 "sr",
21 "pt",
22 "fr",
23 "pl",
24 "en",
25 "zh",
26 "de",
27 "it",
28 "es",
29 "cs",
30 "nl",
31 "ca",
32 "ar",
33 "ja",
34]
35TIMEOUT = 5 # Seconds
36TIME_PERIOD_DELTAS = {
37 "all": None, # No time filter
38 "y": timedelta(days=365), # 1 year
39 "m": timedelta(days=30), # 1 month
40 "w": timedelta(days=7), # 1 week
41 "d": timedelta(days=1), # 24 hours
42}
43DEFAULT_RECENT_BACKWARD_DAYS = 60
44MAX_RETRIES = 3
47class WikinewsSearchEngine(BaseSearchEngine):
48 """Wikinews search engine implementation with LLM query optimization"""
50 # Mark as public and news search engine
51 is_public = True
52 egress_sensitivity = Sensitivity.NON_SENSITIVE
53 egress_exposure = Exposure.EXPOSING
54 is_news = True
55 is_lexical = True
56 needs_llm_relevance_filter = True
58 def __init__(
59 self,
60 search_language: str = "english",
61 adaptive_search: bool = True,
62 time_period: str = "all",
63 llm: Optional[BaseLLM] = None,
64 max_filtered_results: Optional[int] = None,
65 max_results: int = 10,
66 search_snippets_only: bool = True,
67 settings_snapshot: Optional[Dict[str, Any]] = None,
68 **kwargs,
69 ):
70 """
71 Initialize the Wikinews search engine.
73 Args:
74 search_language (str): Language for Wikinews search (e.g. "english").
75 adaptive_search (bool): Whether to expand or shrink date ranges based on query.
76 time_period (str): Defines the look-back window used to filter search results ("all", "y", "m", "w", "d").
77 "all" (or any unrecognized value) applies no lower date bound.
78 llm (Optional[BaseLLM]): Language model used for query optimization and classification.
79 max_filtered_results (Optional[int]): Maximum number of results to keep after filtering.
80 max_results (int): Maximum number of search results to return.
81 search_snippets_only (bool): If True, full article content is ignored.
82 """
84 super().__init__(
85 llm=llm,
86 max_filtered_results=max_filtered_results,
87 max_results=max_results,
88 search_snippets_only=search_snippets_only,
89 settings_snapshot=settings_snapshot,
90 **kwargs,
91 )
93 # Language initialization
94 lang_code = LANGUAGE_CODE_MAP.get(
95 search_language.lower(),
96 "en", # Default to English if not found
97 )
99 if lang_code not in WIKINEWS_LANGUAGES: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true
100 logger.warning(
101 f"Wikinews does not support language '{search_language}' ({lang_code}). Defaulting to English."
102 )
103 lang_code = "en"
105 self.lang_code: str = lang_code
107 # Adaptive search
108 self.adaptive_search: bool = adaptive_search
110 # Date range initialization
111 now = datetime.now(UTC)
112 # Unrecognized values fall back to "all" (no lower bound), matching
113 # the tavily/serpapi/brave convention of unfiltered-on-unknown rather
114 # than silently imposing a year window.
115 delta = (
116 TIME_PERIOD_DELTAS.get(time_period)
117 if isinstance(time_period, str)
118 else None
119 )
120 self.from_date: datetime = (
121 now - delta if delta else datetime.min.replace(tzinfo=UTC)
122 )
123 self.to_date: datetime = now
125 # Preserve original date range so adaptive search can restore it
126 self._original_date_range = (self.from_date, self.to_date)
128 # API base URL
129 self.api_url: str = "https://{lang_code}.wikinews.org/w/api.php"
131 def _optimize_query_for_wikinews(self, query: str) -> str:
132 """
133 Optimize a natural language query for Wikinews search.
134 Uses LLM to transform questions into effective news search queries.
136 Args:
137 query (str): Natural language query
139 Returns:
140 Optimized search query for Wikinews
141 """
142 if not self.llm:
143 return query
145 try:
146 # Prompt for query optimization
147 prompt = f"""You are a query condenser. Your task is to transform the user’s natural-language question into a very short Wikinews search query.
149Input question:
150"{query}"
152STRICT OUTPUT REQUIREMENTS (follow ALL of them):
1531. Return ONLY a JSON object with EXACTLY one field: {{"query": "<refined_query>"}}.
1542. The JSON must be valid, minified, and contain no trailing text.
1553. The refined query must be extremely short: MAXIMUM 3–4 words.
1564. Include only the essential keywords (proper names, events, entities, places).
1575. Remove filler words (e.g., "news", "latest", "about", "what", "how", "is").
1586. DO NOT add Boolean operators (AND, OR).
1597. DO NOT use quotes inside the query.
1608. DO NOT add explanations or comments.
162EXAMPLES:
163- "What's the impact of rising interest rates on UK housing market?" → {{"query": "UK housing rates"}}
164- "Latest developments in the Ukraine-Russia peace negotiations" → {{"query": "Ukraine Russia negotiations"}}
165- "How are tech companies responding to AI regulation?" → {{"query": "tech AI regulation"}}
166- "What is Donald Trump's current political activity?" → {{"query": "Trump political activity"}}
168NOW RETURN ONLY THE JSON OBJECT.
169"""
170 # Get response from LLM
171 response = self.llm.invoke(prompt)
172 response_text = get_llm_response_text(response)
174 data = extract_json(response_text, expected_type=dict)
176 if data is None or not isinstance(data, dict):
177 raise ValueError("No valid JSON found in response") # noqa: TRY301 — caught by except ValueError to fall back to original query
179 optimized_query: str = str(data.get("query", "")).strip()
181 if not optimized_query:
182 raise ValueError("Query field missing or empty") # noqa: TRY301 — caught by except ValueError to fall back to original query
184 except (
185 ValueError,
186 TypeError,
187 AttributeError,
188 json.JSONDecodeError,
189 ):
190 logger.warning(
191 "Error optimizing query for WikinewsUsing original query."
192 )
193 return query
195 logger.info(f"Original query: '{query}'")
196 logger.info(f"Optimized for Wikinews: '{optimized_query}'")
198 return optimized_query
200 def _adapt_date_range_for_query(self, query: str) -> None:
201 """
202 Adapt the date range based on the query type (historical vs recent events).
204 Args:
205 query (str): The search query
206 """
207 # Reset to original date parameters first
208 self.from_date, self.to_date = self._original_date_range
210 if not self.adaptive_search or not self.llm:
211 return
213 # Do not adapt for very short queries (no enough context)
214 if len(query.split()) <= 4:
215 return
217 try:
218 prompt = f"""Classify this query based on temporal scope.
220Query: "{query}"
222Current date: {datetime.now(UTC).strftime("%Y-%m-%d")}
223Cutoff: Events within the last {DEFAULT_RECENT_BACKWARD_DAYS} days are CURRENT
225Classification rules:
226- CURRENT: Recent events (last {DEFAULT_RECENT_BACKWARD_DAYS} days), ongoing situations, "latest", "recent", "today", "this week"
227- HISTORICAL: Events before {DEFAULT_RECENT_BACKWARD_DAYS} days ago, timelines, chronologies, past tense ("what happened", "history of")
228- UNCLEAR: Ambiguous temporal context
230Respond with ONE WORD ONLY: CURRENT, HISTORICAL, or UNCLEAR"""
231 # Get response from LLM
232 response = self.llm.invoke(prompt)
233 response_text = (
234 getattr(response, "content", None)
235 or getattr(response, "text", None)
236 or str(response)
237 )
238 answer = remove_think_tags(response_text).upper()
240 if "CURRENT" in answer:
241 # For current events, focus on recent content
242 logger.info(
243 f"Query '{query}' classified as CURRENT - focusing on recent content"
244 )
245 self.from_date = datetime.now(UTC) - timedelta(
246 days=DEFAULT_RECENT_BACKWARD_DAYS
247 )
248 elif "HISTORICAL" in answer:
249 # For historical queries, go back as far as possible
250 logger.info(
251 f"Query '{query}' classified as HISTORICAL - extending search timeframe"
252 )
253 self.from_date = datetime.min.replace(tzinfo=UTC)
254 else:
255 logger.info(
256 f"Query '{query}' classified as UNCLEAR - keeping original date range"
257 )
259 except (AttributeError, TypeError, ValueError, RuntimeError) as e:
260 # Keep original date parameters on error
261 safe_msg = self._scrub_error(e)
262 logger.exception(
263 f"Error adapting date range for query '{query}' ({type(e).__name__}): keeping original date range: {safe_msg}"
264 )
266 def _fetch_search_results(
267 self, query: str, sroffset: int
268 ) -> List[Dict[str, Any]]:
269 """Fetch search results from Wikinews API.
271 Args:
272 query (str): The search query.
273 sroffset (int): The result offset for pagination.
275 Returns:
276 List of search result items.
277 """
278 retries = 0
279 while retries < MAX_RETRIES:
280 params = {
281 "action": "query",
282 "list": "search",
283 "srsearch": query,
284 "srprop": "snippet|timestamp",
285 "srlimit": 50,
286 "sroffset": sroffset,
287 "format": "json",
288 }
290 # Apply rate limiting before search request
291 self._last_wait_time = self.rate_tracker.apply_rate_limit(
292 self.engine_type
293 )
295 try:
296 response = safe_get(
297 self.api_url.format(lang_code=self.lang_code),
298 params=params,
299 headers=HEADERS,
300 timeout=TIMEOUT,
301 )
302 response.raise_for_status()
303 data = response.json()
304 return data.get("query", {}).get("search", []) # type: ignore[no-any-return]
305 except (
306 requests.exceptions.RequestException,
307 json.JSONDecodeError,
308 ):
309 logger.warning("Error fetching search resultsretrying...")
310 retries += 1
312 return []
314 def _process_search_result(
315 self, result: Dict[str, Any], query: str
316 ) -> Optional[Dict[str, Any]]:
317 """Process and filter a single search result.
319 Args:
320 result (Dict[str, Any]): A single search result item.
321 query (str): The search query.
323 Returns:
324 Processed result or None if filtered out.
325 """
326 page_id = result.get("pageid")
327 title = result.get("title", "")
328 snippet = _clean_wikinews_snippet(result.get("snippet", ""))
330 try:
331 last_edit_timestamp = result.get("timestamp", "")
332 last_edit_date = datetime.fromisoformat(
333 last_edit_timestamp.replace("Z", "+00:00")
334 )
335 except ValueError:
336 logger.warning(
337 f"Error parsing last edit date for page {page_id}, using current date as fallback."
338 )
339 last_edit_date = datetime.now(UTC)
341 # First filter: last edit date must be after from_date
342 if last_edit_date < self.from_date: 342 ↛ 344line 342 didn't jump to line 344 because the condition on line 342 was never true
343 # In this case we can skip fetching full content
344 return None
346 # Fetch full article content and extract actual publication date
347 # Note: Wikinews API do not allow to retrieve publication date in batched search results
348 full_content, publication_date = self._fetch_full_content_and_pubdate(
349 int(page_id) if page_id is not None else 0, last_edit_date
350 )
352 # Second filter: publication date within range
353 if publication_date < self.from_date or publication_date > self.to_date:
354 return None
356 # Third filter: check if all query words are in title or content
357 # Note: Wikinews search return false positive if query words are in "related" articles section
358 # Use word boundary matching to avoid substring matches (e.g., "is" matching "This")
359 combined_text = f"{title} {full_content}".lower()
360 query_words = [
361 w.lower() for w in query.split() if len(w) > 1
362 ] # Skip single chars
363 if query_words and not all( 363 ↛ 367line 363 didn't jump to line 367 because the condition on line 363 was never true
364 re.search(rf"\b{re.escape(word)}\b", combined_text)
365 for word in query_words
366 ):
367 return None
369 # If only snippets are requested, we use snippet as full content
370 if self.search_snippets_only:
371 full_content = snippet
373 return {
374 "id": page_id,
375 "title": title,
376 "snippet": snippet,
377 "source": "wikinews",
378 "url": f"https://{self.lang_code}.wikinews.org/?curid={page_id}", # Used by '_filter_for_relevance' function
379 "link": f"https://{self.lang_code}.wikinews.org/?curid={page_id}", # Used by citation handler
380 "content": full_content,
381 "full_content": full_content,
382 "publication_date": publication_date.isoformat(timespec="seconds"),
383 }
385 def _fetch_full_content_and_pubdate(
386 self, page_id: int, fallback_date: datetime
387 ) -> Tuple[str, datetime]:
388 """Fetch full article content and publication date from Wikinews API.
390 Args:
391 page_id (int): The Wikinews page ID.
392 fallback_date (datetime): Fallback date if publication date cannot be determined.
394 Returns:
395 Tuple of (full_content, publication_date)
396 """
397 try:
398 content_params = {
399 "action": "query",
400 "prop": "revisions|extracts",
401 "pageids": page_id,
402 "rvprop": "timestamp",
403 "rvdir": "newer", # Older revisions first
404 "rvlimit": 1, # Get the first revision (i.e. publication)
405 "explaintext": True,
406 "format": "json",
407 }
409 # Apply rate limiting before content request
410 self._last_wait_time = self.rate_tracker.apply_rate_limit(
411 self.engine_type
412 )
414 content_resp = safe_get(
415 self.api_url.format(lang_code=self.lang_code),
416 params=content_params,
417 headers=HEADERS,
418 timeout=TIMEOUT,
419 )
420 content_resp.raise_for_status()
421 content_data = content_resp.json()
423 page_data = (
424 content_data.get("query", {})
425 .get("pages", {})
426 .get(str(page_id), {})
427 )
428 full_content = page_data.get("extract", "")
429 revisions = page_data.get("revisions", [])
431 if revisions:
432 try:
433 # First revision timestamp is the publication date
434 publication_date = datetime.fromisoformat(
435 revisions[0]["timestamp"].replace("Z", "+00:00")
436 )
437 except ValueError:
438 logger.warning(
439 f"Error parsing publication date for page {page_id}, using fallback date."
440 )
441 publication_date = fallback_date
442 else:
443 logger.warning(
444 f"No revisions found for page {page_id}, using fallback date."
445 )
446 publication_date = fallback_date
448 return full_content, publication_date
450 except (
451 requests.exceptions.RequestException,
452 json.JSONDecodeError,
453 ):
454 logger.warning(f"Error fetching content for page {page_id}")
455 return "", fallback_date
457 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
458 """
459 Retrieve article previews from Wikinews based on the query.
461 Args:
462 query (str): The search query
464 Returns:
465 List of relevant article previews
466 """
467 # Adapt date range based on query and optimize query (if LLM is available)
468 self._adapt_date_range_for_query(query)
469 optimized_query = self._optimize_query_for_wikinews(query)
471 articles: list[dict[str, Any]] = []
472 sroffset = 0
474 while len(articles) < self.max_results:
475 search_results = self._fetch_search_results(
476 optimized_query, sroffset
477 )
478 if not search_results:
479 # No more results available (or multiple retries failed)
480 break
482 for result in search_results: 482 ↛ 489line 482 didn't jump to line 489 because the loop on line 482 didn't complete
483 article = self._process_search_result(result, optimized_query)
484 if article: 484 ↛ 486line 484 didn't jump to line 486 because the condition on line 484 was always true
485 articles.append(article)
486 if len(articles) >= self.max_results: 486 ↛ 482line 486 didn't jump to line 482 because the condition on line 486 was always true
487 break
489 sroffset += len(search_results)
491 return articles
493 def _get_full_content(
494 self, relevant_items: List[Dict[str, Any]]
495 ) -> List[Dict[str, Any]]:
496 """
497 Retrieve full content for relevant Wikinews articles.
499 Args:
500 relevant_items (List[Dict[str, Any]]): List of relevant article previews
502 Returns:
503 List of articles with full content
504 """
505 # Since full content is already fetched in _get_previews, just return relevant items
506 return relevant_items
509def _clean_wikinews_snippet(snippet: str) -> str:
510 """
511 Clean a Wikinews search snippet.
513 Args:
514 snippet (str): Raw snippet from Wikinews API
516 Returns:
517 Clean human-readable text
518 """
519 if not snippet:
520 return ""
522 # Unescape HTML entities
523 unescaped = html.unescape(snippet)
525 # Remove HTML tags
526 clean_text = re.sub(r"<.*?>", "", unescaped)
528 # Normalize whitespace
529 return re.sub(r"\s+", " ", clean_text).strip()