commit 71e4a6115e144202e9ae1b5d41d4e2fbdc6507a4 Author: kuhyx Date: Sun Jul 26 18:12:56 2026 +0200 Consolidate CK3/KCD wiki knowledge bases into one generic wiki-kb Merges ~/ck3-knowledge and ~/kcd-knowledge, which were 90% duplicate code and neither of which had ever been committed anywhere. Works on any wiki, not just MediaWiki: - sources.py splits the engine-specific parts (enumerate, fetch) behind a Source protocol. MediaWikiSource is preferred whenever an api.php exists; SitemapHtmlSource handles custom engines with no API (verified on Fextralife). Everything downstream was already source-agnostic. - scripts/detect_wiki.py probes /api.php, /w/api.php then /sitemap.xml and prints the config to use. The API path and article URL vary independently — Fandom serves /api.php but /wiki/{Title}. - Profiles are plain env files, since config.py was already fully env-driven. Per-wiki behaviour that must be measured rather than assumed: - Subpage policy. On the CK3 wiki "A/B" is a mod; on the KCD wiki it is the sequel, and rejecting subpages deletes 441 pages of wanted content. - Noise sections. "Other languages" tables were 7% of KCD chunks and matched English queries by repeating every title in eight languages. - Variant tagging for wikis covering two editions, with scopes that exclude the other edition but keep both/unknown pages. Build performance: length bucketing in embed_efficiently is 2.7x (fastembed pads each batch to its longest member; 184-char median vs 2000-char max). With noise sections that is 61 min -> 20 min on a 6,140-page wiki at half the workers. Fandom portable-infoboxes are transformed to tables before markdownify — otherwise stat values survive but detach from their labels. 100% branch coverage, ruff clean. Co-Authored-By: Claude Opus 5 (1M context) diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..058776b --- /dev/null +++ b/.gitignore @@ -0,0 +1,19 @@ +# Generated / large / regenerable artifacts +.venv/ +corpus/ +index/ +manifest.json + +# Python +__pycache__/ +*.pyc +.pytest_cache/ +.coverage +htmlcov/ +.ruff_cache/ + +# fastembed / model cache (if it lands locally) +.fastembed_cache/ + +# Build artifacts +*.egg-info/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..d1aa637 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Krzysztof Rudnicki + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/NOTICE.md b/NOTICE.md new file mode 100644 index 0000000..8325b2b --- /dev/null +++ b/NOTICE.md @@ -0,0 +1,18 @@ +# NOTICE + +This project creates a **personal, offline** searchable mirror of the +[Crusader Kings III Wiki](https://ck3.paradoxwikis.com) for use as a local +retrieval/RAG knowledge base. + +- Wiki text content is authored by the CK3 wiki community and is generally + available under [CC BY-SA 4.0](https://creativecommons.org/licenses/by-sa/4.0/). + Original content and attribution belong to the wiki and its contributors. +- Crusader Kings III and related names are trademarks of Paradox Interactive. + This project is unofficial and not affiliated with or endorsed by Paradox. +- The scraper is intended for **personal, non-redistributive** use. It fetches + through the public MediaWiki API with a descriptive User-Agent, `maxlag`, + single-flight ~1 req/s rate limiting, and exponential backoff — following + [MediaWiki API:Etiquette](https://www.mediawiki.org/wiki/API:Etiquette). +- The generated `corpus/` and `index/` are **not** committed to git and are not + meant to be republished. Each mirrored page retains a link back to its source + URL. diff --git a/README.md b/README.md new file mode 100644 index 0000000..1133400 --- /dev/null +++ b/README.md @@ -0,0 +1,137 @@ +# wiki-kb + +Turn a wiki into a local, offline, searchable knowledge base your LLM can query as an +MCP tool. + +**scrape** (MediaWiki API, or sitemap + HTML extraction for wikis with no API) → +**corpus/** markdown → **index** (SQLite FTS5 + local `bge-small` embeddings) → +**hybrid search** (BM25 + cosine, fused with Reciprocal Rank Fusion) → **MCP stdio +server**. + +Everything runs locally. No API keys, no cloud, no torch — `fastembed` uses ONNX on +CPU. + +## Quick start + +```bash +./install.sh # venv (Python 3.12) + deps +export WIKI_KB_CONTACT="you@example.com" # wiki etiquette requires a real contact + +python scripts/detect_wiki.py https://some.wiki.example.com # what kind of wiki is it? +./run.sh --profile ck3 scrape # mirror (resumable, ~1 req/s) +./run.sh --profile ck3 index # build the search index +./run.sh --profile ck3 search "domain limit by rank" +./run.sh check # ruff + pytest (100% branch coverage) +``` + +Register with Claude Code: + +```bash +WIKI_KB_NAMESPACE=ck3 claude mcp add -s user ck3 -- "$PWD/.venv/bin/python" -m wikikb.server +``` + +## Any wiki, not just MediaWiki + +`scripts/detect_wiki.py ` probes `/api.php`, then `/w/api.php`, then +`/sitemap.xml`, and prints the source kind, endpoint, article URL pattern, engine, +page count, licence, and the env vars to configure with. Verified against three +engines: + +| Host | Source | Endpoint | Article URL | +|---|---|---|---| +| ck3.paradoxwikis.com | mediawiki | `/api.php` | `{host}/{title}` | +| kingdom-come-deliverance.fandom.com | mediawiki | `/api.php` | `{host}/wiki/{title}` | +| kingdomcomedeliverance2.wiki.fextralife.com | **html** | `/sitemap.xml` | `{host}/{Title+With+Plus}` | + +Note the API path and the article URL vary *independently* — Fandom serves the API at +`/api.php` but articles at `/wiki/{Title}`. Hardcoding one layout breaks on the next +wiki. + +An API is always preferred: it gives a real page list, server-rendered HTML with +templates expanded, categories, revision ids, and `maxlag` so the server can push +back. The HTML fallback genuinely gives up all of those — `detect_wiki.py` says so +rather than pretending otherwise. + +## Profiles + +`config.py` reads every setting from `WIKI_KB_*` environment variables, so a profile +is just an env file in `profiles/` — no plugin system. Each sets its own +`WIKI_KB_ROOT`, which is what keeps several wikis' corpora and indexes apart. + +See `profiles/README.md`. Shipped: `ck3`, `kcd`, `kcd-fextralife`. + +## Things that are per-wiki and must be measured, not assumed + +- **Are subpages noise?** On the CK3 wiki `A/B` means a total-conversion mod (reject). + On the KCD wiki it means *the sequel* — `Alchemy/KCD2` — and rejecting subpages + deletes 441 pages of exactly the content you want (keep). One API pass decides it: + `Counter(t.rsplit("/", 1)[-1] for t in titles if "/" in t)`. +- **Which sections are boilerplate?** `WIKI_KB_NOISE_SECTIONS` drops them from the + index while `*_get_page` still serves the whole page. On the KCD wiki the "Other + languages" tables were 7% of chunks and matched English queries because they repeat + every page's title in eight languages. +- **Does the wiki cover two editions at once?** `WIKI_KB_VARIANTS="kcd2:KCD2,kcd1:KCD"` + tags every page and lets searches scope to one. A scope excludes the *other* edition + but always keeps pages tagged `both`/`unknown` — filtering *to* one tag is a recall + hole, since ~9% of KCD pages carry no edition category yet matter to both. + +Every search hit carries its `source` and `variant`, so a result can never be silently +attributed to the wrong wiki or the wrong edition. + +## Build performance + +The index build is CPU-heavy. Two optimisations matter, and the first is large: + +- **Length bucketing** (`indexer.embed_efficiently`) — fastembed pads every batch to + its longest member. Wiki chunks are wildly uneven (184-char median, 2000-char max, + 53% under 200 chars), so most of the compute was padding. Sorting by length measured + **2.7x faster**. The sort is undone before the vectors are returned: embedding row + `i` must stay aligned with chunk rowid `i+1`. +- **Noise sections** cut ~17% of chunks outright. + +Together: 61 min → 20 min on a 6,140-page wiki, at *half* the worker count. + +**`nice` does not protect a running game.** Measured with a GPU-bound game: an +8-worker build dropped GPU utilisation from 96.6% to ~50% while the game's *total* CPU +barely moved — it was never denied cycles, its render thread was delayed getting them. +Core reservation did not measurably help either. The real lever is finishing sooner. + +```bash +WIKI_KB_RESERVED_CORES="0-3" # cores to keep free (set "" to use all) +WIKI_KB_NICE=10 +WIKI_KB_EMBED_PARALLEL=4 # 4 workers ~= 20 min on 6k pages; 8 is worse for games +``` + +`build_index` writes only at the end and is not resumable — make sure a build fits in +the time you have. + +## Etiquette and licensing + +- Descriptive User-Agent with a real contact (`WIKI_KB_CONTACT`), `maxlag`, ~1 req/s + single-flight, exponential backoff, resumable. Per + [MediaWiki API:Etiquette](https://www.mediawiki.org/wiki/API:Etiquette). +- Cloudflare bot challenges arrive as **HTTP 200 with an HTML body**, so they pass + `raise_for_status()`. They are treated as retryable. The User-Agent is never spoofed + to evade one. +- `corpus/`, `index/` and `manifest.json` are gitignored. Mirrored content belongs to + the wiki and its contributors — most are CC BY-SA. Each mirrored page keeps a link + back to its source URL. Check a site's `robots.txt` and terms before mirroring it; + `detect_wiki.py` surfaces AI-crawler directives it finds. + +## Layout + +``` +wikikb/ + config.py endpoints, paths, noise filters, variants, CPU settings — all env-driven + sources.py MediaWikiSource | SitemapHtmlSource + host detection + scraper.py politeness, noise filtering, resume, corpus + manifest + html2md.py parse-HTML -> markdown (incl. Fandom portable-infobox -> tables) + chunker.py heading-scoped, overlapping windows + embed.py fastembed bge-small, L2-normalised + indexer.py FTS5 table + row-aligned embeddings.npy + search.py BM25 (OR terms) + cosine -> RRF fusion + variant.py edition tagging for wikis covering two editions + server.py FastMCP stdio: _search / _get_page / _list_pages +profiles/ one env file per wiki +scripts/ detect_wiki.py + the three acceptance checks +``` diff --git a/install.sh b/install.sh new file mode 100755 index 0000000..4b50129 --- /dev/null +++ b/install.sh @@ -0,0 +1,50 @@ +#!/bin/bash + +# ============================================================================ +# install.sh — create an isolated venv (Python 3.12) and install wiki-kb +# ============================================================================ + +set -euo pipefail + +readonly SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +readonly VENV_DIR="${SCRIPT_DIR}/.venv" + +# Python 3.12 is pinned because fastembed/onnxruntime wheels lag on newer +# interpreters. Prefer pyenv's 3.12, fall back to any python3.12 on PATH. +find_python() { + local candidates=( + "${HOME}/.pyenv/versions/3.12.8/bin/python3.12" + "$(command -v python3.12 || true)" + ) + local p + for p in "${candidates[@]}"; do + if [[ -n "$p" && -x "$p" ]]; then + printf '%s\n' "$p" + return 0 + fi + done + echo "Error: python3.12 not found (need it for fastembed/onnxruntime wheels)." >&2 + echo "Install it, e.g.: pyenv install 3.12.8" >&2 + return 1 +} + +main() { + local py + py="$(find_python)" + echo "Using interpreter: $py ($("$py" --version))" + + if [[ ! -d "$VENV_DIR" ]]; then + echo "Creating venv at $VENV_DIR" + "$py" -m venv "$VENV_DIR" + fi + + # shellcheck disable=SC1091 + source "${VENV_DIR}/bin/activate" + python -m pip install --quiet --upgrade pip + echo "Installing wiki-kb (+dev extras)..." + python -m pip install --quiet -e "${SCRIPT_DIR}[dev]" + + echo "Done. Activate with: source ${VENV_DIR}/bin/activate" +} + +main "$@" diff --git a/profiles/README.md b/profiles/README.md new file mode 100644 index 0000000..326862c --- /dev/null +++ b/profiles/README.md @@ -0,0 +1,16 @@ +# Wiki profiles + +`config.py` reads every setting from `WIKI_KB_*` environment variables, so a profile +is just an env file — no plugin system, no registry, no code. + +```bash +./run.sh --profile kcd scrape +./run.sh --profile kcd search "best longsword stats" +``` + +Create a new one by running `python scripts/detect_wiki.py `, which prints the +host/API/page-URL lines to paste in, then adding the per-wiki noise filters you find +by inspecting the corpus (see the two examples here). + +Each profile MUST set `WIKI_KB_ROOT` to its own directory — that is what keeps several +wikis' corpora, indexes and manifests from overwriting each other. diff --git a/profiles/ck3.env b/profiles/ck3.env new file mode 100644 index 0000000..a321ca3 --- /dev/null +++ b/profiles/ck3.env @@ -0,0 +1,10 @@ +# Crusader Kings III wiki — MediaWiki, articles at {host}/{Title}. +WIKI_KB_HOST=https://ck3.paradoxwikis.com +WIKI_KB_NAMESPACE=ck3 +WIKI_KB_ROOT=$HOME/.local/share/wiki-kb/ck3 + +# Total-conversion mods live in ns0 and swamp retrieval (~33% of the corpus). +# On THIS wiki they are subpages, so rejecting "/" titles is correct — it is NOT +# correct in general; see profiles/kcd.env for a wiki where "/" means the sequel. +WIKI_KB_SUBPAGES=reject +WIKI_KB_MOD_PREFIXES=ck3agot,lotr,elder kings,after the end,the fallen eagle,fallen eagle,godherja,princes of darkness,rice,realms in exile,a game of thrones,when the world stopped making sense,more bookmarks diff --git a/profiles/kcd-fextralife.env b/profiles/kcd-fextralife.env new file mode 100644 index 0000000..36b0d6d --- /dev/null +++ b/profiles/kcd-fextralife.env @@ -0,0 +1,10 @@ +# Fextralife KCD2 wiki — NO API (custom engine). Sitemap + HTML content extraction. +# Loses revision ids, maxlag and categories; see references/build-guide.md. +# Check the site's robots.txt and terms before enabling this one. +WIKI_KB_HOST=https://kingdomcomedeliverance2.wiki.fextralife.com +WIKI_KB_SOURCE=html +WIKI_KB_SITEMAP=https://kingdomcomedeliverance2.wiki.fextralife.com/sitemap.xml +WIKI_KB_NAMESPACE=kcd +WIKI_KB_ROOT=$HOME/.local/share/wiki-kb/kcd +WIKI_KB_SUBPAGES=keep +WIKI_KB_MOD_PREFIXES= diff --git a/profiles/kcd.env b/profiles/kcd.env new file mode 100644 index 0000000..f693a11 --- /dev/null +++ b/profiles/kcd.env @@ -0,0 +1,22 @@ +# Kingdom Come: Deliverance wiki (Fandom) — MediaWiki, articles at {host}/wiki/{Title}. +# Covers BOTH games in one namespace; the sequel's pages are "/KCD2" subpages. +WIKI_KB_HOST=https://kingdom-come-deliverance.fandom.com +WIKI_KB_API_URL=https://kingdom-come-deliverance.fandom.com/api.php +WIKI_KB_PAGE_URL=https://kingdom-come-deliverance.fandom.com/wiki/{title} +WIKI_KB_NAMESPACE=kcd +WIKI_KB_ROOT=$HOME/.local/share/wiki-kb/kcd + +# Subpages are STRUCTURAL here ("Alchemy/KCD2" is the sequel's Alchemy page), so +# rejecting them outright would delete all 441 KCD2 pages. Drop only "/loc", which +# holds the same in-game book text repeated in ~8 languages (371 pages, ~32k chars each). +WIKI_KB_SUBPAGES=keep +WIKI_KB_NOISE_TAILS=loc +WIKI_KB_MOD_PREFIXES= + +# Per-page tables of localized UI names: 7% of chunks, and they match English queries +# because they repeat every page's title in every language. +WIKI_KB_NOISE_SECTIONS=other languages + +# Two-edition wiki: tag each page kcd1/kcd2/both/unknown and default searches to kcd2. +WIKI_KB_VARIANTS=kcd2:KCD2,kcd1:KCD +WIKI_KB_DEFAULT_VARIANT=kcd2 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f555b46 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,48 @@ +[project] +name = "wiki-kb" +version = "0.1.0" +description = "Local searchable mirror of the Crusader Kings III wiki (RAG + MCP)." +requires-python = ">=3.12,<3.13" +dependencies = [ + "httpx>=0.27", + "beautifulsoup4>=4.12", + "markdownify>=0.13", + "numpy>=1.26", + "fastembed>=0.3", + "mcp>=1.2", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0", + "pytest-cov>=5.0", + "respx>=0.21", + "ruff>=0.6", +] + +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[tool.setuptools] +packages = ["wikikb"] + +[tool.ruff] +line-length = 100 +target-version = "py312" + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B", "SIM", "PTH", "RUF"] + +[tool.pytest.ini_options] +addopts = "-q" +testpaths = ["tests"] + +[tool.coverage.run] +branch = true +source = ["wikikb"] + +[tool.coverage.report] +show_missing = true +fail_under = 100 +exclude_lines = ["if __name__ == .__main__.:", "pragma: no cover", "\\.\\.\\."] diff --git a/run.sh b/run.sh new file mode 100755 index 0000000..5f968b9 --- /dev/null +++ b/run.sh @@ -0,0 +1,78 @@ +#!/bin/bash + +# ============================================================================ +# run.sh — orchestrate a wiki knowledge base +# ./run.sh install create venv + install deps +# ./run.sh scrape [--limit N] mirror the wiki into corpus/ +# ./run.sh index build FTS5 + embeddings index +# ./run.sh search "" hybrid search from the CLI +# ./run.sh serve run the MCP server (stdio) +# ./run.sh check ruff + pytest (with coverage) +# ============================================================================ + +set -euo pipefail + +readonly SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +readonly VENV_PY="${SCRIPT_DIR}/.venv/bin/python" + +need_venv() { + if [[ ! -x "$VENV_PY" ]]; then + echo "venv missing — run: ./run.sh install" >&2 + exit 1 + fi +} + +usage() { + grep -E '^# \./run\.sh' "${BASH_SOURCE[0]}" | sed 's/^# //' + exit "${1:-0}" +} + +# A profile is just an env file (see profiles/README.md) — config.py reads every +# setting from WIKI_KB_* variables, so no plugin machinery is needed. +# A profile is just an env file (see profiles/README.md) — config.py reads every +# setting from WIKI_KB_* variables, so no plugin machinery is needed. Parsed line by +# line rather than sourced: values legitimately contain spaces ("other languages"), +# which `source` would try to execute. +load_profile() { + local name="$1" file="${SCRIPT_DIR}/profiles/${1}.env" line key value + [[ -f "$file" ]] || { + echo "No such profile: $name (available: $(cd "${SCRIPT_DIR}/profiles" && \ + ls *.env 2>/dev/null | sed 's/\.env$//' | tr '\n' ' '))" >&2 + exit 1 + } + while IFS= read -r line || [[ -n "$line" ]]; do + [[ -z "$line" || "$line" == \#* ]] && continue + key="${line%%=*}" + value="${line#*=}" + value="${value//\$HOME/$HOME}" # only $HOME is expanded, deliberately + export "$key=$value" + done < "$file" + echo "profile: $name -> ${WIKI_KB_HOST:-?} (root ${WIKI_KB_ROOT:-?})" >&2 +} + +main() { + # --profile may precede the command; it only sets env vars. + if [[ "${1:-}" == "--profile" ]]; then + [[ $# -ge 2 ]] || { echo "--profile needs a name" >&2; exit 1; } + load_profile "$2" + shift 2 + fi + local cmd="${1:-}" + [[ $# -gt 0 ]] && shift || true + case "$cmd" in + install) exec "${SCRIPT_DIR}/install.sh" ;; + scrape) need_venv; exec "$VENV_PY" -u -m wikikb.scraper "$@" ;; + index) need_venv; exec "$VENV_PY" -u -m wikikb.indexer "$@" ;; + search) need_venv; exec "$VENV_PY" -u -m wikikb.search "$@" ;; + serve) need_venv; exec "$VENV_PY" -u -m wikikb.server "$@" ;; + check) + need_venv + "$VENV_PY" -m ruff check "${SCRIPT_DIR}/wikikb" "${SCRIPT_DIR}/tests" + "$VENV_PY" -m pytest --cov=wikikb --cov-report=term-missing "${SCRIPT_DIR}/tests" + ;; + ""|-h|--help) usage 0 ;; + *) echo "Unknown command: $cmd" >&2; usage 1 ;; + esac +} + +main "$@" diff --git a/scripts/detect_wiki.py b/scripts/detect_wiki.py new file mode 100644 index 0000000..c7e499a --- /dev/null +++ b/scripts/detect_wiki.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python +"""Probe a wiki host and report how (or whether) it can be mirrored. + +Phase 0 of the build, as a command: + + python scripts/detect_wiki.py https://some.wiki.example.com + +Reports the source kind, the endpoint, the article URL pattern, and — for +MediaWiki — the engine version and page count. Exits non-zero if the host cannot +be mirrored at all, so it can gate a scaffold. +""" + +from __future__ import annotations + +import sys + +import httpx + +from wikikb import config +from wikikb.sources import MEDIAWIKI, SitemapHtmlSource, detect_source_kind + + +def main(argv: list[str] | None = None) -> int: + args = sys.argv[1:] if argv is None else argv + if not args: + print("usage: detect_wiki.py ", file=sys.stderr) + return 2 + host = args[0].rstrip("/") + + with httpx.Client( + headers={"User-Agent": config.USER_AGENT}, timeout=30, follow_redirects=True + ) as client: + try: + kind, endpoint = detect_source_kind(client.get, host) + except RuntimeError as exc: + print(f"[FAIL] {exc}") + return 1 + + print(f"[ OK ] source kind : {kind}") + print(f" endpoint : {endpoint}") + + if kind == MEDIAWIKI: + info = client.get( + f"{endpoint}?action=query&meta=siteinfo&siprop=general|statistics&format=json" + ).json()["query"] + general, stats = info["general"], info["statistics"] + base = general.get("base", "") + pattern = "{host}/wiki/{title}" if "/wiki/" in base else "{host}/{title}" + print(f" engine : {general.get('generator')}") + print(f" content : {stats.get('articles')} articles") + print(f" article URL : {pattern}") + print(f" licence : {info.get('rightsinfo', {}).get('text', 'unknown')}") + print("\nScaffold with:") + print(f" WIKI_KB_HOST={host} WIKI_KB_API_URL={endpoint}") + if pattern.endswith("/wiki/{title}"): + print(f" WIKI_KB_PAGE_URL='{host}/wiki/{{title}}'") + else: + urls = list(SitemapHtmlSource(client.get, sitemap_url=endpoint).iter_titles()) + print(f" content : {len(urls)} sitemap URLs") + print(f" sample : {urls[:5]}") + print("\n[WARN] No API. Revision ids and maxlag are unavailable, so fidelity") + print(" checks fall back to content diffing and rate limiting is purely") + print(" self-imposed. Check the site's robots.txt and terms before use.") + print("\nScaffold with:") + print(f" WIKI_KB_HOST={host} WIKI_KB_SOURCE=html WIKI_KB_SITEMAP={endpoint}") + + robots = client.get(f"{host}/robots.txt") + if robots.status_code == 200 and "gptbot" in robots.text.lower(): + print("\n[NOTE] robots.txt names an AI crawler (e.g. GPTBot). It does not name") + print(" this tool, but it is the operators' clearest statement about AI") + print(" ingestion of their content. Your call — read it before scraping.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/verify_fidelity.py b/scripts/verify_fidelity.py new file mode 100644 index 0000000..bef7acd --- /dev/null +++ b/scripts/verify_fidelity.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python +"""Verify our local mirror matches the LIVE wiki (no corruption / staleness). + +For each key page it re-fetches the live page, runs it through the SAME +HTML->markdown pipeline used at scrape time, and checks: + 1. the freshly-converted live body is byte-identical to our stored copy + (proves faithful capture + no drift since the scrape), and + 2. specific factual tokens (trait names, modifier numbers) appear in BOTH the + live page and our local copy — printed side by side so "the same" is visible. + +DONE = every page's key facts match in live & local (byte-identical is reported +too, and flagged as WARN if the wiki was edited since the scrape). +""" + +from __future__ import annotations + +import difflib +import sys + +import httpx + +from wikikb import config, html2md +from wikikb.scraper import Throttle, fetch_page, load_manifest, sanitize_filename + +# page -> factual tokens that must appear identically in live + local +FACTS: dict[str, list[str]] = { + "Traits": ["Genius", "Intelligent", "Herculean", "Beautiful", "+5"], + "Attributes": ["Diplomacy", "Stewardship", "Prowess", "Domain Taxes"], + "Succession": ["Primogeniture", "Partition"], +} + + +def _stored_text(title: str) -> str: + entry = load_manifest().get(title, {}) + filename = entry.get("filename") or sanitize_filename(title) + return (config.CORPUS_DIR / filename).read_text(encoding="utf-8") + + +def _line_with(text: str, token: str) -> str: + return next((ln.strip() for ln in text.splitlines() if token in ln), "") + + +def main() -> int: + ok = True + with httpx.Client(headers={"User-Agent": config.USER_AGENT}, timeout=30) as client: + throttle = Throttle() + for title, facts in FACTS.items(): + page = fetch_page(client, title, throttle=throttle) + if page is None: + print(f"[FAIL] {title}: could not fetch live page") + ok = False + continue + stored = _stored_text(page.title) + fresh = html2md.html_to_markdown(page.html) + + identical = fresh in stored # stored = frontmatter + "# Title" + this body + ratio = difflib.SequenceMatcher(None, fresh, stored).ratio() + tag = "PASS" if identical else "WARN" + note = "byte-identical to stored copy" if identical else f"differs (ratio {ratio:.3f})" + print(f"[{tag}] {page.title} (rev {page.revid}): freshly-converted live body {note}") + + for fact in facts: + same = fact in fresh and fact in stored + ok &= same + print( + f" [{'PASS' if same else 'FAIL'}] {fact!r} in live & local -> {same}" + f" | live: {_line_with(fresh, fact)[:80]!r}" + ) + + print(f"\nFIDELITY VERIFICATION: {'PASS' if ok else 'FAIL'}") + return 0 if ok else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/verify_mcp.py b/scripts/verify_mcp.py new file mode 100644 index 0000000..732d6b1 --- /dev/null +++ b/scripts/verify_mcp.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python +"""Verify the ck3 MCP server end-to-end over real stdio JSON-RPC. + +DONE = the server starts, completes the MCP handshake, advertises the three +tools, and returns sensible results for wiki_search and wiki_get_page. Exits +non-zero on any failure. Uses the same interface Claude Code will use, so this +confirms the server works without needing a session reload. +""" + +from __future__ import annotations + +import sys + +import anyio +from mcp import ClientSession, StdioServerParameters +from mcp.client.stdio import stdio_client + + +def _text(result: object) -> str: + """Flatten a CallToolResult into a searchable string.""" + parts: list[str] = [] + structured = getattr(result, "structuredContent", None) + if structured is not None: + parts.append(str(structured)) + for block in getattr(result, "content", []) or []: + if hasattr(block, "text"): + parts.append(block.text) + return " ".join(parts) + + +async def run() -> int: + params = StdioServerParameters(command=sys.executable, args=["-m", "wikikb.server"]) + ok = True + async with stdio_client(params) as (read, write), ClientSession(read, write) as session: + await session.initialize() + + tools = {t.name for t in (await session.list_tools()).tools} + want = {"wiki_search", "wiki_get_page", "wiki_list_pages"} + t_ok = want <= tools + ok &= t_ok + print(f"[{'PASS' if t_ok else 'FAIL'}] tools/list -> {sorted(tools)}") + + search_text = _text(await session.call_tool("wiki_search", {"query": "domain limit rank"})) + s_ok = ( + any(k in search_text for k in ("Titles", "Attributes")) + and "domain" in search_text.lower() + ) + ok &= s_ok + print(f"[{'PASS' if s_ok else 'FAIL'}] wiki_search -> {search_text[:80]!r}...") + + page_text = _text(await session.call_tool("wiki_get_page", {"title": "Traits"})) + g_ok = "Genius" in page_text and "Herculean" in page_text + ok &= g_ok + print( + f"[{'PASS' if g_ok else 'FAIL'}] wiki_get_page('Traits') has Genius+Herculean -> {g_ok}" + ) + + list_text = _text(await session.call_tool("wiki_list_pages", {"prefix": "Tr"})) + l_ok = "Traits" in list_text + ok &= l_ok + print(f"[{'PASS' if l_ok else 'FAIL'}] wiki_list_pages('Tr') includes Traits -> {l_ok}") + + print(f"\nMCP VERIFICATION: {'PASS' if ok else 'FAIL'}") + return 0 if ok else 1 + + +if __name__ == "__main__": + sys.exit(anyio.run(run)) diff --git a/scripts/verify_search.py b/scripts/verify_search.py new file mode 100644 index 0000000..2e9b6f7 --- /dev/null +++ b/scripts/verify_search.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python +"""Verify search quality against concrete acceptance criteria. + +DONE = every case's expected wiki page appears in the top-3 results, AND no +mod/subpage title appears in the top-5 of any probe query. Exits non-zero on +any failure so it can gate a build. +""" + +from __future__ import annotations + +import sys + +from wikikb.scraper import is_core_title +from wikikb.search import search + +# (query, one-of-these titles must be in the top 3) +CASES: list[tuple[str, set[str]]] = [ + ("domain limit rank", {"Titles", "Attributes", "Domain"}), + ("genius intelligent trait skill bonus", {"Traits"}), + ("primogeniture succession heir inheritance", {"Succession"}), + ("congenital beauty attraction opinion", {"Traits"}), + ("health aging death lifespan", {"Attributes", "Health"}), +] + +# queries whose top-5 must contain NO mod/subpage results +NO_MOD_QUERIES = [ + "domain limit", + "succession law", + "development growth", + "knight martial army", + "culture names", +] + + +def main() -> int: + ok = True + print("== relevance (expected page in top-3) ==") + for query, expected in CASES: + top3 = [r.title for r in search(query, k=5)[:3]] + hit = any(t in expected for t in top3) + ok &= hit + print( + f" [{'PASS' if hit else 'FAIL'}] {query!r} -> {top3} (want one of {sorted(expected)})" + ) + + print("== no mod/subpage pollution (top-5) ==") + for query in NO_MOD_QUERIES: + mods = [r.title for r in search(query, k=5) if not is_core_title(r.title)] + clean = not mods + ok &= clean + print(f" [{'PASS' if clean else 'FAIL'}] {query!r} -> {'clean' if clean else mods}") + + print(f"\nSEARCH VERIFICATION: {'PASS' if ok else 'FAIL'}") + return 0 if ok else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..acda393 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,47 @@ +"""Shared fixtures: temp KB paths and a deterministic fake embedding model.""" + +from __future__ import annotations + +import hashlib +from collections.abc import Iterable +from pathlib import Path + +import numpy as np +import pytest + +from wikikb import config + + +@pytest.fixture +def tmp_kb(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """Point all config paths at an isolated temp tree.""" + corpus = tmp_path / "corpus" + index = tmp_path / "index" + corpus.mkdir() + index.mkdir() + monkeypatch.setattr(config, "CORPUS_DIR", corpus) + monkeypatch.setattr(config, "INDEX_DIR", index) + monkeypatch.setattr(config, "DB_PATH", index / "wiki.db") + monkeypatch.setattr(config, "EMBEDDINGS_PATH", index / "embeddings.npy") + monkeypatch.setattr(config, "MANIFEST_PATH", tmp_path / "manifest.json") + return tmp_path + + +class FakeEmbedding: + """Deterministic, offline stand-in for fastembed's TextEmbedding.""" + + def _vec(self, text: str) -> np.ndarray: + digest = hashlib.sha256(text.encode()).digest() + raw = (digest * (config.EMBED_DIM // len(digest) + 1))[: config.EMBED_DIM] + return np.frombuffer(raw, dtype=np.uint8).astype(np.float32) + + def embed(self, texts: Iterable[str], **_kwargs: object) -> list[np.ndarray]: + return [self._vec(t) for t in texts] + + def query_embed(self, texts: Iterable[str], **_kwargs: object) -> list[np.ndarray]: + return [self._vec(t) for t in texts] + + +@pytest.fixture +def fake_model() -> FakeEmbedding: + return FakeEmbedding() diff --git a/tests/test_chunker.py b/tests/test_chunker.py new file mode 100644 index 0000000..3bb0d79 --- /dev/null +++ b/tests/test_chunker.py @@ -0,0 +1,83 @@ +"""Tests for markdown chunking.""" + +from __future__ import annotations + +from wikikb import chunker + +_DOC = """--- +title: "Traits" +url: https://ck3.paradoxwikis.com/Traits +revid: 1 +categories: ["Character"] +--- + +# Traits + +Intro paragraph. + +## Congenital + +Genius gives +5 to all skills. + +## Physique + +Herculean gives +1 health. +""" + + +def test_parse_frontmatter_extracts_meta_and_body() -> None: + meta, body = chunker.parse_frontmatter(_DOC) + assert meta["title"] == "Traits" + assert meta["url"].endswith("/Traits") + assert body.lstrip().startswith("# Traits") + + +def test_parse_frontmatter_absent() -> None: + meta, body = chunker.parse_frontmatter("no frontmatter here") + assert meta == {} + assert body == "no frontmatter here" + + +def test_chunk_document_sections() -> None: + chunks = chunker.chunk_document(_DOC) + sections = {c.section for c in chunks} + assert {"Traits", "Congenital", "Physique"} <= sections + assert all(c.title == "Traits" for c in chunks) + assert any("Genius" in c.text for c in chunks) + + +def test_window_splits_long_sections_with_overlap() -> None: + long = "# T\n\n" + ("word " * 2000) + chunks = chunker.chunk_document(long, target_chars=200, overlap_chars=50) + assert len(chunks) > 1 + assert all(len(c.text) <= 200 for c in chunks) + + +def test_window_short_text_single_chunk() -> None: + assert chunker._window("short", size=100, overlap=10) == ["short"] + + +def test_parse_frontmatter_ignores_lines_without_colon() -> None: + text = '---\ntitle: "X"\nnocolonhere\n---\n\nbody' + meta, _ = chunker.parse_frontmatter(text) + assert meta == {"title": "X"} + + +def test_split_sections_drops_empty_text_sections() -> None: + # "H1" has only blank lines under it -> filtered out; "Overview" and "H2" kept. + body = "text0\n# H1\n\n\n# H2\nx" + sections = chunker._split_sections(body) + headings = [h for h, _ in sections] + assert headings == ["Overview", "H2"] + + +def test_split_sections_body_ending_with_heading() -> None: + # trailing heading leaves the buffer empty at loop end (no final flush) + assert chunker._split_sections("text\n# H") == [("Overview", "text")] + + +def test_chunk_document_skips_whitespace_only_windows() -> None: + doc = "# H\n\nA" + " " * 300 + "B" + chunks = chunker.chunk_document(doc, target_chars=100, overlap_chars=10) + assert chunks # some windows are all-whitespace and are skipped + assert all(c.text.strip() for c in chunks) diff --git a/tests/test_cpu.py b/tests/test_cpu.py new file mode 100644 index 0000000..b45a70b --- /dev/null +++ b/tests/test_cpu.py @@ -0,0 +1,53 @@ +"""Tests for CPU isolation and core-spec parsing.""" + +from __future__ import annotations + +import os + +from wikikb import config, cpu + + +def test_parse_core_spec() -> None: + assert config._parse_core_spec("0-3", 24) == {0, 1, 2, 3} + assert config._parse_core_spec("8,10-11", 24) == {8, 10, 11} + assert config._parse_core_spec("", 24) == set() + assert config._parse_core_spec("20-100", 24) == {20, 21, 22, 23} # clamped + + +def test_allowed_cores_excludes_reserved() -> None: + assert config._allowed_cores(8, {0, 1}) == [2, 3, 4, 5, 6, 7] + + +def test_allowed_cores_falls_back_when_all_reserved() -> None: + assert config._allowed_cores(4, {0, 1, 2, 3}) == [0, 1, 2, 3] + + +def test_isolate_sets_affinity_and_nice(monkeypatch) -> None: + seen: dict[str, object] = {} + monkeypatch.setattr( + os, "sched_setaffinity", lambda pid, cores: seen.__setitem__("aff", (pid, set(cores))) + ) + monkeypatch.setattr(os, "nice", lambda inc: seen.__setitem__("nice", inc)) + assert cpu.isolate(allowed=[4, 5, 6], nice_increment=7) == [4, 5, 6] + assert seen["aff"] == (0, {4, 5, 6}) + assert seen["nice"] == 7 + + +def test_isolate_defaults_use_config(monkeypatch) -> None: + seen: dict[str, object] = {} + monkeypatch.setattr( + os, "sched_setaffinity", lambda _pid, cores: seen.__setitem__("c", set(cores)) + ) + monkeypatch.setattr(os, "nice", lambda inc: seen.__setitem__("n", inc)) + assert cpu.isolate() == list(config.ALLOWED_CORES) + assert seen["c"] == set(config.ALLOWED_CORES) + assert seen["n"] == config.NICE_INCREMENT + + +def test_isolate_no_affinity_and_zero_nice(monkeypatch) -> None: + monkeypatch.delattr(os, "sched_setaffinity", raising=False) + called = {"nice": False} + monkeypatch.setattr(os, "nice", lambda _inc: called.__setitem__("nice", True)) + # nice_increment 0 -> nice() skipped; missing sched_setaffinity -> affinity skipped + assert cpu.isolate(allowed=[0], nice_increment=0) == [0] + assert called["nice"] is False diff --git a/tests/test_embed.py b/tests/test_embed.py new file mode 100644 index 0000000..734e3db --- /dev/null +++ b/tests/test_embed.py @@ -0,0 +1,36 @@ +"""Tests for the embedding wrapper (using the fake model fixture).""" + +from __future__ import annotations + +import numpy as np + +from wikikb import config, embed + + +def test_embed_passages_shape_and_normalization(fake_model) -> None: + vectors = embed.embed_passages(["a", "b", "c"], model=fake_model) + assert vectors.shape == (3, config.EMBED_DIM) + norms = np.linalg.norm(vectors, axis=1) + assert np.allclose(norms, 1.0, atol=1e-5) + + +def test_embed_passages_empty(fake_model) -> None: + vectors = embed.embed_passages([], model=fake_model) + assert vectors.shape == (0, config.EMBED_DIM) + + +def test_embed_query_normalized(fake_model) -> None: + vec = embed.embed_query("domain limit", model=fake_model) + assert vec.shape == (config.EMBED_DIM,) + assert np.isclose(np.linalg.norm(vec), 1.0, atol=1e-5) + + +def test_get_model_returns_cached(monkeypatch, fake_model) -> None: + monkeypatch.setattr(embed, "_model", fake_model) + assert embed.get_model() is fake_model + + +def test_normalize_handles_zero_vector() -> None: + zero = np.zeros((1, config.EMBED_DIM), dtype=np.float32) + out = embed._normalize(zero) + assert np.all(out == 0.0) # zero norm replaced by 1, stays zero diff --git a/tests/test_html2md.py b/tests/test_html2md.py new file mode 100644 index 0000000..5022804 --- /dev/null +++ b/tests/test_html2md.py @@ -0,0 +1,181 @@ +"""Tests for HTML -> markdown conversion.""" + +from __future__ import annotations + +from wikikb import html2md + +_HTML = """ +
+
[edit]
+

