# modules/api_utils.py """API utility functions for Wikimedia services""" import requests from typing import Dict, List, Optional import time from config.settings import ( WIKIPEDIA_API, WIKIDATA_API, WIKIBOOKS_API, WIKI_REST_API, CACHE_TIMEOUT, ) # Cache for API responses _cache = {} def _get_cached_or_fetch(url: str, params: Dict = None) -> Optional[Dict]: """Get cached response or fetch from API. Uses a simple in-memory cache.""" cache_key = f"{url}_{str(params)}" if cache_key in _cache: cached_data, timestamp = _cache[cache_key] if time.time() - timestamp < CACHE_TIMEOUT: # Fixed: changed _cache_timeout to CACHE_TIMEOUT return cached_data try: response = requests.get( url, params=params, timeout=10 ) # Increased timeout for robustness if response.status_code == 200: data = response.json() _cache[cache_key] = (data, time.time() + CACHE_TIMEOUT) return data except requests.exceptions.RequestException as e: print(f"API request error: {e}") return None def fetch_wikipedia_summary(topic: str) -> Optional[Dict]: """Fetch Wikipedia page summary with caching""" return _get_cached_or_fetch(f"{WIKI_REST_API}{topic}") def search_wikipedia(query: str, limit: int = 5) -> List[str]: """Search Wikipedia for topics""" params = {"action": "opensearch", "search": query, "limit": limit, "format": "json"} data = _get_cached_or_fetch(WIKIPEDIA_API, params) if data and len(data) > 1: return data[1] return [] def fetch_wikidata_entity(entity_id: str) -> Optional[Dict]: """Fetch Wikidata entity information""" params = { "action": "wbgetentities", "ids": entity_id, "format": "json", "languages": "en", } return _get_cached_or_fetch(WIKIDATA_API, params) def fetch_wikipedia_categories(page_title: str) -> List[str]: """Fetch categories for a Wikipedia page""" params = { "action": "query", "prop": "categories", "titles": page_title, "format": "json", "cllimit": 10, } data = _get_cached_or_fetch(WIKIPEDIA_API, params) if data: pages = data.get("query", {}).get("pages", {}) for page_id, page_data in pages.items(): categories = page_data.get("categories", []) return [cat["title"].replace("Category:", "") for cat in categories] return [] def fetch_related_topics(topic: str, limit: int = 5) -> List[str]: """Fetch related topics from Wikipedia""" params = { "action": "query", "list": "search", "srsearch": topic, "srlimit": limit, "format": "json", } data = _get_cached_or_fetch(WIKIPEDIA_API, params) if data: search_results = data.get("query", {}).get("search", []) return [ result["title"] for result in search_results if result["title"] != topic ] return [] def fetch_wikibooks_content(topic: str) -> Optional[str]: """Fetch content from Wikibooks""" params = {"action": "query", "list": "search", "srsearch": topic, "format": "json"} data = _get_cached_or_fetch(WIKIBOOKS_API, params) if data: search_results = data.get("query", {}).get("search", []) if search_results: return search_results[0].get("snippet", "") return None