ai: add the /api/ai blueprint, the read-only LLM tool catalog

Lean, flat, LLM-shaped read-only endpoints over the library: global,
video, archive, and doc search; paged captions, archive text, and zim
entries; collections, inventories, and a media-directory text reader.
Links are server-computed; big content is paged, never truncated.
This ports the formatting layer from wrolpi_mcp (untouched for now)
into tested code.  First phase of ai.md.

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

conftest.py

@@ -1,3 +1,4 @@
+from modules.ai.conftest import *  # noqa
 from modules.archive.conftest import *  # noqa
 from modules.inventory.conftest import *  # noqa
 from modules.videos.conftest import *  # noqa

modules/ai/__init__.py

@@ -0,0 +1 @@
+from modules.ai.api import ai_bp  # noqa

modules/ai/api.py

@@ -0,0 +1,361 @@
+"""The /api/ai blueprint — read-only, LLM-shaped endpoints.
+
+This blueprint IS the AI tool catalog: the agent loop generates its tool definitions from these
+endpoints' OpenAPI operations, and the external MCP server proxies them.  Every endpoint is
+read-only; descriptions are written for the model, not for humans.
+"""
+from http import HTTPStatus
+
+from sanic import Blueprint, Request
+from sanic_ext import validate
+from sanic_ext.extensions.openapi import openapi
+
+from modules.archive import lib as archive_lib
+from modules.docs.lib import _doc_response, _get_doc, _search_docs
+from modules.inventory.common import get_inventory_configs
+from modules.inventory.errors import UnknownInventory
+from modules.videos.video import lib as videos_lib
+from modules.zim import lib as zim_lib
+from wrolpi.api_utils import json_response
+from wrolpi.collections.lib import search_collections
+from wrolpi.common import api_param_limiter, get_media_directory, logger
+from wrolpi.errors import InvalidFile, SearchEmpty, UnknownFile
+from wrolpi.files.lib import search_files
+from wrolpi.schema import JSONErrorResponse
+from . import lib, schema
+
+ai_bp = Blueprint('AI', url_prefix='/api/ai')
+
+logger = logger.getChild(__name__)
+
+ai_limiter = api_param_limiter(25, default=5)
+
+# Refuse to serve files this large through the text reader; real content is paged anyway.
+MAX_TEXT_FILE_SIZE = 10 * 1024 * 1024
+
+
+def _offset(request: Request) -> int:
+    try:
+        return max(int(request.args.get('offset', 0)), 0)
+    except ValueError:
+        return 0
+
+
+@ai_bp.post('/search')
+@openapi.definition(
+    summary='Search all WROLPi content',
+    description='Search everything in the library at once: videos, archived web pages, ebooks, PDFs, and other'
+                ' files. Returns lean results with a WROLPi link for each item. Prefer the type-specific search'
+                ' (videos/archives/docs) when the user asks about one kind of content. When total is large,'
+                ' narrow the query instead of paging.',
+    body=schema.AISearchRequest,
+)
+@openapi.operation('search_all')
+@openapi.response(HTTPStatus.OK, schema.AISearchResponse)
+@validate(schema.AISearchRequest)
+async def search_all(_: Request, body: schema.AISearchRequest):
+    file_groups, total = search_files(
+        body.search_str,
+        ai_limiter(body.limit),
+        body.offset or 0,
+        mimetypes=body.mimetypes,
+        tag_names=body.tag_names,
+        headline=True,
+    )
+    return json_response(lib.format_file_groups(file_groups, total))
+
+
+@ai_bp.post('/videos/search')
+@openapi.definition(
+    summary='Search videos',
+    description='Search downloaded videos (and audio) by title and captions. Returns lean results with a'
+                ' WROLPi link and a captions_link for each video. When total is large, narrow the query'
+                ' instead of paging.',
+    body=schema.AIVideoSearchRequest,
+)
+@openapi.operation('search_videos')
+@openapi.response(HTTPStatus.OK, schema.AISearchResponse)
+@validate(schema.AIVideoSearchRequest)
+async def search_videos(_: Request, body: schema.AIVideoSearchRequest):
+    file_groups, total = videos_lib.search_videos(
+        search_str=body.search_str,
+        offset=body.offset or 0,
+        limit=ai_limiter(body.limit),
+        channel_id=body.channel_id,
+        tag_names=body.tag_names,
+        headline=True,
+    )
+    return json_response(lib.format_file_groups(file_groups, total))
+
+
+@ai_bp.get('/videos/<file_group_id:int>')
+@openapi.definition(
+    summary='Get one video',
+    description='Get details about one video by its ID from search results: title, channel, description,'
+                ' duration, link, and a captions_link to read what is said in the video.',
+)
+@openapi.operation('get_video')
+@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
+async def get_video(_: Request, file_group_id: int):
+    video, _prev, _next = videos_lib.get_video_for_app(file_group_id, skip_viewed=True)
+    result = lib.format_file_group(video, description_length=lib.DETAIL_DESCRIPTION_LENGTH)
+    return json_response(result)
+
+
+@ai_bp.get('/videos/<file_group_id:int>/captions')
+@openapi.definition(
+    summary='Get video captions',
+    description='Read the timestamped captions of a video by its ID; use this to learn what is said in a video'
+                ' without watching it. Long captions are paged: request again with offset=next_offset to'
+                ' continue reading.',
+)
+@openapi.operation('get_video_captions')
+@openapi.response(HTTPStatus.OK, schema.AIPagedTextResponse)
+@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
+async def get_video_captions(request: Request, file_group_id: int):
+    video = videos_lib.get_video(file_group_id)
+    text = lib.format_caption_chunks(video.get_caption_chunks())
+    return json_response(lib.paginate_text(text, _offset(request)))
+
+
+@ai_bp.post('/archives/search')
+@openapi.definition(
+    summary='Search archived web pages',
+    description='Search archived web pages (saved with SingleFile) by title and content. Returns lean results'
+                ' with a WROLPi link and a text_link to read each page. Filter with domain (e.g. "example.com")'
+                ' when the user asks about one site.',
+    body=schema.AIArchiveSearchRequest,
+)
+@openapi.operation('search_archives')
+@openapi.response(HTTPStatus.OK, schema.AISearchResponse)
+@validate(schema.AIArchiveSearchRequest)
+async def search_archives(_: Request, body: schema.AIArchiveSearchRequest):
+    file_groups, total = archive_lib.search_archives(
+        body.search_str,
+        body.domain,
+        ai_limiter(body.limit),
+        body.offset or 0,
+        None,  # order: default (most recently published)
+        body.tag_names,
+        headline=True,
+    )
+    return json_response(lib.format_file_groups(file_groups, total))
+
+
+@ai_bp.get('/archives/<file_group_id:int>/text')
+@openapi.definition(
+    summary='Read an archived web page',
+    description='Read the plain text content of an archived web page by its ID. Long pages are paged: request'
+                ' again with offset=next_offset to continue reading.',
+)
+@openapi.operation('get_archive_text')
+@openapi.response(HTTPStatus.OK, schema.AIPagedTextResponse)
+@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
+async def get_archive_text(request: Request, file_group_id: int):
+    archive = archive_lib.get_archive_by_file_group_id(request.ctx.session, file_group_id, skip_viewed=True)
+    text = lib.read_archive_text(archive)
+    if text is None:
+        raise UnknownFile(f'No readable text for archive {file_group_id}')
+    return json_response(lib.paginate_text(text, _offset(request)))
+
+
+@ai_bp.post('/docs/search')
+@openapi.definition(
+    summary='Search documents and ebooks',
+    description='Search documents (ebooks, PDFs, comics, office docs) by title, content, author, or subject.'
+                ' Returns lean results with a WROLPi link for each document. Provide at least one of'
+                ' search_str, author, subject, or tag_names.',
+    body=schema.AIDocSearchRequest,
+)
+@openapi.operation('search_docs')
+@openapi.response(HTTPStatus.OK, schema.AISearchResponse)
+@validate(schema.AIDocSearchRequest)
+async def search_docs(_: Request, body: schema.AIDocSearchRequest):
+    file_groups, total = _search_docs(
+        search_str=body.search_str,
+        author=body.author,
+        subject=body.subject,
+        language=body.language,
+        mimetype=body.mimetype,
+        limit=ai_limiter(body.limit),
+        offset=body.offset or 0,
+        order_by='rank' if body.search_str else 'published_datetime',
+        tag_names=body.tag_names,
+    )
+    return json_response(lib.format_file_groups(file_groups, total))
+
+
+@ai_bp.get('/docs/<file_group_id:int>')
+@openapi.definition(
+    summary='Get one document',
+    description='Get details about one document (ebook, PDF, comic, etc.) by its ID from search results:'
+                ' title, author, subject, description, and its WROLPi link.',
+)
+@openapi.operation('get_doc')
+@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
+async def get_doc(request: Request, file_group_id: int):
+    doc = _get_doc(request.ctx.session, file_group_id)
+    response = _doc_response(doc)
+    file_group = response['file_group']
+    # Merge the Doc's details into the FileGroup dict so the lean formatter can use them.
+    file_group['doc'] = response['doc']
+    result = lib.format_file_group(file_group, description_length=lib.DETAIL_DESCRIPTION_LENGTH)
+    return json_response(result)
+
+
+@ai_bp.get('/zims')
+@openapi.definition(
+    summary='List Zim encyclopedias',
+    description='List the Zim encyclopedias in the library (Wikipedia, Wiktionary, etc.). Returns the zim_id'
+                ' needed to search a specific Zim or read one of its entries.',
+)
+@openapi.operation('list_zims')
+async def list_zims(request: Request):
+    zims = zim_lib.get_zims(request.ctx.session)
+    results = []
+    for zim in zims:
+        metadata = lib.zim_metadata_dict(zim.zim_metadata)
+        results.append({k: v for k, v in dict(
+            id=zim.id,
+            title=metadata.get('title'),
+            creator=metadata.get('creator'),
+            description=metadata.get('description'),
+            size=zim.file_group.size if zim.file_group else None,
+            auto_search=zim.auto_search,
+        ).items() if v is not None})
+    return json_response(dict(results=results, total=len(results)))
+
+
+@ai_bp.post('/zims/search')
+@openapi.definition(
+    summary='Search Zim encyclopedias',
+    description='Search Zim encyclopedias (Wikipedia, etc.). Without zim_id this searches only the Zims that'
+                ' have search-by-default enabled; pass zim_id (from list_zims) to search one specific Zim.'
+                ' Use the returned link (or zim_id and path) to read a full entry.',
+    body=schema.AIZimSearchRequest,
+)
+@openapi.operation('search_zims')
+@openapi.response(HTTPStatus.OK, schema.AISearchResponse)
+@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
+@validate(schema.AIZimSearchRequest)
+async def search_zims(request: Request, body: schema.AIZimSearchRequest):
+    if not body.search_str:
+        raise SearchEmpty()
+    limit = ai_limiter(body.limit)
+    offset = body.offset or 0
+    if body.zim_id:
+        result = zim_lib.headline_zim(request.ctx.session, body.search_str, body.zim_id, offset=offset, limit=limit)
+        for entry in result.get('search') or []:
+            entry['zim_id'] = body.zim_id
+        zim_results = [result]
+    else:
+        zim_results = zim_lib.search_all_zims(request.ctx.session, body.search_str, offset=offset, limit=limit)
+    return json_response(lib.format_zim_search(zim_results))
+
+
+@ai_bp.get('/zims/<zim_id:int>/entry')
+@openapi.definition(
+    summary='Read a Zim entry',
+    description='Read one article from a Zim encyclopedia as plain text. Pass the path from Zim search'
+                ' results. Long articles are paged: request again with offset=next_offset to continue'
+                ' reading.',
+)
+@openapi.operation('get_zim_entry')
+@openapi.response(HTTPStatus.OK, schema.AIPagedTextResponse)
+@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
+async def get_zim_entry(request: Request, zim_id: int):
+    path = request.args.get('path')
+    if not path:
+        raise InvalidFile('path query parameter is required')
+    entry = zim_lib.get_entry(path, zim_id)
+    html = bytes(entry.get_item().content).decode('UTF-8', errors='replace')
+    result = lib.paginate_text(lib.html_to_text(html), _offset(request))
+    result['link'] = lib.zim_entry_link(zim_id, path)
+    return json_response(result)
+
+
+@ai_bp.get('/collections')
+@openapi.definition(
+    summary='List collections',
+    description='List collections in the library. Filter with kind: "channel" (video channels), "domain"'
+                ' (archived websites), or "playlist". Use a channel\'s id with search_videos, or a domain\'s'
+                ' name with search_archives.',
+)
+@openapi.operation('list_collections')
+async def list_collections(request: Request):
+    kind = request.args.get('kind')
+    collections = search_collections(request.ctx.session, kind=kind)
+    results = []
+    for collection in collections:
+        results.append({k: v for k, v in dict(
+            id=collection.get('id'),
+            name=collection.get('name'),
+            kind=collection.get('kind'),
+            directory=collection.get('directory'),
+        ).items() if v is not None})
+    return json_response(dict(results=results, total=len(results)))
+
+
+@ai_bp.get('/inventories')
+@openapi.definition(
+    summary='List inventories',
+    description='List the inventories (food storage, emergency supplies, etc.). Use the slug with'
+                ' get_inventory to read the items.',
+)
+@openapi.operation('list_inventories')
+async def list_inventories(_: Request):
+    inventories = get_inventory_configs().all_inventories()
+    results = [{k: v for k, v in dict(
+        slug=i.get('slug'),
+        name=i.get('name'),
+        type=i.get('type'),
+        item_count=len(i.get('items') or []),
+    ).items() if v is not None} for i in inventories]
+    return json_response(dict(results=results, total=len(results)))
+
+
+@ai_bp.get('/inventories/<slug:str>')
+@openapi.definition(
+    summary='Get one inventory',
+    description='Get one inventory in full (its fields and every item) by its slug from list_inventories.',
+)
+@openapi.operation('get_inventory')
+@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
+async def get_inventory(_: Request, slug: str):
+    inventory = get_inventory_configs().get_inventory(slug)
+    if inventory is None:
+        raise UnknownInventory(f'No inventory: {slug}')
+    return json_response(dict(inventory=inventory))
+
+
+@ai_bp.get('/files/read')
+@openapi.definition(
+    summary='Read a text file',
+    description='Read a plain-text file from the media directory by its relative path (from search result'
+                ' links or collection directories). Only text files can be read. Long files are paged:'
+                ' request again with offset=next_offset to continue reading.',
+)
+@openapi.operation('read_file')
+@openapi.response(HTTPStatus.OK, schema.AIPagedTextResponse)
+@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
+async def read_file(request: Request):
+    relative = request.args.get('path')
+    if not relative:
+        raise InvalidFile('path query parameter is required')
+
+    media_directory = get_media_directory().resolve()
+    path = (media_directory / relative.lstrip('/')).resolve()
+    if not str(path).startswith(f'{media_directory}/'):
+        raise InvalidFile('Cannot read outside the media directory')
+    # The config directory can contain secrets (Wi-Fi credentials, API keys); never serve it.
+    if str(path).startswith(f'{media_directory / "config"}/'):
+        raise InvalidFile('Cannot read config files')
+    if not path.is_file():
+        raise UnknownFile(f'No such file: {relative}')
+    if path.stat().st_size > MAX_TEXT_FILE_SIZE:
+        raise InvalidFile('File is too large to read')
+
+    data = path.read_bytes()
+    if b'\x00' in data[:8192]:
+        raise InvalidFile('Not a text file')
+    return json_response(lib.paginate_text(data.decode('UTF-8', errors='replace'), _offset(request)))

