ai: help-docs search — Help mode tools

Search the wrolpi-help markdown in-process: files are parsed on demand
and cached by mtime; title matches outrank headings outrank body.  No
database table — the corpus is dozens of small files.  Pages are read
as paged markdown.  The docker api container gets a read-only bind
mount of the help docs at the native /opt/wrolpi-help location; a
missing help directory degrades to empty results with a message.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

docker-compose.yml

@@ -49,6 +49,8 @@ services:
       - './scripts:/opt/wrolpi/scripts'
       - './test:/opt/wrolpi/test'
       - './test:/media/wrolpi'
+      # Help markdown for the AI help search (native installs read /opt/wrolpi-help directly).
+      - './docker/help/wrolpi-help/docs:/opt/wrolpi-help/docs:ro'
       - './alembic:/opt/wrolpi/alembic'
       - './alembic.ini:/opt/wrolpi/alembic.ini'
     ports:

modules/ai/api.py

@@ -26,7 +26,7 @@ from wrolpi.files.lib import search_files
 from wrolpi.vars import DOCKERIZED
 from wrolpi.version import __version__
 from wrolpi.schema import JSONErrorResponse
-from . import lib, schema
+from . import help_docs, lib, schema
 from .controller_client import controller_get
 from .errors import ControllerUnavailable
 
@@ -334,6 +334,49 @@ async def get_inventory(_: Request, slug: str):
     return json_response(dict(inventory=inventory))
 
 
+# ---------------------------------------------------------------------------
+# Help endpoints (Help mode).  Help markdown is searched in-process; see help_docs.py.
+# ---------------------------------------------------------------------------
+
+@ai_bp.post('/help/search')
+@openapi.definition(
+    summary='Search the WROLPi help documentation',
+    description='Search the WROLPi help documentation for how to use and repair WROLPi. Returns matching'
+                ' help pages with a snippet; read a full page with get_help_doc. Links are pages on the'
+                ' WROLPi help site.',
+    body=schema.AIHelpSearchRequest,
+)
+@openapi.operation('search_help')
+@openapi.response(HTTPStatus.OK, schema.AISearchResponse)
+@validate(schema.AIHelpSearchRequest)
+async def search_help(_: Request, body: schema.AIHelpSearchRequest):
+    if not body.search_str:
+        raise SearchEmpty()
+    if not help_docs.get_help_docs_directory():
+        return json_response(dict(results=[], total=0, message='Help documentation is not installed.'))
+    results, total = help_docs.search_help(body.search_str, ai_limiter(body.limit))
+    return json_response(dict(results=results, total=total))
+
+
+@ai_bp.get('/help/<slug:path>')
+@openapi.definition(
+    summary='Read a help page',
+    description='Read one WROLPi help page as markdown by its doc slug from search_help results. Long'
+                ' pages are paged: request again with offset=next_offset to continue reading.',
+)
+@openapi.operation('get_help_doc')
+@openapi.response(HTTPStatus.OK, schema.AIPagedTextResponse)
+@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
+async def get_help_doc(request: Request, slug: str):
+    doc = help_docs.get_help_doc(slug)
+    if not doc:
+        raise UnknownFile(f'No such help doc: {slug}')
+    result = lib.paginate_text(doc.body, _offset(request))
+    result['title'] = doc.title
+    result['link'] = help_docs.help_doc_link(doc.slug)
+    return json_response(result)
+
+
 # ---------------------------------------------------------------------------
 # System endpoints (System mode).  The Controller is proxied so the model sees one API; every
 # proxy degrades gracefully — a Controller outage becomes an entry in `errors`, never a 500.

modules/ai/help_docs.py

