"""Wikipedia API: lightweight topic lookups and article search.""" import logging from urllib.parse import quote as url_quote import httpx logger = logging.getLogger(__name__) _SUMMARY_URL = "https://en.wikipedia.org/api/rest_v1/page/summary" _SEARCH_URL = "https://en.wikipedia.org/w/api.php" _TIMEOUT = 5.0 _USER_AGENT = "FabledAssistant/1.0 (https://fabledsword.com)" def _extract_thumbnail(data: dict) -> str: """Prefer the 320px thumbnail (kinder to cache/bandwidth) over originalimage.""" thumb = data.get("thumbnail", {}).get("source", "") if thumb: return thumb return data.get("originalimage", {}).get("source", "") async def wiki_summary(query: str) -> dict | None: """Fetch a Wikipedia summary for a given title/query. Does a direct title lookup via the REST v1 summary endpoint. Returns a dict with title, extract, url, and thumbnail_url on success; thumbnail_url is an empty string when the article has no image. Returns None on any failure. """ title = url_quote(query.replace(" ", "_")) url = f"{_SUMMARY_URL}/{title}" headers = {"User-Agent": _USER_AGENT} try: async with httpx.AsyncClient() as client: resp = await client.get(url, headers=headers, timeout=_TIMEOUT, follow_redirects=True) resp.raise_for_status() data = resp.json() if data.get("type") == "disambiguation": logger.debug("wiki_summary: disambiguation page for %r", query) return None extract = data.get("extract", "").strip() if not extract: logger.debug("wiki_summary: empty extract for %r", query) return None return { "title": data.get("title", query), "extract": extract, "url": data.get("content_urls", {}).get("desktop", {}).get("page", ""), "thumbnail_url": _extract_thumbnail(data), } except Exception as exc: logger.debug("wiki_summary failed for %r: %s", query, exc) return None async def wiki_search(query: str, limit: int = 3) -> list[dict]: """Search Wikipedia and return summaries for the top results. Uses the MediaWiki search API then fetches summaries for each hit. Skips disambiguation pages and empty extracts. Returns a list of dicts with title, extract, and url. """ headers = {"User-Agent": _USER_AGENT} params = { "action": "query", "list": "search", "srsearch": query, "srlimit": str(limit), "format": "json", } results: list[dict] = [] try: async with httpx.AsyncClient() as client: resp = await client.get( _SEARCH_URL, params=params, headers=headers, timeout=_TIMEOUT ) resp.raise_for_status() data = resp.json() hits = data.get("query", {}).get("search", []) for hit in hits: title = hit.get("title", "") encoded = url_quote(title.replace(" ", "_")) summary_url = f"{_SUMMARY_URL}/{encoded}" try: sr = await client.get( summary_url, headers=headers, timeout=_TIMEOUT, follow_redirects=True ) sr.raise_for_status() sdata = sr.json() if sdata.get("type") == "disambiguation": logger.debug("wiki_search: skipping disambiguation %r", title) continue extract = sdata.get("extract", "").strip() if not extract: logger.debug("wiki_search: empty extract for %r", title) continue results.append( { "title": sdata.get("title", title), "extract": extract, "url": sdata.get("content_urls", {}) .get("desktop", {}) .get("page", ""), "thumbnail_url": _extract_thumbnail(sdata), } ) except Exception as exc: logger.debug("wiki_search: summary fetch failed for %r: %s", title, exc) except Exception as exc: logger.debug("wiki_search failed for %r: %s", query, exc) return [] return results