modules/ai/conftest.py

@@ -0,0 +1,3 @@
+# AI module test fixtures.  The docs conftest is not star-imported by the root conftest, but the
+# AI endpoints cover docs; import its fixtures so AI tests can use doc_factory.
+from modules.docs.conftest import *  # noqa

modules/ai/lib.py

@@ -0,0 +1,219 @@
+"""Formatting helpers for the /api/ai blueprint — the LLM tool catalog.
+
+Results are lean, flat JSON with server-computed links so the model never constructs URLs.
+Links are relative to the WROLPi host (e.g. /videos/123); the caller (frontend, MCP client)
+prefixes its own base URL.
+"""
+import pathlib
+from typing import Optional
+
+from bs4 import BeautifulSoup
+
+from wrolpi.common import get_relative_to_media_directory, logger
+
+logger = logger.getChild(__name__)
+
+# Big content (captions, archive text, zim entries, help docs) is paged in chunks of this many
+# characters, never silently truncated.  Sized for small (8k-token) model contexts.
+PAGE_SIZE = 4_000
+# Description length in a detail response; descriptions in search listings are shorter.
+DETAIL_DESCRIPTION_LENGTH = 1_000
+LISTING_DESCRIPTION_LENGTH = 300
+
+
+def _relative_path(path) -> Optional[str]:
+    """Return the media-relative string of a path from a FileGroup JSON dict, if possible."""
+    if not path:
+        return None
+    try:
+        return str(get_relative_to_media_directory(path))
+    except Exception:
+        return str(path)
+
+
+def file_group_kind(fg: dict) -> str:
+    """Categorize a FileGroup JSON dict for the model: video/archive/doc/image/file."""
+    mimetype = fg.get('mimetype') or ''
+    if fg.get('model') == 'video' or isinstance(fg.get('video'), dict):
+        return 'video'
+    if fg.get('model') == 'archive' or isinstance(fg.get('archive'), dict):
+        return 'archive'
+    if fg.get('model') == 'doc':
+        return 'doc'
+    if mimetype.startswith('image/'):
+        return 'image'
+    return 'file'
+
+
+def wrolpi_link(fg: dict) -> Optional[str]:
+    """Build the relative WROLPi UI link for a FileGroup JSON dict.
+
+    Videos/docs/archives deep-link by FileGroup ID; epubs open in the reader; everything else is
+    served directly from /media/ or /download/.
+    """
+    fg_id = fg.get('id')
+    kind = file_group_kind(fg)
+    if kind == 'video' and fg_id:
+        return f'/videos/{fg_id}'
+    if kind == 'doc' and fg_id:
+        return f'/docs/{fg_id}'
+    if kind == 'archive' and fg_id:
+        return f'/archives/{fg_id}'
+
+    mimetype = fg.get('mimetype') or ''
+    primary_path = _relative_path(fg.get('primary_path'))
+    if not primary_path:
+        return None
+    if mimetype.startswith('application/epub'):
+        return f'/epub/epub.html?url=/download/{primary_path}'
+    if mimetype.startswith('application/x-mobipocket-ebook'):
+        return f'/download/{primary_path}'
+    # Caddy serves any media file at /media/{path}.
+    return f'/media/{primary_path}'
+
+
+def _truncate(text: Optional[str], length: int) -> Optional[str]:
+    if text and len(text) > length:
+        return text[:length] + '…'
+    return text
+
+
+def format_file_group(fg: dict, description_length: int = LISTING_DESCRIPTION_LENGTH) -> dict:
+    """Format a FileGroup JSON dict into the lean, flat shape the model consumes.
+
+    Empty fields are omitted so results stay compact on small contexts.
+    """
+    kind = file_group_kind(fg)
+    result = dict(
+        id=fg.get('id'),
+        kind=kind,
+        title=fg.get('title') or fg.get('name'),
+        link=wrolpi_link(fg),
+        mimetype=fg.get('mimetype'),
+        size=fg.get('size'),
+        published=fg.get('published_datetime'),
+        author=fg.get('author'),
+        tags=fg.get('tags') or None,
+        url=fg.get('url'),
+    )
+
+    # FTS headline (content match context), if the search requested headlines.
+    headline = fg.get('d_headline') or fg.get('b_headline') or fg.get('c_headline')
+    if headline:
+        result['headline'] = headline
+
+    video = fg.get('video')
+    if kind == 'video':
+        result['duration'] = fg.get('length')
+        result['captions_link'] = f'/api/ai/videos/{fg["id"]}/captions' if fg.get('id') else None
+        if isinstance(video, dict):
+            channel = video.get('channel')
+            if isinstance(channel, dict):
+                result['channel'] = channel.get('name') or channel.get('id')
+            elif video.get('channel_id'):
+                result['channel_id'] = video['channel_id']
+            result['description'] = _truncate(video.get('description'), description_length)
+
+    if kind == 'archive' and fg.get('id'):
+        result['text_link'] = f'/api/ai/archives/{fg["id"]}/text'
+
+    if kind == 'doc':
+        doc = fg.get('doc') or {}
+        result['subject'] = doc.get('subject')
+        result['language'] = doc.get('language')
+        result['pages'] = doc.get('page_count')
+        result['description'] = _truncate(doc.get('description'), description_length)
+
+    return {k: v for k, v in result.items() if v is not None}
+
+
+def format_file_groups(file_groups: list, total: int) -> dict:
+    """Format search results: lean items plus the total so the model narrows instead of paging."""
+    return dict(results=[format_file_group(i) for i in file_groups], total=total)
+
+
+def paginate_text(text: Optional[str], offset: int = 0) -> dict:
+    """Page big content: ~PAGE_SIZE chars plus next_offset/total_chars, never silent truncation."""
+    text = text or ''
+    offset = max(int(offset or 0), 0)
+    end = offset + PAGE_SIZE
+    next_offset = end if end < len(text) else None
+    return dict(content=text[offset:end], next_offset=next_offset, total_chars=len(text))
+
+
+def html_to_text(html: str) -> str:
+    """Strip HTML down to clean plain text for the model.  Pagination handles size."""
+    soup = BeautifulSoup(html, 'html.parser')
+    for tag in soup(['script', 'style', 'nav', 'footer', 'header']):
+        tag.decompose()
+
+    text = soup.get_text(separator='\n', strip=True)
+
+    # Collapse multiple blank lines.
+    lines = []
+    prev_blank = False
+    for line in text.splitlines():
+        stripped = line.strip()
+        if not stripped:
+            if not prev_blank:
+                lines.append('')
+            prev_blank = True
+        else:
+            lines.append(stripped)
+            prev_blank = False
+    return '\n'.join(lines)
+
+
+def format_caption_chunks(chunks: Optional[list]) -> str:
+    """Render caption chunks ({start_seconds, text}) as timestamped lines."""
+    if not chunks:
+        return ''
+    lines = []
+    for chunk in chunks:
+        minutes, seconds = divmod(int(chunk.get('start_seconds') or 0), 60)
+        hours, minutes = divmod(minutes, 60)
+        lines.append(f'[{hours:02d}:{minutes:02d}:{seconds:02d}] {chunk.get("text") or ""}')
+    return '\n'.join(lines)
+
+
+def zim_entry_link(zim_id: int, path: str) -> str:
+    return f'/api/zim/{zim_id}/entry/{path}'
+
+
+def zim_metadata_dict(metadata) -> dict:
+    """Zim lib functions return a ZimMetadata object; normalize it to a dict."""
+    if metadata is None:
+        return {}
+    return metadata if isinstance(metadata, dict) else metadata.__json__()
+
+
+def format_zim_search(zim_results: list) -> dict:
+    """Flatten per-Zim search results into lean entries with links."""
+    results = []
+    total = 0
+    for zim_result in zim_results:
+        metadata = zim_metadata_dict(zim_result.get('metadata'))
+        zim_title = metadata.get('title')
+        total += zim_result.get('estimate') or 0
+        for entry in zim_result.get('search') or []:
+            zim_id = entry.get('zim_id')
+            path = entry.get('path')
+            result = dict(
+                zim_id=zim_id,
+                zim=zim_title,
+                path=path,
+                title=entry.get('title'),
+                headline=entry.get('headline'),
+                link=zim_entry_link(zim_id, path) if zim_id and path else None,
+            )
+            results.append({k: v for k, v in result.items() if v is not None})
+    return dict(results=results, total=total)
+
+
+def read_archive_text(archive) -> Optional[str]:
+    """Return the readable plain text of an Archive: readability text, else stripped singlefile."""
+    if (path := archive.readability_txt_path) and pathlib.Path(path).is_file():
+        return pathlib.Path(path).read_text(errors='replace')
+    if (path := archive.singlefile_path) and pathlib.Path(path).is_file():
+        return html_to_text(pathlib.Path(path).read_text(errors='replace'))
+    return None

