Coverage for src/local_deep_research/web_search_engines/engines/search_engine_wikinews.py: 95%
163 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
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 = "y",
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 llm (Optional[BaseLLM]): Language model used for query optimization and classification.
78 max_filtered_results (Optional[int]): Maximum number of results to keep after filtering.
79 max_results (int): Maximum number of search results to return.
80 search_snippets_only (bool): If True, full article content is ignored.
81 """
83 super().__init__(
84 llm=llm,
85 max_filtered_results=max_filtered_results,
86 max_results=max_results,
87 search_snippets_only=search_snippets_only,
88 settings_snapshot=settings_snapshot,
89 **kwargs,
90 )
92 # Language initialization
93 lang_code = LANGUAGE_CODE_MAP.get(
94 search_language.lower(),
95 "en", # Default to English if not found
96 )
98 if lang_code not in WIKINEWS_LANGUAGES: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true
99 logger.warning(
100 f"Wikinews does not support language '{search_language}' ({lang_code}). Defaulting to English."
101 )
102 lang_code = "en"
104 self.lang_code: str = lang_code
106 # Adaptive search
107 self.adaptive_search: bool = adaptive_search
109 # Date range initialization
110 now = datetime.now(UTC)
111 delta = TIME_PERIOD_DELTAS.get(time_period, timedelta(days=365))
112 self.from_date: datetime = (
113 now - delta if delta else datetime.min.replace(tzinfo=UTC)
114 )
115 self.to_date: datetime = now
117 # Preserve original date range so adaptive search can restore it
118 self._original_date_range = (self.from_date, self.to_date)
120 # API base URL
121 self.api_url: str = "https://{lang_code}.wikinews.org/w/api.php"
123 def _optimize_query_for_wikinews(self, query: str) -> str:
124 """
125 Optimize a natural language query for Wikinews search.
126 Uses LLM to transform questions into effective news search queries.
128 Args:
129 query (str): Natural language query
131 Returns:
132 Optimized search query for Wikinews
133 """
134 if not self.llm:
135 return query
137 try:
138 # Prompt for query optimization
139 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.
141Input question:
142"{query}"
144STRICT OUTPUT REQUIREMENTS (follow ALL of them):
1451. Return ONLY a JSON object with EXACTLY one field: {{"query": "<refined_query>"}}.
1462. The JSON must be valid, minified, and contain no trailing text.
1473. The refined query must be extremely short: MAXIMUM 3–4 words.
1484. Include only the essential keywords (proper names, events, entities, places).
1495. Remove filler words (e.g., "news", "latest", "about", "what", "how", "is").
1506. DO NOT add Boolean operators (AND, OR).
1517. DO NOT use quotes inside the query.
1528. DO NOT add explanations or comments.
154EXAMPLES:
155- "What's the impact of rising interest rates on UK housing market?" → {{"query": "UK housing rates"}}
156- "Latest developments in the Ukraine-Russia peace negotiations" → {{"query": "Ukraine Russia negotiations"}}
157- "How are tech companies responding to AI regulation?" → {{"query": "tech AI regulation"}}
158- "What is Donald Trump's current political activity?" → {{"query": "Trump political activity"}}
160NOW RETURN ONLY THE JSON OBJECT.
161"""
162 # Get response from LLM
163 response = self.llm.invoke(prompt)
164 response_text = get_llm_response_text(response)
166 data = extract_json(response_text, expected_type=dict)
168 if data is None or not isinstance(data, dict):
169 raise ValueError("No valid JSON found in response") # noqa: TRY301 — caught by except ValueError to fall back to original query
171 optimized_query: str = str(data.get("query", "")).strip()
173 if not optimized_query:
174 raise ValueError("Query field missing or empty") # noqa: TRY301 — caught by except ValueError to fall back to original query
176 except (
177 ValueError,
178 TypeError,
179 AttributeError,
180 json.JSONDecodeError,
181 ):
182 logger.warning(
183 "Error optimizing query for WikinewsUsing original query."
184 )
185 return query
187 logger.info(f"Original query: '{query}'")
188 logger.info(f"Optimized for Wikinews: '{optimized_query}'")
190 return optimized_query
192 def _adapt_date_range_for_query(self, query: str) -> None:
193 """
194 Adapt the date range based on the query type (historical vs recent events).
196 Args:
197 query (str): The search query
198 """
199 # Reset to original date parameters first
200 self.from_date, self.to_date = self._original_date_range
202 if not self.adaptive_search or not self.llm:
203 return
205 # Do not adapt for very short queries (no enough context)
206 if len(query.split()) <= 4:
207 return
209 try:
210 prompt = f"""Classify this query based on temporal scope.
212Query: "{query}"
214Current date: {datetime.now(UTC).strftime("%Y-%m-%d")}
215Cutoff: Events within the last {DEFAULT_RECENT_BACKWARD_DAYS} days are CURRENT
217Classification rules:
218- CURRENT: Recent events (last {DEFAULT_RECENT_BACKWARD_DAYS} days), ongoing situations, "latest", "recent", "today", "this week"
219- HISTORICAL: Events before {DEFAULT_RECENT_BACKWARD_DAYS} days ago, timelines, chronologies, past tense ("what happened", "history of")
220- UNCLEAR: Ambiguous temporal context
222Respond with ONE WORD ONLY: CURRENT, HISTORICAL, or UNCLEAR"""
223 # Get response from LLM
224 response = self.llm.invoke(prompt)
225 response_text = (
226 getattr(response, "content", None)
227 or getattr(response, "text", None)
228 or str(response)
229 )
230 answer = remove_think_tags(response_text).upper()
232 if "CURRENT" in answer:
233 # For current events, focus on recent content
234 logger.info(
235 f"Query '{query}' classified as CURRENT - focusing on recent content"
236 )
237 self.from_date = datetime.now(UTC) - timedelta(
238 days=DEFAULT_RECENT_BACKWARD_DAYS
239 )
240 elif "HISTORICAL" in answer:
241 # For historical queries, go back as far as possible
242 logger.info(
243 f"Query '{query}' classified as HISTORICAL - extending search timeframe"
244 )
245 self.from_date = datetime.min.replace(tzinfo=UTC)
246 else:
247 logger.info(
248 f"Query '{query}' classified as UNCLEAR - keeping original date range"
249 )
251 except (AttributeError, TypeError, ValueError, RuntimeError) as e:
252 # Keep original date parameters on error
253 safe_msg = self._scrub_error(e)
254 logger.exception(
255 f"Error adapting date range for query '{query}' ({type(e).__name__}): keeping original date range: {safe_msg}"
256 )
258 def _fetch_search_results(
259 self, query: str, sroffset: int
260 ) -> List[Dict[str, Any]]:
261 """Fetch search results from Wikinews API.
263 Args:
264 query (str): The search query.
265 sroffset (int): The result offset for pagination.
267 Returns:
268 List of search result items.
269 """
270 retries = 0
271 while retries < MAX_RETRIES:
272 params = {
273 "action": "query",
274 "list": "search",
275 "srsearch": query,
276 "srprop": "snippet|timestamp",
277 "srlimit": 50,
278 "sroffset": sroffset,
279 "format": "json",
280 }
282 # Apply rate limiting before search request
283 self._last_wait_time = self.rate_tracker.apply_rate_limit(
284 self.engine_type
285 )
287 try:
288 response = safe_get(
289 self.api_url.format(lang_code=self.lang_code),
290 params=params,
291 headers=HEADERS,
292 timeout=TIMEOUT,
293 )
294 response.raise_for_status()
295 data = response.json()
296 return data.get("query", {}).get("search", []) # type: ignore[no-any-return]
297 except (
298 requests.exceptions.RequestException,
299 json.JSONDecodeError,
300 ):
301 logger.warning("Error fetching search resultsretrying...")
302 retries += 1
304 return []
306 def _process_search_result(
307 self, result: Dict[str, Any], query: str
308 ) -> Optional[Dict[str, Any]]:
309 """Process and filter a single search result.
311 Args:
312 result (Dict[str, Any]): A single search result item.
313 query (str): The search query.
315 Returns:
316 Processed result or None if filtered out.
317 """
318 page_id = result.get("pageid")
319 title = result.get("title", "")
320 snippet = _clean_wikinews_snippet(result.get("snippet", ""))
322 try:
323 last_edit_timestamp = result.get("timestamp", "")
324 last_edit_date = datetime.fromisoformat(
325 last_edit_timestamp.replace("Z", "+00:00")
326 )
327 except ValueError:
328 logger.warning(
329 f"Error parsing last edit date for page {page_id}, using current date as fallback."
330 )
331 last_edit_date = datetime.now(UTC)
333 # First filter: last edit date must be after from_date
334 if last_edit_date < self.from_date: 334 ↛ 336line 334 didn't jump to line 336 because the condition on line 334 was never true
335 # In this case we can skip fetching full content
336 return None
338 # Fetch full article content and extract actual publication date
339 # Note: Wikinews API do not allow to retrieve publication date in batched search results
340 full_content, publication_date = self._fetch_full_content_and_pubdate(
341 int(page_id) if page_id is not None else 0, last_edit_date
342 )
344 # Second filter: publication date within range
345 if publication_date < self.from_date or publication_date > self.to_date:
346 return None
348 # Third filter: check if all query words are in title or content
349 # Note: Wikinews search return false positive if query words are in "related" articles section
350 # Use word boundary matching to avoid substring matches (e.g., "is" matching "This")
351 combined_text = f"{title} {full_content}".lower()
352 query_words = [
353 w.lower() for w in query.split() if len(w) > 1
354 ] # Skip single chars
355 if query_words and not all( 355 ↛ 359line 355 didn't jump to line 359 because the condition on line 355 was never true
356 re.search(rf"\b{re.escape(word)}\b", combined_text)
357 for word in query_words
358 ):
359 return None
361 # If only snippets are requested, we use snippet as full content
362 if self.search_snippets_only:
363 full_content = snippet
365 return {
366 "id": page_id,
367 "title": title,
368 "snippet": snippet,
369 "source": "wikinews",
370 "url": f"https://{self.lang_code}.wikinews.org/?curid={page_id}", # Used by '_filter_for_relevance' function
371 "link": f"https://{self.lang_code}.wikinews.org/?curid={page_id}", # Used by citation handler
372 "content": full_content,
373 "full_content": full_content,
374 "publication_date": publication_date.isoformat(timespec="seconds"),
375 }
377 def _fetch_full_content_and_pubdate(
378 self, page_id: int, fallback_date: datetime
379 ) -> Tuple[str, datetime]:
380 """Fetch full article content and publication date from Wikinews API.
382 Args:
383 page_id (int): The Wikinews page ID.
384 fallback_date (datetime): Fallback date if publication date cannot be determined.
386 Returns:
387 Tuple of (full_content, publication_date)
388 """
389 try:
390 content_params = {
391 "action": "query",
392 "prop": "revisions|extracts",
393 "pageids": page_id,
394 "rvprop": "timestamp",
395 "rvdir": "newer", # Older revisions first
396 "rvlimit": 1, # Get the first revision (i.e. publication)
397 "explaintext": True,
398 "format": "json",
399 }
401 # Apply rate limiting before content request
402 self._last_wait_time = self.rate_tracker.apply_rate_limit(
403 self.engine_type
404 )
406 content_resp = safe_get(
407 self.api_url.format(lang_code=self.lang_code),
408 params=content_params,
409 headers=HEADERS,
410 timeout=TIMEOUT,
411 )
412 content_resp.raise_for_status()
413 content_data = content_resp.json()
415 page_data = (
416 content_data.get("query", {})
417 .get("pages", {})
418 .get(str(page_id), {})
419 )
420 full_content = page_data.get("extract", "")
421 revisions = page_data.get("revisions", [])
423 if revisions:
424 try:
425 # First revision timestamp is the publication date
426 publication_date = datetime.fromisoformat(
427 revisions[0]["timestamp"].replace("Z", "+00:00")
428 )
429 except ValueError:
430 logger.warning(
431 f"Error parsing publication date for page {page_id}, using fallback date."
432 )
433 publication_date = fallback_date
434 else:
435 logger.warning(
436 f"No revisions found for page {page_id}, using fallback date."
437 )
438 publication_date = fallback_date
440 return full_content, publication_date
442 except (
443 requests.exceptions.RequestException,
444 json.JSONDecodeError,
445 ):
446 logger.warning(f"Error fetching content for page {page_id}")
447 return "", fallback_date
449 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
450 """
451 Retrieve article previews from Wikinews based on the query.
453 Args:
454 query (str): The search query
456 Returns:
457 List of relevant article previews
458 """
459 # Adapt date range based on query and optimize query (if LLM is available)
460 self._adapt_date_range_for_query(query)
461 optimized_query = self._optimize_query_for_wikinews(query)
463 articles: list[dict[str, Any]] = []
464 sroffset = 0
466 while len(articles) < self.max_results:
467 search_results = self._fetch_search_results(
468 optimized_query, sroffset
469 )
470 if not search_results:
471 # No more results available (or multiple retries failed)
472 break
474 for result in search_results: 474 ↛ 481line 474 didn't jump to line 481 because the loop on line 474 didn't complete
475 article = self._process_search_result(result, optimized_query)
476 if article: 476 ↛ 478line 476 didn't jump to line 478 because the condition on line 476 was always true
477 articles.append(article)
478 if len(articles) >= self.max_results: 478 ↛ 474line 478 didn't jump to line 474 because the condition on line 478 was always true
479 break
481 sroffset += len(search_results)
483 return articles
485 def _get_full_content(
486 self, relevant_items: List[Dict[str, Any]]
487 ) -> List[Dict[str, Any]]:
488 """
489 Retrieve full content for relevant Wikinews articles.
491 Args:
492 relevant_items (List[Dict[str, Any]]): List of relevant article previews
494 Returns:
495 List of articles with full content
496 """
497 # Since full content is already fetched in _get_previews, just return relevant items
498 return relevant_items
501def _clean_wikinews_snippet(snippet: str) -> str:
502 """
503 Clean a Wikinews search snippet.
505 Args:
506 snippet (str): Raw snippet from Wikinews API
508 Returns:
509 Clean human-readable text
510 """
511 if not snippet:
512 return ""
514 # Unescape HTML entities
515 unescaped = html.unescape(snippet)
517 # Remove HTML tags
518 clean_text = re.sub(r"<.*?>", "", unescaped)
520 # Normalize whitespace
521 return re.sub(r"\s+", " ", clean_text).strip()