From 6a8f0e9143b4b5d335df758ced6bdbaf8adb262f Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 17 Apr 2026 21:54:44 -0400 Subject: [PATCH] feat: add wikipedia service with summary lookup and search Co-Authored-By: Claude Sonnet 4.6 --- src/fabledassistant/services/wikipedia.py | 101 +++++++++++++++ tests/test_wikipedia.py | 146 ++++++++++++++++++++++ 2 files changed, 247 insertions(+) create mode 100644 src/fabledassistant/services/wikipedia.py create mode 100644 tests/test_wikipedia.py diff --git a/src/fabledassistant/services/wikipedia.py b/src/fabledassistant/services/wikipedia.py new file mode 100644 index 0000000..e705b7c --- /dev/null +++ b/src/fabledassistant/services/wikipedia.py @@ -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 diff --git a/tests/test_wikipedia.py b/tests/test_wikipedia.py new file mode 100644 index 0000000..e82e0b2 --- /dev/null +++ b/tests/test_wikipedia.py @@ -0,0 +1,146 @@ +"""Tests for the Wikipedia service module.""" +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +from fabledassistant.services.wikipedia import wiki_summary, wiki_search + + +def _make_mock_response(json_data: dict, status_code: int = 200) -> MagicMock: + mock_resp = MagicMock() + mock_resp.status_code = status_code + mock_resp.json.return_value = json_data + mock_resp.raise_for_status = MagicMock() + return mock_resp + + +@pytest.mark.asyncio +async def test_wiki_summary_returns_extract(): + """Successful lookup returns a dict with title, extract, and url.""" + payload = { + "type": "standard", + "title": "Python (programming language)", + "extract": "Python is a high-level programming language.", + "content_urls": {"desktop": {"page": "https://en.wikipedia.org/wiki/Python_(programming_language)"}}, + } + mock_resp = _make_mock_response(payload) + + with patch("fabledassistant.services.wikipedia.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client.get = AsyncMock(return_value=mock_resp) + mock_client_cls.return_value = mock_client + + result = await wiki_summary("Python programming language") + + assert result is not None + assert result["title"] == "Python (programming language)" + assert "Python" in result["extract"] + assert result["url"].startswith("https://") + + +@pytest.mark.asyncio +async def test_wiki_summary_returns_none_on_404(): + """HTTP error (e.g. 404) causes wiki_summary to return None.""" + import httpx + + mock_resp = MagicMock() + mock_resp.raise_for_status.side_effect = httpx.HTTPStatusError( + "404", request=MagicMock(), response=MagicMock() + ) + + with patch("fabledassistant.services.wikipedia.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client.get = AsyncMock(return_value=mock_resp) + mock_client_cls.return_value = mock_client + + result = await wiki_summary("NonExistentPageXYZ123") + + assert result is None + + +@pytest.mark.asyncio +async def test_wiki_summary_returns_none_on_disambiguation(): + """Disambiguation pages cause wiki_summary to return None.""" + payload = { + "type": "disambiguation", + "title": "Mercury", + "extract": "Mercury may refer to:", + "content_urls": {"desktop": {"page": "https://en.wikipedia.org/wiki/Mercury"}}, + } + mock_resp = _make_mock_response(payload) + + with patch("fabledassistant.services.wikipedia.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client.get = AsyncMock(return_value=mock_resp) + mock_client_cls.return_value = mock_client + + result = await wiki_summary("Mercury") + + assert result is None + + +@pytest.mark.asyncio +async def test_wiki_search_returns_results(): + """wiki_search returns a list of result dicts for a successful query.""" + search_payload = { + "query": { + "search": [ + {"title": "Quantum mechanics"}, + {"title": "Quantum field theory"}, + ] + } + } + summary_payloads = [ + { + "type": "standard", + "title": "Quantum mechanics", + "extract": "Quantum mechanics is a fundamental theory in physics.", + "content_urls": {"desktop": {"page": "https://en.wikipedia.org/wiki/Quantum_mechanics"}}, + }, + { + "type": "standard", + "title": "Quantum field theory", + "extract": "Quantum field theory is a framework in theoretical physics.", + "content_urls": {"desktop": {"page": "https://en.wikipedia.org/wiki/Quantum_field_theory"}}, + }, + ] + + search_resp = _make_mock_response(search_payload) + summary_resp_1 = _make_mock_response(summary_payloads[0]) + summary_resp_2 = _make_mock_response(summary_payloads[1]) + + with patch("fabledassistant.services.wikipedia.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client.get = AsyncMock(side_effect=[search_resp, summary_resp_1, summary_resp_2]) + mock_client_cls.return_value = mock_client + + results = await wiki_search("quantum physics", limit=2) + + assert len(results) == 2 + assert results[0]["title"] == "Quantum mechanics" + assert "quantum" in results[0]["extract"].lower() + assert results[1]["title"] == "Quantum field theory" + + +@pytest.mark.asyncio +async def test_wiki_search_returns_empty_on_failure(): + """wiki_search returns an empty list when the network request fails.""" + import httpx + + with patch("fabledassistant.services.wikipedia.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client.get = AsyncMock(side_effect=httpx.ConnectError("connection refused")) + mock_client_cls.return_value = mock_client + + results = await wiki_search("anything") + + assert results == []