modules/ai/schema.py

@@ -0,0 +1,62 @@
+import dataclasses
+from typing import Optional
+
+
+@dataclasses.dataclass
+class AISearchRequest:
+    search_str: Optional[str] = None
+    limit: int = 5
+    offset: int = 0
+    mimetypes: Optional[list] = None
+    tag_names: Optional[list] = None
+
+
+@dataclasses.dataclass
+class AIVideoSearchRequest:
+    search_str: Optional[str] = None
+    limit: int = 5
+    offset: int = 0
+    channel_id: Optional[int] = None
+    tag_names: Optional[list] = None
+
+
+@dataclasses.dataclass
+class AIArchiveSearchRequest:
+    search_str: Optional[str] = None
+    limit: int = 5
+    offset: int = 0
+    domain: Optional[str] = None
+    tag_names: Optional[list] = None
+
+
+@dataclasses.dataclass
+class AIDocSearchRequest:
+    search_str: Optional[str] = None
+    author: Optional[str] = None
+    subject: Optional[str] = None
+    language: Optional[str] = None
+    mimetype: Optional[str] = None
+    limit: int = 5
+    offset: int = 0
+    tag_names: Optional[list] = None
+
+
+@dataclasses.dataclass
+class AIZimSearchRequest:
+    search_str: Optional[str] = None
+    zim_id: Optional[int] = None
+    limit: int = 5
+    offset: int = 0
+
+
+@dataclasses.dataclass
+class AISearchResponse:
+    results: list = dataclasses.field(default_factory=list)
+    total: int = 0
+
+
+@dataclasses.dataclass
+class AIPagedTextResponse:
+    content: str = ''
+    next_offset: Optional[int] = None
+    total_chars: int = 0