A count can hold up to 4 counties.

+ +
TraitDiplomacy
Genius+5
+
navigation
+
+""" + + +def test_clean_html_strips_chrome_keeps_content() -> None: + cleaned = html2md.clean_html(_HTML) + assert "editsection" not in cleaned + assert "navbox" not in cleaned + assert "count" in cleaned + + +def test_clean_html_without_parser_output_wrapper() -> None: + cleaned = html2md.clean_html("

bare

") + assert "bare" in cleaned + + +def test_html_to_markdown_preserves_table_numbers() -> None: + md = html2md.html_to_markdown(_HTML) + assert "Genius" in md + assert "+5" in md + assert "\n\n\n" not in md # blank lines collapsed + + +def test_build_document_frontmatter_and_quote_escaping() -> None: + doc = html2md.build_document( + title='War of the "Roses"', + url="https://wiki.example/War", + revid=42, + categories=["Warfare", "1.19"], + html="

hello

", + ) + assert doc.startswith("---\n") + assert 'title: "War of the \\"Roses\\""' in doc + assert "revid: 42" in doc + assert 'categories: ["Warfare", "1.19"]' in doc + assert "# War of the" in doc + assert "hello" in doc + + +# A cut-down Fandom PortableInfobox: labels are

siblings of their values, not +# table cells, and the tier variants are inlined behind tabs. +_INFOBOX = """ +
+ +

