feat: add full-text search via Awesomebar

compendium/hooks.py

@@ -46,6 +46,9 @@
 	},
 ]
 
+# Extra Awesome Bar results: list of dicts with label, description, route, index.
+awesomebar_search = ["compendium.search.awesomebar_results"]
+
 # include js, css files in header of web template
 # web_include_css = "/assets/compendium/css/compendium.css"
 # web_include_js = "/assets/compendium/js/compendium.js"

compendium/search.py

@@ -0,0 +1,139 @@
+# Copyright (c) 2026, ALYF GmbH and Contributors
+# License: MIT. See LICENSE
+
+import os
+import re
+import sqlite3
+import threading
+
+import frappe
+from frappe import _
+from frappe.utils import escape_html
+
+from compendium import docs
+from compendium.docs import DEFAULT_LANG, DOCS_FOLDER, is_permitted, normalize_locale
+
+RESULT_LIMIT = 20
+TOKEN_PATTERN = re.compile(r"\w+")
+SNIPPET_TOKENS = 12
+ROLE_SEPARATOR = "\n"
+# Columns are (path, title, roles, body); bm25 ranks a title hit above a body hit.
+SEARCH_SQL = f"""
+	SELECT path, title, roles, snippet(pages, 3, '', '', '…', {SNIPPET_TOKENS})
+	FROM pages WHERE pages MATCH ? ORDER BY bm25(pages, 2.0, 10.0, 0.0, 1.0)
+"""
+
+# ponytail: one lock for all indexes; per-index locks if search ever gets hot
+INDEX_LOCK = threading.Lock()
+INDEXES = {}
+
+
+def awesomebar_results(txt):
+	"""Return matching docs pages for the Awesome Bar `awesomebar_search` hook."""
+	match_query = build_match_query(txt)
+	if not match_query:
+		return []
+
+	locale = normalize_locale(frappe.local.lang or DEFAULT_LANG)
+	results = []
+
+	for path, title, roles, snippet in search(locale, match_query):
+		if not is_permitted(frappe._dict(roles=roles.split(ROLE_SEPARATOR))):
+			continue
+
+		route = f"/app/docs/{locale}/{path}" if path else f"/app/docs/{locale}"
+		results.append(
+			{
+				"label": title,
+				# the Awesome Bar renders the description as HTML
+				"description": escape_html(snippet) or path or _("Documentation"),
+				"route": route,
+				"index": 50,
+			}
+		)
+		if len(results) >= RESULT_LIMIT:
+			break
+
+	return results
+
+
+def build_match_query(txt):
+	"""Turn free text into an FTS5 MATCH expression: every word as a quoted prefix term.
+
+	Quoting keeps words like `and` or `not` from being read as query operators.
+	"""
+	return " ".join(f'"{token}"*' for token in TOKEN_PATTERN.findall((txt or "").lower()))
+
+
+def search(locale, match_query):
+	with INDEX_LOCK:
+		return get_index(locale).execute(SEARCH_SQL, (match_query,)).fetchall()
+
+
+def get_index(locale):
+	"""Full-content FTS5 index for one locale, kept in memory for the life of the worker.
+
+	Rebuilt when a Markdown file appears, disappears or changes, so a docs deployment
+	needs no restart and an author editing a page sees the change on the next search.
+	Sites share a worker but not their installed apps, hence the site in the key.
+	"""
+	key = (frappe.local.site, locale)
+	fingerprint = get_docs_fingerprint()
+	cached = INDEXES.get(key)
+	if cached and cached[0] == fingerprint:
+		return cached[1]
+
+	index = build_index(locale)
+	INDEXES[key] = (fingerprint, index)
+	return index
+
+
+def build_index(locale):
+	# the index outlives the request, so it outlives the thread that built it
+	connection = sqlite3.connect(":memory:", check_same_thread=False)
+	connection.execute(
+		"CREATE VIRTUAL TABLE pages USING fts5"
+		"(path, title, roles UNINDEXED, body, tokenize='unicode61 remove_diacritics 2')"
+	)
+	connection.executemany(
+		"INSERT INTO pages (path, title, roles, body) VALUES (?, ?, ?, ?)",
+		(
+			(page.path, page.title, ROLE_SEPARATOR.join(page.roles), to_plain_text(page.body))
+			for page in docs.discover_pages(locale).values()
+		),
+	)
+	return connection
+
+
+@frappe.request_cache
+def get_docs_fingerprint():
+	"""Staleness check for the index: how many Markdown files there are, and the newest mtime.
+
+	Stat-only, so it costs a fraction of the reading and parsing an index build does.
+	"""
+	count = 0
+	latest = 0
+
+	for app in docs.get_installed_apps():
+		docs_root = os.path.join(docs.get_app_path(app), DOCS_FOLDER)
+		for basepath, _folders, files in os.walk(docs_root):
+			for fname in files:
+				if not fname.endswith(".md"):
+					continue
+
+				count += 1
+				latest = max(latest, os.stat(os.path.join(basepath, fname)).st_mtime_ns)
+
+	return count, latest
+
+
+def to_plain_text(markdown):
+	"""The words of a page, without the markup around them.
+
+	Markdown source makes both a poor preview and a poor index: a snippet cut out of it
+	shows syntax mid-sentence, and link targets match queries the reader never sees.
+	"""
+	from bs4 import BeautifulSoup
+
+	text = BeautifulSoup(frappe.utils.md_to_html(markdown or ""), "html.parser").get_text(" ")
+	return re.sub(r"\s+", " ", text).strip()