modules/ai/test/test_api.py

@@ -0,0 +1,267 @@
+"""Tests for the /api/ai blueprint — the LLM tool catalog."""
+import json
+from http import HTTPStatus
+
+import pytest
+
+from modules.ai import lib
+
+
+@pytest.mark.asyncio
+async def test_ai_search_all(async_client, test_session, video_factory, archive_factory):
+    """Global search returns lean results of every kind with links and a total."""
+    video_factory(title='canning tomatoes')
+    archive_factory(domain='example.com', title='canning peppers', contents='all about canning peppers')
+    test_session.commit()
+
+    request, response = await async_client.post('/api/ai/search', content=json.dumps(dict(search_str='canning')))
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['total'] == 2
+    kinds = {i['kind'] for i in response.json['results']}
+    assert kinds == {'video', 'archive'}
+    for result in response.json['results']:
+        assert result['link']
+        assert result['id']
+
+
+@pytest.mark.asyncio
+async def test_ai_search_all_limit(async_client, test_session, archive_factory):
+    """The limit is clamped and the total reports all matches so the model narrows, not pages."""
+    for i in range(3):
+        archive_factory(domain='example.com', title=f'gardening {i}', contents='gardening guide')
+    test_session.commit()
+
+    content = dict(search_str='gardening', limit=1)
+    request, response = await async_client.post('/api/ai/search', content=json.dumps(content))
+    assert response.status_code == HTTPStatus.OK
+    assert len(response.json['results']) == 1
+    assert response.json['total'] == 3
+
+    # An excessive limit is clamped to the maximum.
+    content = dict(search_str='gardening', limit=10_000)
+    request, response = await async_client.post('/api/ai/search', content=json.dumps(content))
+    assert response.status_code == HTTPStatus.OK
+
+
+@pytest.mark.asyncio
+async def test_ai_search_videos(async_client, test_session, video_factory):
+    """Video search returns lean video results with captions links."""
+    video_factory(title='sourdough starter', with_caption_file=True)
+    video_factory(title='unrelated')
+    test_session.commit()
+
+    content = dict(search_str='sourdough')
+    request, response = await async_client.post('/api/ai/videos/search', content=json.dumps(content))
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['total'] == 1
+    result = response.json['results'][0]
+    assert result['kind'] == 'video'
+    assert result['link'] == f'/videos/{result["id"]}'
+    assert result['captions_link'] == f'/api/ai/videos/{result["id"]}/captions'
+
+
+@pytest.mark.asyncio
+async def test_ai_get_video(async_client, test_session, video_factory):
+    """A single video can be fetched by its FileGroup ID."""
+    video = video_factory(title='wood stove install')
+    test_session.commit()
+    file_group_id = video.file_group_id
+
+    request, response = await async_client.get(f'/api/ai/videos/{file_group_id}')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['id'] == file_group_id
+    assert response.json['kind'] == 'video'
+    assert response.json['link'] == f'/videos/{file_group_id}'
+
+    # Fetching does not mark the video viewed.
+    test_session.expire_all()
+    assert video.file_group.viewed is None
+
+    request, response = await async_client.get('/api/ai/videos/123456')
+    assert response.status_code == HTTPStatus.NOT_FOUND
+
+
+@pytest.mark.asyncio
+async def test_ai_get_video_captions(async_client, test_session, video_factory):
+    """Captions are returned as timestamped, paged text."""
+    video = video_factory(title='captioned', with_caption_file=True)
+    test_session.commit()
+
+    request, response = await async_client.get(f'/api/ai/videos/{video.file_group_id}/captions')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['content'].startswith('[00:00:0')
+    assert response.json['total_chars'] > 0
+
+    # An offset beyond the end returns an empty page, never an error.
+    offset = response.json['total_chars'] + 100
+    request, response = await async_client.get(f'/api/ai/videos/{video.file_group_id}/captions?offset={offset}')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['content'] == ''
+    assert response.json['next_offset'] is None
+
+
+@pytest.mark.asyncio
+async def test_ai_search_archives(async_client, test_session, archive_factory):
+    """Archive search supports the domain filter and returns text links."""
+    archive_factory(domain='example.com', title='pressure canning', contents='pressure canning guide')
+    archive_factory(domain='other.org', title='water bath canning', contents='water bath canning guide')
+    test_session.commit()
+
+    content = dict(search_str='canning', domain='example.com')
+    request, response = await async_client.post('/api/ai/archives/search', content=json.dumps(content))
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['total'] == 1
+    result = response.json['results'][0]
+    assert result['kind'] == 'archive'
+    assert result['text_link'] == f'/api/ai/archives/{result["id"]}/text'
+
+
+@pytest.mark.asyncio
+async def test_ai_get_archive_text(async_client, test_session, archive_factory):
+    """Archive text is served from the readability text file and paged."""
+    contents = 'word ' * 2_000  # 10,000 chars, more than two pages.
+    archive = archive_factory(domain='example.com', title='long read', contents=contents)
+    test_session.commit()
+    file_group_id = archive.file_group_id
+
+    request, response = await async_client.get(f'/api/ai/archives/{file_group_id}/text')
+    assert response.status_code == HTTPStatus.OK
+    assert len(response.json['content']) == lib.PAGE_SIZE
+    assert response.json['next_offset'] == lib.PAGE_SIZE
+    assert response.json['total_chars'] == len(contents)
+
+    # The second page continues where the first ended.
+    request, response = await async_client.get(
+        f'/api/ai/archives/{file_group_id}/text?offset={response.json["next_offset"]}')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['content'] == contents[lib.PAGE_SIZE:2 * lib.PAGE_SIZE]
+
+    request, response = await async_client.get('/api/ai/archives/123456/text')
+    assert response.status_code == HTTPStatus.NOT_FOUND
+
+
+@pytest.mark.asyncio
+async def test_ai_search_docs(async_client, test_session, test_directory, example_epub, refresh_files):
+    """Docs can be searched and fetched with links."""
+    await refresh_files()
+
+    content = dict(search_str='WROLPi')
+    request, response = await async_client.post('/api/ai/docs/search', content=json.dumps(content))
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['total'] >= 1
+    result = response.json['results'][0]
+    assert result['kind'] == 'doc'
+    assert result['link'] == f'/docs/{result["id"]}'
+
+    request, response = await async_client.get(f'/api/ai/docs/{result["id"]}')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['id'] == result['id']
+    assert response.json['kind'] == 'doc'
+
+
+@pytest.mark.asyncio
+async def test_ai_zims(async_client, test_session, test_zim):
+    """Zims can be listed, searched, and their entries read as paged text."""
+    request, response = await async_client.get('/api/ai/zims')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['total'] == 1
+    zim_id = response.json['results'][0]['id']
+
+    # Search one specific Zim.
+    content = dict(search_str='item', zim_id=zim_id)
+    request, response = await async_client.post('/api/ai/zims/search', content=json.dumps(content))
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['total'] >= 1
+    result = response.json['results'][0]
+    assert result['zim_id'] == zim_id
+    assert result['link'] == f'/api/zim/{zim_id}/entry/{result["path"]}'
+
+    # An empty search is a clear error for the model.
+    request, response = await async_client.post('/api/ai/zims/search', content=json.dumps(dict()))
+    assert response.status_code == HTTPStatus.BAD_REQUEST
+
+    # Read the entry the search found.
+    request, response = await async_client.get(f'/api/ai/zims/{zim_id}/entry?path={result["path"]}')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['content']
+    assert response.json['link'] == f'/api/zim/{zim_id}/entry/{result["path"]}'
+
+    # The path is required.
+    request, response = await async_client.get(f'/api/ai/zims/{zim_id}/entry')
+    assert response.status_code == HTTPStatus.BAD_REQUEST
+
+
+@pytest.mark.asyncio
+async def test_ai_list_collections(async_client, test_session, archive_factory):
+    """Collections can be listed and filtered by kind."""
+    archive_factory(domain='example.com', title='a page', contents='contents')
+    test_session.commit()
+
+    request, response = await async_client.get('/api/ai/collections?kind=domain')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['total'] == 1
+    assert response.json['results'][0]['name'] == 'example.com'
+    assert response.json['results'][0]['kind'] == 'domain'
+
+    request, response = await async_client.get('/api/ai/collections?kind=channel')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['total'] == 0
+
+
+@pytest.mark.asyncio
+async def test_ai_inventories(async_client, food_inventory_factory):
+    """Inventories are listed lean, and fetched in full by slug."""
+    slug = food_inventory_factory(items=[dict(name='rice', count=2)])
+
+    request, response = await async_client.get('/api/ai/inventories')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['total'] >= 1
+    lean = next(i for i in response.json['results'] if i['slug'] == slug)
+    assert lean['item_count'] == 1
+    assert 'items' not in lean
+
+    request, response = await async_client.get(f'/api/ai/inventories/{slug}')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['inventory']['items'][0]['name'] == 'rice'
+
+    request, response = await async_client.get('/api/ai/inventories/no-such-inventory')
+    assert response.status_code == HTTPStatus.NOT_FOUND
+
+
+@pytest.mark.asyncio
+async def test_ai_read_file(async_client, test_directory):
+    """Text files in the media directory can be read; everything else is refused."""
+    (test_directory / 'notes').mkdir()
+    (test_directory / 'notes/todo.txt').write_text('buy more rice\n' * 400)  # > one page
+
+    request, response = await async_client.get('/api/ai/files/read?path=notes/todo.txt')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['content'].startswith('buy more rice')
+    assert response.json['next_offset'] == lib.PAGE_SIZE
+
+    # Missing path parameter.
+    request, response = await async_client.get('/api/ai/files/read')
+    assert response.status_code == HTTPStatus.BAD_REQUEST
+
+    # Missing file.
+    request, response = await async_client.get('/api/ai/files/read?path=notes/nope.txt')
+    assert response.status_code == HTTPStatus.NOT_FOUND
+
+    # Binary files are refused.
+    (test_directory / 'notes/blob.bin').write_bytes(b'\x00\x01\x02')
+    request, response = await async_client.get('/api/ai/files/read?path=notes/blob.bin')
+    assert response.status_code == HTTPStatus.BAD_REQUEST
+
+
+@pytest.mark.asyncio
+async def test_ai_read_file_refuses_escapes(async_client, test_directory):
+    """Reads outside the media directory, and of config files, are refused."""
+    # Path traversal.
+    request, response = await async_client.get('/api/ai/files/read?path=../secrets.txt')
+    assert response.status_code == HTTPStatus.BAD_REQUEST
+
+    # The config directory can contain secrets.
+    (test_directory / 'config').mkdir(exist_ok=True)
+    (test_directory / 'config/wrolpi.yaml').write_text('secret: hunter2')
+    request, response = await async_client.get('/api/ai/files/read?path=config/wrolpi.yaml')
+    assert response.status_code == HTTPStatus.BAD_REQUEST

