diff --git a/conftest.py b/conftest.py index 04a24e10..1e08eef2 100644 --- a/conftest.py +++ b/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 diff --git a/modules/ai/__init__.py b/modules/ai/__init__.py new file mode 100644 index 00000000..8f05d9c0 --- /dev/null +++ b/modules/ai/__init__.py @@ -0,0 +1 @@ +from modules.ai.api import ai_bp # noqa diff --git a/modules/ai/api.py b/modules/ai/api.py new file mode 100644 index 00000000..191d5730 --- /dev/null +++ b/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/') +@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//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//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/') +@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//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/') +@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))) diff --git a/modules/ai/conftest.py b/modules/ai/conftest.py new file mode 100644 index 00000000..25fcf989 --- /dev/null +++ b/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 diff --git a/modules/ai/lib.py b/modules/ai/lib.py new file mode 100644 index 00000000..b780b333 --- /dev/null +++ b/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 diff --git a/modules/ai/schema.py b/modules/ai/schema.py new file mode 100644 index 00000000..0a2cc717 --- /dev/null +++ b/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 diff --git a/modules/ai/test/__init__.py b/modules/ai/test/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/modules/ai/test/test_api.py b/modules/ai/test/test_api.py new file mode 100644 index 00000000..ea0928a8 --- /dev/null +++ b/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 diff --git a/modules/ai/test/test_lib.py b/modules/ai/test/test_lib.py new file mode 100644 index 00000000..00826954 --- /dev/null +++ b/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 canning', + ) + 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 canning' + 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 = '' \ + '

Hello

World

' + 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] diff --git a/wrolpi/root_api.py b/wrolpi/root_api.py index 01d459c2..736bdfe0 100644 --- a/wrolpi/root_api.py +++ b/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)