The lead sentence.

+
+""" + + +def test_infobox_becomes_a_table_with_labels_bound_to_values() -> None: + md = html2md.html_to_markdown(_INFOBOX) + # The point of the transform: not merely that "74" survives, but that it stays + # attached to "Stab dmg." Plain markdownify emits headings then orphaned numbers. + assert "| Stab dmg. | 74 |" in md + assert "| Slash dmg. | 71 |" in md + assert "| Classification | Weapons |" in md + + +def test_infobox_tier_tabs_are_labelled() -> None: + md = html2md.html_to_markdown(_INFOBOX) + # Without the tab label, both tiers' stat blocks are indistinguishable. + assert "## Infobox: Attributes (I)" in md + assert "## Infobox: Attributes (II)" in md + assert "| Stab dmg. | 92 |" in md + + +def test_infobox_lead_text_gets_its_own_section() -> None: + md = html2md.html_to_markdown(_INFOBOX) + # The lead paragraph must not be filed under the last infobox group. + assert md.index("## Overview") < md.index("The lead sentence.") + + +def test_infobox_without_a_group_header_is_named_plainly() -> None: + html = ( + '' + ) + md = html2md.html_to_markdown(html) + assert "## Infobox\n" in md + assert "| ID | 4 |" in md + + +def test_empty_infobox_is_dropped() -> None: + # Theme/dummy infoboxes hold no data and would otherwise leave a stray heading. + md = html2md.html_to_markdown( + '

body

' + ) + assert "Infobox" not in md + assert "body" in md + + +def test_mismatched_smart_group_keeps_the_pairs_that_line_up() -> None: + html = ( + '" + ) + md = html2md.html_to_markdown(html) + assert "| A | 1 |" in md + assert "| B |" not in md + + +def test_incomplete_pi_data_is_skipped_without_losing_the_rest() -> None: + # A hand-edited infobox row can carry a label with no value. Skip that row and + # keep going rather than emitting a label bound to nothing. + html = ( + '" + ) + md = html2md.html_to_markdown(html) + assert "| Weight | 3.9 |" in md + assert "Orphan" not in md + + +def test_page_tabber_switcher_is_stripped() -> None: + # It links to the other game's page; keeping it would put a bare "KCD" link at + # the top of every dual-game article. + html = ( + '
' + '
KCD
' + "

real content

" + ) + md = html2md.html_to_markdown(html) + assert "real content" in md + assert "/wiki/Alchemy" not in md diff --git a/tests/test_indexer.py b/tests/test_indexer.py new file mode 100644 index 0000000..91decdc --- /dev/null +++ b/tests/test_indexer.py @@ -0,0 +1,110 @@ +"""Tests for index building.""" + +from __future__ import annotations + +import numpy as np + +from wikikb import config, embed, indexer + +_DOC_A = """--- +title: "A" +url: https://x/A +revid: 1 +categories: [] +--- + +# A + +Genius trait gives plus five diplomacy. +""" + +_DOC_B = """--- +title: "B" +url: https://x/B +revid: 2 +categories: [] +--- + +# B + +Herculean physique gives health. +""" + + +def _embed_fn(fake_model): + return lambda texts: embed.embed_passages(texts, model=fake_model) + + +def test_collect_chunks_empty(tmp_kb) -> None: + assert indexer.collect_chunks(config.CORPUS_DIR) == [] + + +def test_build_index_creates_db_and_embeddings(tmp_kb, fake_model) -> None: + (config.CORPUS_DIR / "a.md").write_text(_DOC_A, encoding="utf-8") + (config.CORPUS_DIR / "b.md").write_text(_DOC_B, encoding="utf-8") + count = indexer.build_index(embed_fn=_embed_fn(fake_model), log=lambda _: None) + assert count >= 2 + assert config.DB_PATH.exists() + arr = np.load(config.EMBEDDINGS_PATH) + assert arr.shape == (count, config.EMBED_DIM) + + +def test_build_index_empty_corpus(tmp_kb, fake_model) -> None: + count = indexer.build_index(embed_fn=_embed_fn(fake_model), log=lambda _: None) + assert count == 0 + arr = np.load(config.EMBEDDINGS_PATH) + assert arr.shape == (0, config.EMBED_DIM) + + +def test_build_index_embeds_context_augmented_text(tmp_kb, fake_model) -> None: + (config.CORPUS_DIR / "a.md").write_text(_DOC_A, encoding="utf-8") + captured: list[str] = [] + + def capture_embed(texts): + captured.extend(texts) + return embed.embed_passages(texts, model=fake_model) + + indexer.build_index(embed_fn=capture_embed, log=lambda _: None) + assert captured + # embedded text is prefixed with the page title (context), body stays clean in DB + assert all(t.startswith("A. ") for t in captured) + + +def test_embed_efficiently_returns_vectors_in_original_order(fake_model) -> None: + # Worthless if it corrupts alignment: embedding row i must still be rowid i+1 + # after the internal length sort and dedupe. + texts = ["a" * 500, "bb", "a" * 500, "ccc", "d" * 900] + got = indexer.embed_efficiently(texts, _embed_fn(fake_model)) + want = embed.embed_passages(texts, model=fake_model) + np.testing.assert_allclose(got, want) + + +def test_embed_efficiently_embeds_each_distinct_text_once(fake_model) -> None: + seen: list[list[str]] = [] + + def counting_embed(texts): + seen.append(list(texts)) + return embed.embed_passages(texts, model=fake_model) + + got = indexer.embed_efficiently(["dup", "dup", "dup", "other"], counting_embed) + assert seen == [["dup", "other"]] # deduped, then sorted by length + assert got.shape[0] == 4 + np.testing.assert_allclose(got[0], got[2]) + + +def test_embed_efficiently_empty(fake_model) -> None: + assert indexer.embed_efficiently([], _embed_fn(fake_model)).shape == (0, config.EMBED_DIM) + + +def test_noise_sections_are_not_indexed(tmp_kb, fake_model, monkeypatch) -> None: + monkeypatch.setattr(config, "NOISE_SECTIONS", frozenset({"other languages"})) + doc = _DOC_A + "\n## Other languages\n\nlocalized UI names table\n" + (config.CORPUS_DIR / "a.md").write_text(doc, encoding="utf-8") + chunks = indexer.collect_chunks(config.CORPUS_DIR) + assert chunks + assert not any("localized UI names" in c.text for c in chunks) + + +def test_main_delegates(monkeypatch) -> None: + monkeypatch.setattr(indexer, "build_index", lambda: 3) + assert indexer.main() == 0 diff --git a/tests/test_scraper.py b/tests/test_scraper.py new file mode 100644 index 0000000..fe65157 --- /dev/null +++ b/tests/test_scraper.py @@ -0,0 +1,294 @@ +"""Tests for the MediaWiki scraper (HTTP mocked with respx).""" + +from __future__ import annotations + +import httpx +import pytest +import respx +from httpx import Response + +from wikikb import config, scraper +from wikikb.scraper import Throttle + + +def _noop(_: float) -> None: + return None + + +def _throttle() -> Throttle: + return Throttle(interval=0.0, sleep=_noop, clock=lambda: 0.0) + + +def test_throttle_sleeps_then_skips() -> None: + calls: list[float] = [] + times = iter([0.0, 0.0, 100.0, 100.0]) + throttle = Throttle(interval=10.0, sleep=calls.append, clock=lambda: next(times)) + throttle.wait() # now=0, last=0 -> delay 10 -> sleep + throttle.wait() # now=100, last=0 -> delay negative -> no sleep + assert calls == [10.0] + + +def test_sanitize_filename() -> None: + assert scraper.sanitize_filename("A/B: C?") == "A_B__C.md" + assert scraper.sanitize_filename("...") == "untitled.md" + + +def test_is_core_title() -> None: + assert scraper.is_core_title("Traits") + assert scraper.is_core_title("Domain limit") + assert not scraper.is_core_title("CK3AGOT/Dragon") # subpage + assert not scraper.is_core_title("LotR: Realms in Exile") # mod prefix + assert not scraper.is_core_title("Elder Kings 2 overview") # mod prefix + + +@respx.mock +def test_api_get_success() -> None: + respx.get(config.API_URL).mock(return_value=Response(200, json={"ok": 1})) + with httpx.Client() as client: + data = scraper._api_get(client, {"action": "query"}, throttle=_throttle(), sleep=_noop) + assert data == {"ok": 1} + + +@respx.mock +def test_api_get_retries_on_503() -> None: + route = respx.get(config.API_URL).mock( + side_effect=[Response(503), Response(200, json={"ok": 2})] + ) + with httpx.Client() as client: + data = scraper._api_get(client, {}, throttle=_throttle(), sleep=_noop) + assert data == {"ok": 2} + assert route.call_count == 2 + + +@respx.mock +def test_api_get_retries_on_maxlag() -> None: + respx.get(config.API_URL).mock( + side_effect=[ + Response(200, json={"error": {"code": "maxlag"}}), + Response(200, json={"ok": 3}), + ] + ) + with httpx.Client() as client: + data = scraper._api_get(client, {}, throttle=_throttle(), sleep=_noop) + assert data == {"ok": 3} + + +@respx.mock +def test_api_get_retries_on_bot_challenge() -> None: + # Cloudflare serves a challenge as 200 + HTML, so status checks pass and only + # the JSON decode fails. Without this the challenge page reaches the corpus. + challenge = Response(200, html="Just a moment...") + respx.get(config.API_URL).mock(side_effect=[challenge, Response(200, json={"ok": 4})]) + with httpx.Client() as client: + assert scraper._api_get(client, {}, throttle=_throttle(), sleep=_noop) == {"ok": 4} + + +@respx.mock +def test_api_get_exhausts_retries_on_persistent_challenge() -> None: + respx.get(config.API_URL).mock( + return_value=Response(200, html="Just a moment...") + ) + with httpx.Client() as client, pytest.raises(RuntimeError): + scraper._api_get(client, {}, throttle=_throttle(), sleep=_noop) + + +@respx.mock +def test_api_get_exhausts_retries() -> None: + respx.get(config.API_URL).mock(return_value=Response(503)) + with httpx.Client() as client, pytest.raises(RuntimeError): + scraper._api_get(client, {}, throttle=_throttle(), sleep=_noop) + + +def test_manifest_roundtrip(tmp_kb) -> None: + assert scraper.load_manifest() == {} + scraper.save_manifest({"A": {"revid": 1}}) + assert scraper.load_manifest() == {"A": {"revid": 1}} + + +@respx.mock +def test_scrape_writes(tmp_kb) -> None: + respx.get(config.API_URL).mock( + side_effect=[ + Response(200, json={"query": {"allpages": [{"title": "Army"}]}}), + Response( + 200, + json={ + "parse": { + "title": "Army", + "revid": 7, + "categories": [], + "text": "

