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