@@ -0,0 +1,152 @@
+"""In-memory search over the WROLPi help markdown.
+
+The help docs are a small corpus (dozens of markdown files) served by mkdocs from the wrolpi-help
+repository.  They live outside the media directory and get no FileGroups, so they are searched
+in-process: files are parsed on demand and cached by mtime.  No database, no index to build.
+"""
+import dataclasses
+import os
+import pathlib
+import re
+from typing import Dict, List, Optional
+
+from wrolpi.common import logger
+from wrolpi.vars import PROJECT_DIR
+
+logger = logger.getChild(__name__)
+
+# Native installs clone wrolpi-help to /opt/wrolpi-help; development uses the git submodule.
+# The docker api container gets a read-only bind mount at the native location.
+DEFAULT_HELP_DIRS = (
+    pathlib.Path('/opt/wrolpi-help/docs'),
+    PROJECT_DIR / 'docker/help/wrolpi-help/docs',
+)
+
+SNIPPET_LENGTH = 200
+
+
+def get_help_docs_directory() -> Optional[pathlib.Path]:
+    """The directory containing the help markdown, or None when help is not installed."""
+    if directory := os.environ.get('HELP_DOCS_DIR'):
+        directory = pathlib.Path(directory)
+        return directory if directory.is_dir() else None
+    for directory in DEFAULT_HELP_DIRS:
+        if directory.is_dir():
+            return directory
+    return None
+
+
+@dataclasses.dataclass
+class HelpDoc:
+    slug: str  # relative path without .md, e.g. "system/logs"
+    title: str
+    headings: List[str]
+    body: str
+    mtime: float
+
+
+# slug -> HelpDoc, invalidated per-file by mtime.
+_CACHE: Dict[str, HelpDoc] = {}
+
+
+def _parse_doc(slug: str, path: pathlib.Path) -> HelpDoc:
+    body = path.read_text(errors='replace')
+    title = None
+    headings = []
+    for line in body.splitlines():
+        if match := re.match(r'^(#{1,6})\s+(.+)', line):
+            heading = match.group(2).strip()
+            headings.append(heading)
+            if title is None and match.group(1) == '#':
+                title = heading
+    return HelpDoc(slug=slug, title=title or slug, headings=headings, body=body, mtime=path.stat().st_mtime)
+
+
+def load_help_docs() -> List[HelpDoc]:
+    """Load every help markdown file, reusing cached parses of unchanged files."""
+    directory = get_help_docs_directory()
+    if not directory or not directory.is_dir():
+        return []
+
+    docs = []
+    seen = set()
+    for path in sorted(directory.rglob('*.md')):
+        slug = str(path.relative_to(directory)).removesuffix('.md')
+        seen.add(slug)
+        try:
+            cached = _CACHE.get(slug)
+            if not cached or cached.mtime != path.stat().st_mtime:
+                _CACHE[slug] = _parse_doc(slug, path)
+            docs.append(_CACHE[slug])
+        except OSError as e:
+            logger.warning(f'Could not read help doc {path}', exc_info=e)
+    # Drop deleted files from the cache.
+    for slug in set(_CACHE) - seen:
+        del _CACHE[slug]
+    return docs
+
+
+def help_doc_link(slug: str) -> str:
+    """The mkdocs URL path of a doc; the help server (port 8086) serves index.md at /."""
+    if slug == 'index':
+        return '/'
+    slug = slug.removesuffix('/index')
+    return f'/{slug}/'
+
+
+def _tokenize(text: str) -> List[str]:
+    return [i for i in re.split(r'\W+', text.lower()) if len(i) >= 2]
+
+
+def _snippet(body: str, tokens: List[str]) -> Optional[str]:
+    """A little context around the first token match in the body."""
+    lower = body.lower()
+    position = min((p for token in tokens if (p := lower.find(token)) >= 0), default=-1)
+    if position < 0:
+        return None
+    start = max(position - SNIPPET_LENGTH // 4, 0)
+    snippet = body[start:start + SNIPPET_LENGTH]
+    snippet = re.sub(r'\s+', ' ', snippet).strip()
+    return f'…{snippet}…'
+
+
+def search_help(search_str: str, limit: int = 5) -> (List[dict], int):
+    """Score every help doc against the query; title matches beat heading matches beat body matches."""
+    tokens = _tokenize(search_str or '')
+    if not tokens:
+        return []
+
+    scored = []
+    for doc in load_help_docs():
+        title = doc.title.lower()
+        headings = ' '.join(doc.headings).lower()
+        body = doc.body.lower()
+        score = 0
+        for token in tokens:
+            score += 10 * title.count(token)
+            score += 4 * headings.count(token)
+            score += body.count(token)
+        if score:
+            scored.append((score, doc))
+
+    scored.sort(key=lambda i: (-i[0], i[1].slug))
+    results = []
+    for score, doc in scored[:limit]:
+        result = dict(
+            doc=doc.slug,
+            title=doc.title,
+            link=help_doc_link(doc.slug),
+            snippet=_snippet(doc.body, tokens),
+        )
+        results.append({k: v for k, v in result.items() if v is not None})
+    return results, len(scored)
+
+
+def get_help_doc(slug: str) -> Optional[HelpDoc]:
+    """Fetch one help doc by its slug.  Returns None when it (or the help directory) is missing."""
+    if '..' in slug.split('/'):
+        return None
+    for doc in load_help_docs():
+        if doc.slug == slug:
+            return doc
+    return None

modules/ai/schema.py

@@ -49,6 +49,12 @@ class AIZimSearchRequest:
     offset: int = 0
 
 
+@dataclasses.dataclass
+class AIHelpSearchRequest:
+    search_str: Optional[str] = None
+    limit: int = 5
+
+
 @dataclasses.dataclass
 class AISearchResponse:
     results: list = dataclasses.field(default_factory=list)

modules/ai/test/test_help_docs.py

@@ -0,0 +1,114 @@
+"""Tests for the help-docs search (Help mode).  Uses a fixture corpus, never the real submodule."""
+import json
+from http import HTTPStatus
+
+import pytest
+
+from modules.ai import help_docs, lib
+
+
+@pytest.fixture
+def help_corpus(test_directory, monkeypatch):
+    """A small help-docs corpus in a temporary directory."""
+    docs = test_directory / 'help-docs'
+    (docs / 'system').mkdir(parents=True)
+    (docs / 'modules').mkdir()
+    (docs / 'index.md').write_text('# Home\n\nWelcome to WROLPi help.\n')
+    (docs / 'system/logs.md').write_text(
+        '# Logs\n\n## Reading logs\n\nUse journalctl to read service logs on a WROLPi.\n')
+    (docs / 'modules/videos.md').write_text(
+        '# Videos\n\n## Downloading\n\nVideos are downloaded with yt-dlp into channels.\n'
+        + ('More text about videos. ' * 400))  # > one page
+    monkeypatch.setenv('HELP_DOCS_DIR', str(docs))
+    help_docs._CACHE.clear()
+    try:
+        yield docs
+    finally:
+        help_docs._CACHE.clear()
+
+
+def test_search_help_scoring(help_corpus):
+    """Title matches outrank body matches; snippets show context."""
+    results, total = help_docs.search_help('logs')
+    assert total == 1
+    assert results[0]['doc'] == 'system/logs'
+    assert results[0]['title'] == 'Logs'
+    assert results[0]['link'] == '/system/logs/'
+    assert 'journalctl' in results[0]['snippet'] or 'Logs' in results[0]['snippet']
+
+    # 'videos' matches the videos doc far better than any other.
+    results, _ = help_docs.search_help('downloading videos')
+    assert results[0]['doc'] == 'modules/videos'
+
+    # No matches.
+    results, total = help_docs.search_help('xyzzy')
+    assert results == [] and total == 0
+
+
+def test_help_doc_cache_invalidation(help_corpus):
+    """Edited files are re-parsed; deleted files leave the corpus."""
+    results, _ = help_docs.search_help('welcome')
+    assert results[0]['doc'] == 'index'
+    assert results[0]['link'] == '/'
+
+    (help_corpus / 'index.md').write_text('# Home\n\nGreetings, traveler.\n')
+    import os
+    os.utime(help_corpus / 'index.md', (0, 0))  # force a distinct mtime
+    results, total = help_docs.search_help('welcome')
+    assert total == 0
+    results, _ = help_docs.search_help('greetings')
+    assert results[0]['doc'] == 'index'
+
+    (help_corpus / 'system/logs.md').unlink()
+    _, total = help_docs.search_help('logs')
+    assert total == 0
+
+
+def test_get_help_doc_refuses_traversal(help_corpus):
+    assert help_docs.get_help_doc('../secrets') is None
+    assert help_docs.get_help_doc('system/logs') is not None
+
+
+@pytest.mark.asyncio
+async def test_ai_help_endpoints(async_client, help_corpus):
+    """Help can be searched and pages read as paged markdown."""
+    content = dict(search_str='logs')
+    request, response = await async_client.post('/api/ai/help/search', content=json.dumps(content))
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['total'] == 1
+    result = response.json['results'][0]
+    assert result['doc'] == 'system/logs'
+
+    request, response = await async_client.get(f'/api/ai/help/{result["doc"]}')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['title'] == 'Logs'
+    assert 'journalctl' in response.json['content']
+
+    # Long pages are paged.
+    request, response = await async_client.get('/api/ai/help/modules/videos')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['next_offset'] == lib.PAGE_SIZE
+
+    # Unknown page.
+    request, response = await async_client.get('/api/ai/help/no/such/page')
+    assert response.status_code == HTTPStatus.NOT_FOUND
+
+    # Empty search is a clear error.
+    request, response = await async_client.post('/api/ai/help/search', content=json.dumps(dict()))
+    assert response.status_code == HTTPStatus.BAD_REQUEST
+
+
+@pytest.mark.asyncio
+async def test_ai_help_not_installed(async_client, monkeypatch, test_directory):
+    """A missing help directory returns empty results with a message, never a 500."""
+    monkeypatch.setenv('HELP_DOCS_DIR', str(test_directory / 'nope'))
+    help_docs._CACHE.clear()
+
+    content = dict(search_str='logs')
+    request, response = await async_client.post('/api/ai/help/search', content=json.dumps(content))
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['results'] == []
+    assert 'not installed' in response.json['message']
+
+    request, response = await async_client.get('/api/ai/help/system/logs')
+    assert response.status_code == HTTPStatus.NOT_FOUND