Levies

", + } + }, + ), + ] + ) + with httpx.Client() as client: + result = scraper.scrape(client=client, throttle=_throttle(), log=lambda _: None) + assert result == {"written": 1, "skipped": 0, "missing": 0} + assert (config.CORPUS_DIR / "Army.md").exists() + assert "Army" in scraper.load_manifest() + + +@respx.mock +def test_scrape_skips_existing(tmp_kb) -> None: + (config.CORPUS_DIR / "Army.md").write_text("x", encoding="utf-8") + scraper.save_manifest({"Army": {"revid": 1, "filename": "Army.md"}}) + respx.get(config.API_URL).mock( + return_value=Response(200, json={"query": {"allpages": [{"title": "Army"}]}}) + ) + with httpx.Client() as client: + result = scraper.scrape(client=client, throttle=_throttle(), log=lambda _: None) + assert result == {"written": 0, "skipped": 1, "missing": 0} + + +@respx.mock +def test_scrape_missing(tmp_kb) -> None: + respx.get(config.API_URL).mock( + side_effect=[ + Response(200, json={"query": {"allpages": [{"title": "Ghost"}]}}), + Response(200, json={"error": {"code": "missingtitle"}}), + ] + ) + with httpx.Client() as client: + result = scraper.scrape(client=client, throttle=_throttle(), log=lambda _: None) + assert result == {"written": 0, "skipped": 0, "missing": 1} + + +@respx.mock +def test_scrape_periodic_save_and_owns_client(tmp_kb) -> None: + titles = [{"title": f"P{i}"} for i in range(25)] + responses = [Response(200, json={"query": {"allpages": titles}})] + responses += [ + Response( + 200, + json={ + "parse": { + "title": f"P{i}", + "revid": i, + "categories": [], + "text": f"

