feat: add wikipedia service with summary lookup and search

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-17 21:54:44 -04:00
parent 6f8c2201ed
commit 6a8f0e9143
2 changed files with 247 additions and 0 deletions
+101
View File
@@ -0,0 +1,101 @@
"""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)"
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, and url on success; 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", ""),
}
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", ""),
}
)
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