compendium/tests/test_search.py

@@ -0,0 +1,151 @@
+import os
+from unittest.mock import patch
+
+import frappe
+from frappe.tests.utils import FrappeTestCase
+
+from compendium.search import awesomebar_results, get_index
+from compendium.tests.test_docs import DocsTestEnvironment
+
+
+class TestSearch(FrappeTestCase):
+	def setUp(self):
+		frappe.set_user("Administrator")
+		frappe.local.lang = "en"
+		self.next_request()
+
+	def next_request(self):
+		"""Drop everything a single request caches, so the next call starts like a fresh one."""
+		if hasattr(frappe.local, "request_cache"):
+			frappe.local.request_cache.clear()
+
+	def tearDown(self):
+		frappe.set_user("Administrator")
+		frappe.local.lang = "en"
+
+	def test_empty_query_returns_nothing(self):
+		self.assertEqual(awesomebar_results(""), [])
+		self.assertEqual(awesomebar_results("   "), [])
+		self.assertEqual(awesomebar_results("!?-"), [])
+
+	def test_matches_title_and_builds_docs_route(self):
+		with DocsTestEnvironment(
+			{
+				"en/guides/setup.md": "---\ntitle: Setup Guide\n---\n# Setup\n\nInstall the app.",
+				"en/other.md": "---\ntitle: Other\n---\n# Other",
+			}
+		):
+			results = awesomebar_results("setup")
+
+		self.assertEqual(len(results), 1)
+		self.assertEqual(results[0]["label"], "Setup Guide")
+		self.assertEqual(results[0]["route"], "/app/docs/en/guides/setup")
+		self.assertEqual(results[0]["index"], 50)
+
+	def test_matches_path(self):
+		with DocsTestEnvironment(
+			{
+				"en/guides/setup.md": "---\ntitle: Setup Guide\n---\n# Setup",
+			}
+		):
+			results = awesomebar_results("guides")
+
+		self.assertEqual(results[0]["route"], "/app/docs/en/guides/setup")
+
+	def test_matches_body_content(self):
+		with DocsTestEnvironment(
+			{
+				"en/setup.md": "---\ntitle: Setup\n---\nRun bench migrate to apply patches.",
+				"en/other.md": "---\ntitle: Other\n---\nNothing to see here.",
+			}
+		):
+			results = awesomebar_results("patches")
+
+		self.assertEqual([item["label"] for item in results], ["Setup"])
+		self.assertIn("patches", results[0]["description"])
+
+	def test_matches_word_prefix(self):
+		with DocsTestEnvironment({"en/setup.md": "---\ntitle: Setup\n---\nRun bench migrate."}):
+			results = awesomebar_results("migr")
+
+		self.assertEqual([item["label"] for item in results], ["Setup"])
+
+	def test_all_words_must_match(self):
+		with DocsTestEnvironment(
+			{
+				"en/setup.md": "---\ntitle: Setup\n---\nRun bench migrate.",
+				"en/other.md": "---\ntitle: Other\n---\nRun the tests.",
+			}
+		):
+			results = awesomebar_results("run migrate")
+
+		self.assertEqual([item["label"] for item in results], ["Setup"])
+
+	def test_title_hit_ranks_above_body_hit(self):
+		with DocsTestEnvironment(
+			{
+				"en/backup.md": "---\ntitle: Backup\n---\nHow to keep copies.",
+				"en/setup.md": "---\ntitle: Setup\n---\nTake a backup first.",
+			}
+		):
+			results = awesomebar_results("backup")
+
+		self.assertEqual([item["label"] for item in results], ["Backup", "Setup"])
+
+	def test_snippet_is_plain_text(self):
+		with DocsTestEnvironment(
+			{
+				"en/setup.md": (
+					"---\ntitle: Setup\n---\n## Bench\n\n"
+					"*Run* <b>bench</b> [migrate](https://example.com/docs) & wait."
+				),
+			}
+		):
+			results = awesomebar_results("migrate")
+
+		self.assertEqual(results[0]["description"], "Bench Run bench migrate &amp; wait.")
+
+	def test_link_targets_are_not_searchable(self):
+		with DocsTestEnvironment(
+			{"en/setup.md": "---\ntitle: Setup\n---\nSee [the manual](https://example.com/hyperion)."}
+		):
+			self.assertEqual(awesomebar_results("hyperion"), [])
+			self.assertEqual([item["label"] for item in awesomebar_results("manual")], ["Setup"])
+
+	def test_index_page_route_has_no_trailing_path(self):
+		with DocsTestEnvironment({"en/index.md": "---\ntitle: Home\n---\n# Home"}):
+			results = awesomebar_results("home")
+
+		self.assertEqual(results[0]["route"], "/app/docs/en")
+
+	def test_skips_unpermitted_pages(self):
+		with DocsTestEnvironment(
+			{
+				"en/public.md": "---\ntitle: Public Guide\nroles: Desk User\n---\n# Public",
+				"en/admin.md": "---\ntitle: Admin Guide\nroles: System Manager\n---\n# Admin",
+			}
+		):
+			frappe.set_user("test@example.com")
+			with patch("compendium.docs.get_user_roles", return_value=["Desk User"]):
+				results = awesomebar_results("guide")
+
+		labels = [item["label"] for item in results]
+		self.assertEqual(labels, ["Public Guide"])
+
+	def test_index_is_reused_across_requests(self):
+		with DocsTestEnvironment({"en/setup.md": "---\ntitle: Setup\n---\nRun bench migrate."}):
+			index = get_index("en")
+			self.next_request()
+			self.assertIs(get_index("en"), index)
+
+	def test_index_rebuilds_when_a_page_changes(self):
+		with DocsTestEnvironment({"en/setup.md": "---\ntitle: Setup\n---\nRun bench migrate."}) as docs_root:
+			self.assertEqual(awesomebar_results("hyphenation"), [])
+
+			with open(os.path.join(docs_root, "en", "extra.md"), "w", encoding="utf-8") as f:
+				f.write("---\ntitle: Extra\n---\nAbout hyphenation.")
+
+			self.next_request()
+			results = awesomebar_results("hyphenation")
+
+		self.assertEqual([item["label"] for item in results], ["Extra"])