{i}

", + } + }, + ) + for i in range(25) + ] + respx.get(config.API_URL).mock(side_effect=responses) + result = scraper.scrape(throttle=_throttle(), log=lambda _: None) # client=None + assert result["written"] == 25 + assert len(scraper.load_manifest()) == 25 + + +def test_main_delegates(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(scraper, "scrape", lambda **_: {"written": 1, "skipped": 0, "missing": 0}) + assert scraper.main(["--limit", "5"]) == 0 + + +def test_build_source_picks_mediawiki_by_default() -> None: + with httpx.Client() as client: + source = scraper.build_source(client, throttle=_throttle()) + assert source.name == "mediawiki" + + +def test_build_source_picks_html_when_configured(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(config, "SOURCE_KIND", "html") + with httpx.Client() as client: + source = scraper.build_source(client, throttle=_throttle()) + assert source.name == "html" + + +@respx.mock +def test_get_returns_404_without_retrying() -> None: + # For the sitemap source a missing page is a normal outcome, not a transient + # failure, so 404 must come back rather than burn the retry budget. + route = respx.get("https://w.example/Gone").mock(return_value=Response(404)) + with httpx.Client() as client: + assert ( + scraper._get(client, "https://w.example/Gone", throttle=_throttle()).status_code == 404 + ) + assert route.call_count == 1 + + +@respx.mock +def test_get_retries_then_succeeds() -> None: + respx.get("https://w.example/P").mock(side_effect=[Response(503), Response(200, text="ok")]) + with httpx.Client() as client: + got = scraper._get(client, "https://w.example/P", throttle=_throttle(), sleep=_noop) + assert got.text == "ok" + + +@respx.mock +def test_get_exhausts_retries() -> None: + respx.get("https://w.example/P").mock(return_value=Response(503)) + with httpx.Client() as client, pytest.raises(RuntimeError): + scraper._get(client, "https://w.example/P", throttle=_throttle(), sleep=_noop) + + +@respx.mock +def test_scrape_skips_noise_titles(tmp_kb) -> None: + # Filtering moved out of the source and into scrape(), so it applies identically + # whichever source enumerated the titles. + respx.get(config.API_URL).mock( + side_effect=[ + Response( + 200, + json={"query": {"allpages": [{"title": "CK3AGOT/Dragon"}, {"title": "Army"}]}}, + ), + Response( + 200, + json={"parse": {"title": "Army", "revid": 1, "categories": [], "text": "

x

"}}, + ), + ] + ) + with httpx.Client() as client: + result = scraper.scrape(client=client, throttle=_throttle(), log=lambda _: None) + assert result == {"written": 1, "skipped": 0, "missing": 0} + assert sorted(scraper.load_manifest()) == ["Army"] + + +def test_is_core_title_subpage_policy_reject(monkeypatch: pytest.MonkeyPatch) -> None: + # CK3-style wiki: subpages are total-conversion mods. + monkeypatch.setattr(config, "SUBPAGE_POLICY", "reject") + monkeypatch.setattr(config, "NOISE_SUBPAGE_TAILS", ()) + assert scraper.is_core_title("Traits") + assert not scraper.is_core_title("CK3AGOT/Dragon") + + +def test_is_core_title_subpage_policy_keep(monkeypatch: pytest.MonkeyPatch) -> None: + # KCD-style wiki: subpages ARE the sequel's pages. Rejecting them would delete + # 441 pages of exactly the content the user wants. + monkeypatch.setattr(config, "SUBPAGE_POLICY", "keep") + monkeypatch.setattr(config, "NOISE_SUBPAGE_TAILS", ("loc",)) + monkeypatch.setattr(config, "MOD_TITLE_PREFIXES", ()) + assert scraper.is_core_title("Alchemy/KCD2") + assert scraper.is_core_title("Archery master/Uzhitz") + assert not scraper.is_core_title("Argonautica/loc") + + +@respx.mock +def test_scrape_with_explicit_titles_skips_enumeration(tmp_kb) -> None: + # Only the parse call is mocked: enumerating the whole wiki would hit an + # unmocked allpages request and fail the test. + respx.get(config.API_URL).mock( + return_value=Response( + 200, + json={"parse": {"title": "Army", "revid": 3, "categories": [], "text": "

x