modules/ai/test/test_lib.py

@@ -0,0 +1,130 @@
+"""Tests for the /api/ai formatting helpers."""
+from modules.ai import lib
+
+
+def test_wrolpi_link_video():
+    fg = dict(id=123, model='video', video=dict(), mimetype='video/mp4')
+    assert lib.wrolpi_link(fg) == '/videos/123'
+
+
+def test_wrolpi_link_doc():
+    fg = dict(id=5, model='doc', mimetype='application/pdf')
+    assert lib.wrolpi_link(fg) == '/docs/5'
+
+
+def test_wrolpi_link_archive():
+    fg = dict(id=7, model='archive', mimetype='text/html')
+    assert lib.wrolpi_link(fg) == '/archives/7'
+
+
+def test_wrolpi_link_epub(test_directory):
+    fg = dict(id=9, mimetype='application/epub+zip', primary_path=test_directory / 'books/foo.epub')
+    assert lib.wrolpi_link(fg) == '/epub/epub.html?url=/download/books/foo.epub'
+
+
+def test_wrolpi_link_mobi(test_directory):
+    fg = dict(id=9, mimetype='application/x-mobipocket-ebook', primary_path=test_directory / 'books/foo.mobi')
+    assert lib.wrolpi_link(fg) == '/download/books/foo.mobi'
+
+
+def test_wrolpi_link_media_fallback(test_directory):
+    fg = dict(id=2, mimetype='image/png', primary_path=test_directory / 'photos/a.png')
+    assert lib.wrolpi_link(fg) == '/media/photos/a.png'
+
+
+def test_wrolpi_link_no_path():
+    assert lib.wrolpi_link(dict(mimetype='image/png')) is None
+
+
+def test_format_file_group_video():
+    fg = dict(
+        id=42,
+        model='video',
+        title='Canning Basics',
+        mimetype='video/mp4',
+        size=1000,
+        length=4021,
+        published_datetime='2024-01-02',
+        tags=['food'],
+        video=dict(channel=dict(id=1, name='Homesteading'), description='x' * 500),
+        d_headline='pressure <b>canning</b>',
+    )
+    result = lib.format_file_group(fg)
+    assert result['id'] == 42
+    assert result['kind'] == 'video'
+    assert result['link'] == '/videos/42'
+    assert result['captions_link'] == '/api/ai/videos/42/captions'
+    assert result['channel'] == 'Homesteading'
+    assert result['duration'] == 4021
+    assert result['headline'] == 'pressure <b>canning</b>'
+    assert result['tags'] == ['food']
+    # Listing descriptions are truncated.
+    assert len(result['description']) == lib.LISTING_DESCRIPTION_LENGTH + 1
+    # Empty fields are omitted.
+    assert 'author' not in result and 'url' not in result
+
+
+def test_format_file_group_archive():
+    fg = dict(id=8, model='archive', title='Some Page', mimetype='text/html', url='https://example.com/a')
+    result = lib.format_file_group(fg)
+    assert result['kind'] == 'archive'
+    assert result['link'] == '/archives/8'
+    assert result['text_link'] == '/api/ai/archives/8/text'
+    assert result['url'] == 'https://example.com/a'
+
+
+def test_format_file_groups_total():
+    result = lib.format_file_groups([dict(id=1, model='archive')], 55)
+    assert result['total'] == 55
+    assert len(result['results']) == 1
+
+
+def test_paginate_text():
+    text = 'a' * (lib.PAGE_SIZE + 100)
+    page = lib.paginate_text(text, 0)
+    assert len(page['content']) == lib.PAGE_SIZE
+    assert page['next_offset'] == lib.PAGE_SIZE
+    assert page['total_chars'] == lib.PAGE_SIZE + 100
+
+    page = lib.paginate_text(text, page['next_offset'])
+    assert len(page['content']) == 100
+    assert page['next_offset'] is None
+
+    # Exact boundary has no next page.
+    page = lib.paginate_text('a' * lib.PAGE_SIZE, 0)
+    assert page['next_offset'] is None
+
+    # Empty text.
+    page = lib.paginate_text(None, 0)
+    assert page == dict(content='', next_offset=None, total_chars=0)
+
+
+def test_html_to_text():
+    html = '<html><head><style>.x{}</style><script>evil()</script></head>' \
+           '<body><nav>menu</nav><p>Hello</p><p></p><p>World</p></body></html>'
+    text = lib.html_to_text(html)
+    assert 'Hello' in text and 'World' in text
+    assert 'evil' not in text and 'menu' not in text and '.x{}' not in text
+
+
+def test_format_caption_chunks():
+    chunks = [dict(start_seconds=0, text='hello'), dict(start_seconds=3671, text='goodbye')]
+    text = lib.format_caption_chunks(chunks)
+    assert text == '[00:00:00] hello\n[01:01:11] goodbye'
+    assert lib.format_caption_chunks(None) == ''
+
+
+def test_format_zim_search():
+    zim_results = [dict(
+        metadata=dict(title='Test Zim'),
+        estimate=2,
+        search=[
+            dict(zim_id=1, path='one', title='One', headline='the first', rank=1),
+            dict(zim_id=1, path='two', title='Two', headline=None, rank=1),
+        ],
+    )]
+    result = lib.format_zim_search(zim_results)
+    assert result['total'] == 2
+    assert result['results'][0]['link'] == '/api/zim/1/entry/one'
+    assert result['results'][0]['zim'] == 'Test Zim'
+    assert 'headline' not in result['results'][1]

wrolpi/root_api.py

@@ -13,6 +13,7 @@ from sanic_ext.extensions.openapi import openapi
 from vininfo import Vin
 from vininfo.details._base import VinDetails
 
+from modules.ai.api import ai_bp
 from modules.archive.api import archive_bp
 from modules.docs.api import docs_bp
 from modules.flasher.api import flasher_bp
@@ -54,6 +55,7 @@ api_bp = Blueprint('RootAPI', url_prefix='/api')
 
 # Blueprints order here defines what order they are displayed in OpenAPI Docs.
 api_app.blueprint(api_bp)
+api_app.blueprint(ai_bp)
 api_app.blueprint(archive_bp)
 api_app.blueprint(collection_bp)  # Unified collection endpoints
 api_app.blueprint(config_bp)