"}}, + ) + ) + with httpx.Client() as client: + result = scraper.scrape( + titles=["Army"], client=client, throttle=_throttle(), log=lambda _: None + ) + assert result == {"written": 1, "skipped": 0, "missing": 0} + + +def test_main_parses_titles(monkeypatch: pytest.MonkeyPatch) -> None: + seen: dict[str, object] = {} + + def fake_scrape(**kwargs: object) -> dict[str, int]: + seen.update(kwargs) + return {"written": 0, "skipped": 0, "missing": 0} + + monkeypatch.setattr(scraper, "scrape", fake_scrape) + assert scraper.main(["--titles", "Alchemy/KCD2, Lockpicking ,"]) == 0 + assert seen["titles"] == ["Alchemy/KCD2", "Lockpicking"] diff --git a/tests/test_search.py b/tests/test_search.py new file mode 100644 index 0000000..99acad0 --- /dev/null +++ b/tests/test_search.py @@ -0,0 +1,181 @@ +"""Tests for hybrid search.""" + +from __future__ import annotations + +import numpy as np +import pytest + +from wikikb import config, embed, indexer, search + +_TRAITS = """--- +title: "Traits" +url: https://x/Traits +revid: 1 +categories: [] +--- + +# Traits + +Genius gives plus five diplomacy. Domain limit rises with stewardship. +""" + +_ARMY = """--- +title: "Army" +url: https://x/Army +revid: 2 +categories: [] +--- + +# Army + +Levies and knights fight battles. +""" + + +@pytest.fixture +def indexed(tmp_kb, fake_model): + (config.CORPUS_DIR / "traits.md").write_text(_TRAITS, encoding="utf-8") + (config.CORPUS_DIR / "army.md").write_text(_ARMY, encoding="utf-8") + indexer.build_index( + embed_fn=lambda texts: embed.embed_passages(texts, model=fake_model), + log=lambda _: None, + ) + return fake_model + + +def test_fts_match_query_ors_terms() -> None: + assert search._fts_match_query("Domain limit") == '"domain" OR "limit"' + assert search._fts_match_query("!!!") == "" + + +def test_reciprocal_rank_fusion_orders_and_dedupes() -> None: + fused = search.reciprocal_rank_fusion([[1, 2, 3], [2, 1]]) + assert set(fused) == {1, 2, 3} + assert fused[-1] == 3 # only appears once, at rank 2 in one list + + +def test_vector_ids_empty() -> None: + empty = np.zeros((0, config.EMBED_DIM), dtype=np.float32) + query = np.zeros(config.EMBED_DIM, dtype=np.float32) + assert search._vector_ids(empty, query, 5, None) == [] + + +def test_search_ranks_relevant_page_first(indexed, fake_model) -> None: + results = search.search("domain limit stewardship", model=fake_model) + assert results + assert results[0].title == "Traits" + assert "stewardship" in results[0].text.lower() + + +def test_search_no_fts_match_still_returns_vector_hits(indexed, fake_model) -> None: + results = search.search("zzznonexistentxyz", model=fake_model) + assert isinstance(results, list) + assert results # vector retrieval still yields candidates + + +def test_search_punctuation_query_skips_fts(indexed, fake_model) -> None: + # punctuation-only query -> empty FTS match -> _fts_ids returns []; vector still runs + results = search.search("!!!", model=fake_model) + assert isinstance(results, list) + + +def test_search_empty_index_returns_nothing(tmp_kb, fake_model) -> None: + indexer.build_index( + embed_fn=lambda texts: embed.embed_passages(texts, model=fake_model), + log=lambda _: None, + ) + assert search.search("anything", model=fake_model) == [] + + +def test_format_empty() -> None: + assert search._format([]) == "(no results)" + + +def test_format_truncates_long_text() -> None: + result = search.Result( + title="T", + url="u", + source="mediawiki", + variant="kcd2", + section="S", + text="x" * 600, + score=1.0, + ) + out = search._format([result]) + assert "…" in out + assert "[kcd2] T > S" in out + + +def test_main_requires_query() -> None: + assert search.main([]) == 2 + + +def test_main_runs(monkeypatch, capsys) -> None: + monkeypatch.setattr( + search, + "search", + lambda *_a, **_k: [search.Result("T", "u", "mediawiki", "kcd2", "S", "body", 1.0)], + ) + assert search.main(["domain", "limit"]) == 0 + assert "[kcd2] T > S" in capsys.readouterr().out + + +def test_vector_ids_filters_before_truncating() -> None: + # Three vectors ranked 3,2,1 by similarity; only rowid 1 (the WORST match) is in + # scope. Truncating to one candidate first returns nothing; filtering during + # selection returns the in-scope hit. + vectors = np.array([[1.0], [2.0], [3.0]], dtype=np.float32) + query = np.array([1.0], dtype=np.float32) + assert search._vector_ids(vectors, query, 1, {1}) == [1] + assert search._vector_ids(vectors, query, 1, None) == [3] + + +def _variant_doc(title: str, name: str, body: str) -> str: + return ( + f'---\ntitle: "{title}"\nurl: https://x/{title}\nrevid: 1\n' + f"source: mediawiki\nvariant: {name}\ncategories: []\n---\n\n# {title}\n\n{body}\n" + ) + + +@pytest.fixture +def indexed_variants(tmp_kb, fake_model, monkeypatch): + monkeypatch.setattr(config, "VARIANTS", (("kcd2", "KCD2"), ("kcd1", "KCD"))) + (config.CORPUS_DIR / "new.md").write_text( + _variant_doc("Alchemy KCD2", "kcd2", "Brewing potions with herbs."), encoding="utf-8" + ) + (config.CORPUS_DIR / "old.md").write_text( + _variant_doc("Alchemy", "kcd1", "Brewing potions with herbs in the first game."), + encoding="utf-8", + ) + (config.CORPUS_DIR / "shared.md").write_text( + _variant_doc("Moravia", "unknown", "Brewing herbs region shared by both."), + encoding="utf-8", + ) + indexer.build_index( + embed_fn=lambda texts: embed.embed_passages(texts, model=fake_model), log=lambda _: None + ) + return fake_model + + +def test_search_variant_scope_excludes_the_other_edition(indexed_variants, fake_model) -> None: + titles = { + r.title + for r in search.search("brewing potions herbs", k=10, variant="kcd2", model=fake_model) + } + assert "Alchemy KCD2" in titles + assert "Moravia" in titles # unknown is always in scope + assert "Alchemy" not in titles + + +def test_search_any_scope_returns_every_edition(indexed_variants, fake_model) -> None: + titles = { + r.title + for r in search.search("brewing potions herbs", k=10, variant="any", model=fake_model) + } + assert {"Alchemy KCD2", "Alchemy", "Moravia"} <= titles + + +def test_search_results_carry_source_and_variant(indexed_variants, fake_model) -> None: + results = search.search("brewing potions herbs", k=10, variant="any", model=fake_model) + assert {r.source for r in results} == {"mediawiki"} + assert {r.title: r.variant for r in results}["Alchemy"] == "kcd1" diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..ef595b2 --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,63 @@ +"""Tests for the MCP server tool functions.""" + +from __future__ import annotations + +from wikikb import config, scraper, search, server + + +def test_wiki_search_maps_results(monkeypatch) -> None: + monkeypatch.setattr( + server, + "search", + lambda _q, k, variant: [search.Result("T", "u", "mediawiki", "kcd2", "S", "body", 1.0)], + ) + assert server.wiki_search("q", k=3) == [ + { + "title": "T", + "source": "mediawiki", + "variant": "kcd2", + "section": "S", + "url": "u", + "text": "body", + } + ] + + +def test_wiki_get_page_from_manifest(tmp_kb) -> None: + (config.CORPUS_DIR / "Army.md").write_text("PAGE BODY", encoding="utf-8") + scraper.save_manifest({"Army": {"filename": "Army.md"}}) + assert "PAGE BODY" in server.wiki_get_page("Army") + + +def test_wiki_get_page_sanitize_fallback(tmp_kb) -> None: + (config.CORPUS_DIR / "Foo_Bar.md").write_text("FB", encoding="utf-8") + assert "FB" in server.wiki_get_page("Foo Bar") # no manifest entry -> sanitized name + + +def test_wiki_get_page_missing(tmp_kb) -> None: + assert "No mirrored page" in server.wiki_get_page("Nope") + + +def test_wiki_list_pages(tmp_kb) -> None: + scraper.save_manifest({"Army": {}, "Archery": {}, "Barony": {}}) + assert server.wiki_list_pages() == ["Archery", "Army", "Barony"] + assert server.wiki_list_pages(prefix="ar") == ["Archery", "Army"] + assert server.wiki_list_pages(limit=1) == ["Archery"] + + +class _FakeServer: + def __init__(self) -> None: + self.registered: list[str] = [] + + def tool(self): + def decorator(fn): + self.registered.append(fn.__name__) + return fn + + return decorator + + +def test_build_server_registers_tools() -> None: + fake = _FakeServer() + assert server.build_server(fake) is fake + assert fake.registered == ["wiki_search", "wiki_get_page", "wiki_list_pages"] diff --git a/tests/test_sources.py b/tests/test_sources.py new file mode 100644 index 0000000..722629b --- /dev/null +++ b/tests/test_sources.py @@ -0,0 +1,190 @@ +"""Tests for the pluggable page sources and host detection.""" + +from __future__ import annotations + +import httpx +import pytest + +from wikikb import config +from wikikb.sources import ( + HTML, + MEDIAWIKI, + MediaWikiSource, + SitemapHtmlSource, + detect_source_kind, +) + +_SITEMAP = ( + "https://w.example/Common+Longsword" + "https://w.example/Stealth" +) +_INDEX = "https://w.example/s1.xml" +_PAGE = '

Stab 122

' + + +# --------------------------------------------------------------------- MediaWiki + + +def _mw(pages: list[dict]) -> MediaWikiSource: + return MediaWikiSource(lambda _params: pages.pop(0)) + + +def test_mediawiki_iter_titles_paginates() -> None: + source = _mw( + [ + { + "query": {"allpages": [{"title": "A"}, {"title": "B"}]}, + "continue": {"apcontinue": "C"}, + }, + {"query": {"allpages": [{"title": "C"}]}}, + ] + ) + assert list(source.iter_titles()) == ["A", "B", "C"] + + +def test_mediawiki_iter_titles_respects_limit() -> None: + source = _mw( + [{"query": {"allpages": [{"title": "A"}, {"title": "B"}]}, "continue": {"apcontinue": "C"}}] + ) + assert list(source.iter_titles(limit=1)) == ["A"] + + +def test_mediawiki_fetch_page_carries_metadata() -> None: + source = _mw( + [ + { + "parse": { + "title": "Army", + "revid": 7, + "categories": [{"category": "War_stuff"}], + "text": "

hi

", + } + } + ] + ) + page = source.fetch_page("Army") + assert page is not None + assert (page.title, page.revid, page.source) == ("Army", 7, MEDIAWIKI) + assert page.categories == ["War stuff"] + + +def test_mediawiki_fetch_page_missing() -> None: + assert _mw([{"error": {"code": "missingtitle"}}]).fetch_page("Nope") is None + + +# --------------------------------------------------------------------- Sitemap/HTML + + +def _html_source(responses: dict[str, httpx.Response]) -> SitemapHtmlSource: + return SitemapHtmlSource( + lambda url: responses[url], sitemap_url="https://w.example/sitemap.xml" + ) + + +def test_title_from_url_decodes_slug() -> None: + assert SitemapHtmlSource.title_from_url("https://w.example/Common+Longsword") == ( + "Common Longsword" + ) + assert SitemapHtmlSource.title_from_url("https://w.example/A_B/") == "A B" + assert SitemapHtmlSource.title_from_url("https://w.example/") == "" + + +def test_sitemap_iter_titles() -> None: + source = _html_source({"https://w.example/sitemap.xml": httpx.Response(200, text=_SITEMAP)}) + assert list(source.iter_titles()) == ["Common Longsword", "Stealth"] + + +def test_sitemap_iter_titles_skips_empty_slug_and_respects_limit() -> None: + xml = "https://w.example/" + _SITEMAP + source = _html_source({"https://w.example/sitemap.xml": httpx.Response(200, text=xml)}) + assert list(source.iter_titles(limit=1)) == ["Common Longsword"] + + +def test_sitemap_follows_a_sitemap_index() -> None: + source = _html_source( + { + "https://w.example/sitemap.xml": httpx.Response(200, text=_INDEX), + "https://w.example/s1.xml": httpx.Response(200, text=_SITEMAP), + } + ) + assert list(source.iter_titles()) == ["Common Longsword", "Stealth"] + + +def test_sitemap_fetch_page_extracts_the_content_container() -> None: + source = _html_source( + { + "https://w.example/sitemap.xml": httpx.Response(200, text=_SITEMAP), + "https://w.example/Stealth": httpx.Response(200, text=_PAGE), + } + ) + list(source.iter_titles()) # populates the title -> url map + page = source.fetch_page("Stealth") + assert page is not None + assert page.source == HTML + assert page.revid == 0 # no revision concept without an API + assert "Stab 122" in page.html + assert "menu" not in page.html # nav is outside the content container + + +def test_sitemap_fetch_page_404_and_unextractable(monkeypatch) -> None: + monkeypatch.setattr(config, "WIKI_HOST", "https://w.example") + source = _html_source( + { + "https://w.example/Gone": httpx.Response(404), + "https://w.example/Empty": httpx.Response(200, text="no container"), + } + ) + assert source.fetch_page("Gone") is None # honest 404 from the server + assert source.fetch_page("Empty") is None # no recognisable content container + + +def test_sitemap_fetch_page_falls_back_to_a_derived_url(monkeypatch) -> None: + monkeypatch.setattr(config, "WIKI_HOST", "https://w.example") + source = _html_source({"https://w.example/Never%20Listed": httpx.Response(200, text=_PAGE)}) + assert source.fetch_page("Never Listed") is not None + + +# --------------------------------------------------------------------- detection + + +def test_detect_prefers_the_mediawiki_api() -> None: + def get(url: str) -> httpx.Response: + if url.startswith("https://w.example/api.php"): + return httpx.Response(200, json={"query": {"general": {"generator": "MediaWiki 1.43"}}}) + return httpx.Response(404) + + assert detect_source_kind(get, "https://w.example") == (MEDIAWIKI, "https://w.example/api.php") + + +def test_detect_finds_the_wikipedia_style_api_path() -> None: + def get(url: str) -> httpx.Response: + if url.startswith("https://w.example/w/api.php"): + return httpx.Response(200, json={"query": {"general": {"generator": "MediaWiki 1.41"}}}) + return httpx.Response(404) + + assert detect_source_kind(get, "https://w.example")[1] == "https://w.example/w/api.php" + + +def test_detect_falls_back_to_sitemap_when_there_is_no_api() -> None: + def get(url: str) -> httpx.Response: + if url.endswith("sitemap.xml"): + return httpx.Response(200, text=_SITEMAP) + return httpx.Response(404) + + assert detect_source_kind(get, "https://w.example") == (HTML, "https://w.example/sitemap.xml") + + +def test_detect_tolerates_a_non_json_or_non_mediawiki_api_response() -> None: + # A host that serves HTML at /api.php (or some other JSON API) must not be + # mistaken for MediaWiki; detection falls through to the sitemap. + def get(url: str) -> httpx.Response: + if "api.php" in url: + return httpx.Response(200, text="not an api") + return httpx.Response(200, text=_SITEMAP) + + assert detect_source_kind(get, "https://w.example")[0] == HTML + + +def test_detect_raises_when_neither_is_available() -> None: + with pytest.raises(RuntimeError, match="neither a MediaWiki API nor a sitemap"): + detect_source_kind(lambda _url: httpx.Response(404), "https://w.example") diff --git a/tests/test_variant.py b/tests/test_variant.py new file mode 100644 index 0000000..224711a --- /dev/null +++ b/tests/test_variant.py @@ -0,0 +1,50 @@ +"""Tests for edition/variant tagging on wikis that cover two editions at once.""" + +from __future__ import annotations + +import pytest + +from wikikb import config, variant + +# Ordered most-specific-first, exactly as a profile declares it. +_KCD = (("kcd2", "KCD2"), ("kcd1", "KCD")) + + +@pytest.fixture(autouse=True) +def _variants(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(config, "VARIANTS", _KCD) + + +@pytest.mark.parametrize( + ("title", "categories", "expected"), + [ + ("Alchemy/KCD2", [], "kcd2"), # title suffix + ("Common longsword", ["KCD2: Weapons"], "kcd2"), # bare title, category only + ("Lockpicking", ["KCD: Skills"], "kcd1"), + # "KCD main quests" has no colon: a "KCD:"-prefix test would miss it, and a + # naive "KCD" test would also match every KCD2 category. + ("Awakening", ["KCD main quests"], "kcd1"), + ("Weapons", ["KCD: Items", "KCD2: Weapons"], "both"), + ("Moravia", ["Countries"], "unknown"), + ], +) +def test_classify(title: str, categories: list[str], expected: str) -> None: + assert variant.classify(title, categories) == expected + + +def test_classify_without_configured_variants(monkeypatch: pytest.MonkeyPatch) -> None: + # Most wikis cover one subject; tagging is off unless a profile declares it. + monkeypatch.setattr(config, "VARIANTS", ()) + assert variant.classify("Alchemy/KCD2", ["KCD2: Skills"]) == "unknown" + + +def test_matches_excludes_only_the_other_edition() -> None: + assert variant.matches("kcd2", "kcd2") + assert variant.matches("both", "kcd2") + assert variant.matches("unknown", "kcd2") # untagged pages stay in scope + assert not variant.matches("kcd1", "kcd2") + + +def test_matches_any_scope() -> None: + for name in ("kcd1", "kcd2", "both", "unknown"): + assert variant.matches(name, "any") diff --git a/wikikb/__init__.py b/wikikb/__init__.py new file mode 100644 index 0000000..396e72c --- /dev/null +++ b/wikikb/__init__.py @@ -0,0 +1 @@ +"""wiki-kb: local searchable mirror of the Crusader Kings III wiki.""" diff --git a/wikikb/chunker.py b/wikikb/chunker.py new file mode 100644 index 0000000..cdf12b2 --- /dev/null +++ b/wikikb/chunker.py @@ -0,0 +1,95 @@ +"""Split corpus markdown documents into overlapping, heading-scoped chunks.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +from . import config + +_FRONTMATTER = re.compile(r"^---\n(.*?)\n---\n", re.DOTALL) +_HEADING = re.compile(r"^#{1,6}\s+(.*\S)\s*$") + + +@dataclass(frozen=True) +class Chunk: + """A retrievable unit of text with its source page + section.""" + + title: str + url: str + source: str + variant: str + section: str + text: str + + +def parse_frontmatter(text: str) -> tuple[dict[str, str], str]: + """Return (metadata, body). Metadata values are unquoted single-line scalars.""" + match = _FRONTMATTER.match(text) + if not match: + return {}, text + meta: dict[str, str] = {} + for line in match.group(1).splitlines(): + if ":" in line: + key, value = line.split(":", 1) + meta[key.strip()] = value.strip().strip('"') + return meta, text[match.end() :] + + +def _split_sections(body: str) -> list[tuple[str, str]]: + """Group body lines under their most recent heading.""" + sections: list[tuple[str, str]] = [] + heading = "Overview" + buffer: list[str] = [] + for line in body.splitlines(): + match = _HEADING.match(line) + if match: + if buffer: + sections.append((heading, "\n".join(buffer).strip())) + buffer = [] + heading = match.group(1) + else: + buffer.append(line) + if buffer: + sections.append((heading, "\n".join(buffer).strip())) + return [(h, t) for h, t in sections if t] + + +def _window(text: str, size: int, overlap: int) -> list[str]: + """Slice text into <=size windows overlapping by `overlap` chars.""" + if len(text) <= size: + return [text] + step = max(size - overlap, 1) + return [text[start : start + size] for start in range(0, len(text), step)] + + +def chunk_document( + text: str, + *, + target_chars: int = config.CHUNK_TARGET_CHARS, + overlap_chars: int = config.CHUNK_OVERLAP_CHARS, +) -> list[Chunk]: + """Parse a corpus document into heading-scoped, size-bounded chunks.""" + meta, body = parse_frontmatter(text) + title = meta.get("title", "") + url = meta.get("url", "") + source = meta.get("source", "mediawiki") + page_variant = meta.get("variant", "unknown") + chunks: list[Chunk] = [] + for heading, section_text in _split_sections(body): + if heading.lower() in config.NOISE_SECTIONS: + continue + for piece in _window(section_text, target_chars, overlap_chars): + stripped = piece.strip() + if stripped: + chunks.append( + Chunk( + title=title, + url=url, + source=source, + variant=page_variant, + section=heading, + text=stripped, + ) + ) + return chunks diff --git a/wikikb/config.py b/wikikb/config.py new file mode 100644 index 0000000..acfecf1 --- /dev/null +++ b/wikikb/config.py @@ -0,0 +1,175 @@ +"""Central configuration: endpoints, paths, crawl politeness, model choice.""" + +from __future__ import annotations + +import os +from pathlib import Path + +# --- Wiki endpoints --------------------------------------------------------- +# Override WIKI_KB_HOST to re-point this KB at a different MediaWiki (scaffolding +# rewrites the literal default below for a new wiki). +WIKI_HOST = os.environ.get("WIKI_KB_HOST", "https://ck3.paradoxwikis.com").rstrip("/") +# The API path varies by install: paradoxwikis / wiki.gg use /api.php, Wikipedia +# uses /w/api.php. The article URL pattern varies too: /{Title} vs /wiki/{Title}. +# Override both when re-pointing at a wiki that doesn't match the paradoxwikis layout. +API_URL = os.environ.get("WIKI_KB_API_URL", f"{WIKI_HOST}/api.php") +PAGE_URL_TEMPLATE = os.environ.get("WIKI_KB_PAGE_URL", f"{WIKI_HOST}/{{title}}") + +# Namespace 0 is the main (content) namespace. +CONTENT_NAMESPACE = 0 + +# Which page source to use: "mediawiki" (an open api.php — always preferred) or +# "html" (sitemap.xml + content-container extraction, for wikis with no API at all). +# scripts/detect_wiki.py probes a host and reports which applies. +SOURCE_KIND = os.environ.get("WIKI_KB_SOURCE", "mediawiki") +SITEMAP_URL = os.environ.get("WIKI_KB_SITEMAP", f"{WIKI_HOST}/sitemap.xml") + +# --- Subpage policy ----------------------------------------------------------- +# Whether "A/B" titles are noise or content. This is PER WIKI and must be measured, +# never assumed: on the CK3 wiki subpages are total-conversion mods (reject), while on +# the KCD wiki they are the sequel's pages -- "Alchemy/KCD2" -- so rejecting them +# deletes 441 pages of exactly the content you want (keep). +# Counter(t.rsplit("/", 1)[-1] for t in titles if "/" in t) +# answers it in one API pass. +SUBPAGE_POLICY = os.environ.get("WIKI_KB_SUBPAGES", "reject") +# Subpage tails dropped even when the policy is "keep" (e.g. "loc" localisation dumps). +NOISE_SUBPAGE_TAILS = tuple( + t.strip().lower() for t in os.environ.get("WIKI_KB_NOISE_TAILS", "").split(",") if t.strip() +) + +# --- Editions / variants ------------------------------------------------------ +# Some wikis cover two editions of one subject in a single namespace (a game and its +# sequel, two versions of a product). Declare them as "name:PREFIX" pairs, MOST +# SPECIFIC FIRST -- ordering is what lets "KCD" mean edition 1 without also matching +# "KCD2". A page matches a variant if its title ends "/PREFIX" or a category starts +# with PREFIX. Empty (the default) disables variant tagging entirely. +VARIANTS = tuple( + (part.split(":", 1)[0].strip(), part.split(":", 1)[1].strip()) + for part in os.environ.get("WIKI_KB_VARIANTS", "").split(",") + if ":" in part +) +# Default search scope. "any" returns everything; a variant name excludes the OTHER +# variants while always keeping pages tagged "both" or "unknown". +DEFAULT_VARIANT = os.environ.get("WIKI_KB_DEFAULT_VARIANT", "any") + +# Total-conversion mod pages live in ns0 as subpages ("CK3AGOT/Dragon") or under +# these top-level prefixes. They are not base-game mechanics and pollute retrieval, +# so the scraper skips them (any title containing "/" is also treated as a mod/subpage). +_DEFAULT_MOD_PREFIXES = ( + "ck3agot", + "lotr", + "elder kings", + "after the end", + "the fallen eagle", + "fallen eagle", + "godherja", + "princes of darkness", + "rice", + "realms in exile", + "a game of thrones", + "when the world stopped making sense", + "more bookmarks", +) +# Override per-wiki with WIKI_KB_MOD_PREFIXES (comma-separated). Branchless so the +# import line is always exercised regardless of whether the env var is set. +MOD_TITLE_PREFIXES = tuple( + p.strip().lower() + for p in os.environ.get("WIKI_KB_MOD_PREFIXES", ",".join(_DEFAULT_MOD_PREFIXES)).split(",") + if p.strip() +) + +# --- Politeness / crawl ----------------------------------------------------- +# Identify the bot honestly with a contact, per wiki etiquette. +# MediaWiki API:Etiquette requires a descriptive User-Agent with a real contact. +# Set WIKI_KB_CONTACT — the placeholder is deliberately obvious so an unconfigured +# run is embarrassing rather than silently anonymous. +CONTACT = os.environ.get("WIKI_KB_CONTACT", "UNCONFIGURED-set-WIKI_KB_CONTACT") +USER_AGENT = f"wiki-kb/0.1 (personal offline RAG mirror; contact: {CONTACT})" + +# Tool prefix for the MCP server: _search / _get_page / _list_pages. +MCP_NAMESPACE = os.environ.get("WIKI_KB_NAMESPACE", "wiki") +# Minimum seconds between requests (single-flight, ~1 req/s). +REQUEST_INTERVAL_S = 1.0 +# maxlag tells MediaWiki to reject our request if replication lag is high, +# so we back off instead of adding load. See MediaWiki API:Etiquette. +MAXLAG = 5 +REQUEST_TIMEOUT_S = 30.0 +MAX_RETRIES = 5 +BACKOFF_BASE_S = 2.0 # exponential: BACKOFF_BASE ** attempt + +# --- Paths ------------------------------------------------------------------ +# WIKI_KB_ROOT isolates a KB's generated data (corpus/index/manifest) — lets one +# codebase serve multiple wikis by pointing at different roots. Branchless default. +_DEFAULT_ROOT = Path(__file__).resolve().parent.parent +ROOT = Path(os.environ.get("WIKI_KB_ROOT", str(_DEFAULT_ROOT))) +CORPUS_DIR = ROOT / "corpus" +INDEX_DIR = ROOT / "index" +DB_PATH = INDEX_DIR / "wiki.db" +EMBEDDINGS_PATH = INDEX_DIR / "embeddings.npy" +MANIFEST_PATH = ROOT / "manifest.json" + +# --- Chunking --------------------------------------------------------------- +CHUNK_TARGET_CHARS = 2000 # ~500 tokens +CHUNK_OVERLAP_CHARS = 200 +# Section headings excluded from the index (the page is still mirrored in full and +# still served by *_get_page). Wikis carry per-page boilerplate sections that are +# both expensive to embed and harmful to retrieval — on the KCD wiki, "Other +# languages" tables of localized UI names were 7% of all chunks and repeated every +# page's title in eight languages. Empty by default; set per wiki after inspecting +# the section histogram. Compared case-insensitively. +NOISE_SECTIONS = frozenset( + s.strip().lower() for s in os.environ.get("WIKI_KB_NOISE_SECTIONS", "").split(",") if s.strip() +) + +# --- Embeddings ------------------------------------------------------------- +EMBED_MODEL = "BAAI/bge-small-en-v1.5" # 384-dim, CPU-friendly via fastembed +EMBED_DIM = 384 +EMBED_BATCH_SIZE = 32 + + +def _parse_core_spec(spec: str, total: int) -> set[int]: + """Parse "0-3,8" into a set of core ids clamped to [0, total).""" + cores: set[int] = set() + for part in spec.split(","): + part = part.strip() + if not part: + continue + if "-" in part: + lo, hi = part.split("-", 1) + cores.update(range(int(lo), int(hi) + 1)) + else: + cores.add(int(part)) + return {c for c in cores if 0 <= c < total} + + +def _allowed_cores(total: int, reserved: set[int]) -> list[int]: + """Cores usable by heavy jobs = all minus reserved (never empty).""" + allowed = sorted(set(range(total)) - reserved) + return allowed or list(range(total)) + + +# --- CPU isolation ---------------------------------------------------------- +# The bottleneck is core *contention* with the foreground game, not thread count. +# Keep RESERVED_CORES free for the game and let heavy jobs use every other core +# fully. Override WIKI_KB_RESERVED_CORES (e.g. "0-3" or "8,9"); set "" to use all cores. +TOTAL_CPUS = os.cpu_count() or 4 +RESERVED_CORES = _parse_core_spec(os.environ.get("WIKI_KB_RESERVED_CORES", "0-3"), TOTAL_CPUS) +ALLOWED_CORES = _allowed_cores(TOTAL_CPUS, RESERVED_CORES) +# GENTLE BY DEFAULT: a running game floats across ALL cores, so heavy background +# work drops FPS no matter which cores it "owns" — nice only softens it. Default to +# a single-threaded, single-worker build (~1 core) pinned off RESERVED_CORES and +# niced. When the game is CLOSED, set WIKI_KB_EMBED_PARALLEL= for a fast build. +EMBED_PARALLEL = int(os.environ.get("WIKI_KB_EMBED_PARALLEL", "1")) +EMBED_THREADS = int(os.environ.get("WIKI_KB_EMBED_THREADS", "1")) +# Lower scheduling priority so the game still wins on any shared core. +NICE_INCREMENT = int(os.environ.get("WIKI_KB_NICE", "10")) + +# --- Retrieval -------------------------------------------------------------- +RRF_K = 60 # Reciprocal Rank Fusion constant +DEFAULT_TOP_K = 8 +CANDIDATE_POOL = 40 # per-retriever candidates before fusion + + +def page_url(title: str) -> str: + """Public wiki URL for a page title.""" + return PAGE_URL_TEMPLATE.format(title=title.replace(" ", "_")) diff --git a/wikikb/cpu.py b/wikikb/cpu.py new file mode 100644 index 0000000..d259351 --- /dev/null +++ b/wikikb/cpu.py @@ -0,0 +1,33 @@ +"""Keep heavy jobs (embedding) off reserved cores and at low priority. + +The foreground game keeps its cores; this process is pinned to the rest and niced +down so it yields on any shared core. Linux-only affinity is applied when present. +""" + +from __future__ import annotations + +import os +from collections.abc import Sequence + +from . import config + + +def isolate( + *, + allowed: Sequence[int] | None = None, + nice_increment: int | None = None, +) -> list[int]: + """Pin this process to `allowed` cores and lower its priority. + + Returns the cores it was pinned to. No-ops the affinity call on platforms + without ``os.sched_setaffinity`` (non-Linux). + """ + cores = list(config.ALLOWED_CORES if allowed is None else allowed) + increment = config.NICE_INCREMENT if nice_increment is None else nice_increment + + set_affinity = getattr(os, "sched_setaffinity", None) + if set_affinity is not None: + set_affinity(0, set(cores)) + if increment and hasattr(os, "nice"): + os.nice(increment) + return cores diff --git a/wikikb/embed.py b/wikikb/embed.py new file mode 100644 index 0000000..f86af31 --- /dev/null +++ b/wikikb/embed.py @@ -0,0 +1,60 @@ +"""Local embeddings via fastembed (bge-small), L2-normalized for cosine = dot.""" + +from __future__ import annotations + +from collections.abc import Iterable +from typing import Protocol + +import numpy as np + +from . import config + + +class EmbeddingModel(Protocol): + """Minimal fastembed-compatible surface (also satisfied by test fakes).""" + + def embed(self, texts: Iterable[str], **kwargs: object) -> Iterable[np.ndarray]: ... + def query_embed(self, texts: Iterable[str], **kwargs: object) -> Iterable[np.ndarray]: ... + + +_model: EmbeddingModel | None = None + + +def get_model() -> EmbeddingModel: + """Lazily construct the (heavy) fastembed model once.""" + global _model + if _model is None: # pragma: no cover - exercised only with the real model + from fastembed import TextEmbedding + + # threads caps onnxruntime's intra-op parallelism (the main CPU driver); + # CPUExecutionProvider avoids grabbing any accelerator. + _model = TextEmbedding( + model_name=config.EMBED_MODEL, + threads=config.EMBED_THREADS, + providers=["CPUExecutionProvider"], + ) + return _model + + +def _normalize(matrix: np.ndarray) -> np.ndarray: + if matrix.size == 0: + return matrix.reshape(0, config.EMBED_DIM).astype(np.float32) + norms = np.linalg.norm(matrix, axis=1, keepdims=True) + norms[norms == 0] = 1.0 + return (matrix / norms).astype(np.float32) + + +def embed_passages(texts: list[str], *, model: EmbeddingModel | None = None) -> np.ndarray: + """Embed corpus passages -> (N, dim) normalized float32.""" + model = model or get_model() + vectors = list( + model.embed(texts, batch_size=config.EMBED_BATCH_SIZE, parallel=config.EMBED_PARALLEL) + ) + return _normalize(np.array(vectors, dtype=np.float32)) + + +def embed_query(text: str, *, model: EmbeddingModel | None = None) -> np.ndarray: + """Embed a search query -> (dim,) normalized float32 (with bge query prefix).""" + model = model or get_model() + vector = np.array(next(iter(model.query_embed([text]))), dtype=np.float32) + return _normalize(vector.reshape(1, -1))[0] diff --git a/wikikb/html2md.py b/wikikb/html2md.py new file mode 100644 index 0000000..2e43dc3 --- /dev/null +++ b/wikikb/html2md.py @@ -0,0 +1,167 @@ +"""Convert MediaWiki `action=parse` HTML into clean markdown with frontmatter.""" + +from __future__ import annotations + +import re + +from bs4 import BeautifulSoup +from markdownify import markdownify as _markdownify + +# Chrome / navigation / reference elements we don't want in the corpus. +_REMOVE_SELECTORS = ( + "style", + "script", + "link", + ".mw-editsection", + ".mw-jump-link", + ".noprint", + ".navbox", + ".toc", + "#toc", + ".mw-empty-elt", + ".reference", + ".mw-references-wrap", + ".reflist", + ".ambox", + ".metadata", + ".pagetabber-absolute", + ".pi-image", +) + +_BLANK_LINES = re.compile(r"\n{3,}") + + +def _infobox_rows(element, group: str, tab: str, rows: list[tuple[str, str, str, str]]) -> None: + """Collect (group, tab, label, value) from a portable-infobox subtree, in order. + + Fandom stat blocks are an