mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 02:48:38 -04:00
ai: tags, downloads, map, and search-preview tools
Five new read-only /api/ai endpoints, each with a matching wrolpi_mcp proxy tool: list_tags (tags with what they are applied to, plus recent names), search_suggestions (matching channels/domains/authors/subjects and file/Zim result estimates), list_downloads (queue summary and lean recurring/one-time entries with relative destinations; status filter; errors keep their tail, which holds the exception message), get_map_overview (map files, search-index state, subscriptions, pins with map links), and search_places (place search with map links and optional proximity ranking). Research mode allowlists the library-facing four. Nothing here can start, stop, or retry a download. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Roland
parent
6cbc48c78e
commit
c5ca3dae78
6 changed files with 504 additions and 3 deletions
|
|
@ -4,6 +4,7 @@ This blueprint IS the AI tool catalog: the agent loop generates its tool definit
|
|||
endpoints' OpenAPI operations, and the external MCP server proxies them. Every endpoint is
|
||||
read-only; descriptions are written for the model, not for humans.
|
||||
"""
|
||||
import asyncio
|
||||
import mimetypes
|
||||
import pathlib
|
||||
from http import HTTPStatus
|
||||
|
|
@ -13,18 +14,26 @@ from sanic_ext import validate
|
|||
from sanic_ext.extensions.openapi import openapi
|
||||
|
||||
from modules.archive import lib as archive_lib
|
||||
from modules.archive.lib import search_domains_by_name
|
||||
from modules.docs.lib import search_authors_by_name, search_subjects_by_name
|
||||
from modules.map import lib as map_lib
|
||||
from modules.map import search as map_search
|
||||
from modules.map.pins import get_map_pins_config
|
||||
from modules.videos.channel.lib import search_channels_by_name
|
||||
from modules.zim.models import Zims
|
||||
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 import flags
|
||||
from wrolpi import flags, tags
|
||||
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, wrol_mode_enabled
|
||||
from wrolpi.common import api_param_limiter, get_media_directory, get_relative_to_media_directory, logger, \
|
||||
wrol_mode_enabled
|
||||
from wrolpi.downloader import download_manager
|
||||
from wrolpi.errors import InvalidFile, SearchEmpty, UnknownFile, ValidationError
|
||||
from wrolpi.files.lib import search_files, HIDDEN_DIRECTORIES, HIDDEN_FILES
|
||||
from wrolpi.files.lib import search_files, search_file_suggestion_count, HIDDEN_DIRECTORIES, HIDDEN_FILES
|
||||
from wrolpi.vars import DOCKERIZED, IS_RPI4
|
||||
from wrolpi.version import __version__
|
||||
from wrolpi.schema import JSONErrorResponse
|
||||
|
|
@ -577,6 +586,193 @@ async def ai_list_disks(_: Request):
|
|||
return json_response(ret)
|
||||
|
||||
|
||||
# Errors on downloads can be long tracebacks; the model only needs the gist.
|
||||
DOWNLOAD_ERROR_LENGTH = 300
|
||||
MAP_LINK_ZOOM = 12
|
||||
|
||||
|
||||
def _map_link(lat, lon) -> str:
|
||||
return f'/map?lat={lat}&lon={lon}&z={MAP_LINK_ZOOM}'
|
||||
|
||||
|
||||
@ai_bp.get('/tags')
|
||||
@openapi.definition(
|
||||
summary='List tags',
|
||||
description='List every tag in the library with how many files, Zim entries, channels, and domains carry'
|
||||
' it, plus the most recently used tag names. Use a tag name in the tag_names filter of the'
|
||||
' search tools. Users tag things they care about, so tags are a good map of their interests.',
|
||||
)
|
||||
@openapi.operation('list_tags')
|
||||
async def list_tags(_: Request):
|
||||
results = [dict(
|
||||
name=i['name'],
|
||||
file_groups=i['file_group_count'],
|
||||
zim_entries=i['zim_entry_count'],
|
||||
channels=i['channel_count'],
|
||||
domains=i['domain_count'],
|
||||
) for i in tags.get_tags()]
|
||||
return json_response(dict(results=results, recent=tags.get_recent_tags(), total=len(results)))
|
||||
|
||||
|
||||
def _format_download(download: dict) -> dict:
|
||||
error = download.get('error')
|
||||
if error and len(error) > DOWNLOAD_ERROR_LENGTH:
|
||||
# Errors are usually tracebacks; the exception message at the end is the useful part.
|
||||
error = '…' + error[-DOWNLOAD_ERROR_LENGTH:]
|
||||
destination = download.get('destination')
|
||||
return {k: v for k, v in dict(
|
||||
id=download.get('id'),
|
||||
url=download.get('url'),
|
||||
status=download.get('status'),
|
||||
downloader=download.get('downloader'),
|
||||
sub_downloader=download.get('sub_downloader'),
|
||||
frequency=download.get('frequency'),
|
||||
destination=str(get_relative_to_media_directory(destination)) if destination else None,
|
||||
collection_id=download.get('collection_id'),
|
||||
tag_names=download.get('tag_names') or None,
|
||||
next_download=download.get('next_download'),
|
||||
last_successful_download=download.get('last_successful_download'),
|
||||
error=error,
|
||||
).items() if v is not None}
|
||||
|
||||
|
||||
@ai_bp.get('/downloads')
|
||||
@openapi.definition(
|
||||
summary='List downloads',
|
||||
description='Read the download queue: a summary (pending count, whether downloading is disabled or'
|
||||
' stopped, daily limit), the recurring downloads (channels, feeds; frequency is in seconds),'
|
||||
' and the newest one-time downloads. Filter with status: new, pending, failed, deferred, or'
|
||||
' complete. Use this when the user asks what is downloading, why a download failed, or what'
|
||||
' is scheduled. You cannot start, stop, or retry downloads.',
|
||||
)
|
||||
@openapi.operation('list_downloads')
|
||||
@openapi.parameter('status', str, 'query')
|
||||
@openapi.parameter('limit', int, 'query')
|
||||
async def list_downloads(request: Request):
|
||||
status = request.args.get('status')
|
||||
try:
|
||||
limit = ai_limiter(int(request.args.get('limit', 0)) or None)
|
||||
except ValueError:
|
||||
raise ValidationError('limit must be an integer')
|
||||
summary = download_manager.get_summary()
|
||||
data = download_manager.get_fe_downloads()
|
||||
|
||||
def select(downloads: list) -> list:
|
||||
if status:
|
||||
downloads = [i for i in downloads if i.get('status') == status]
|
||||
return [_format_download(i) for i in downloads[:limit]]
|
||||
|
||||
return json_response(dict(
|
||||
summary=summary,
|
||||
recurring=select(data['recurring_downloads']),
|
||||
once=select(data['once_downloads']),
|
||||
pending_once=data.get('pending_once_downloads', 0),
|
||||
))
|
||||
|
||||
|
||||
@ai_bp.get('/map')
|
||||
@openapi.definition(
|
||||
summary='Get map overview',
|
||||
description='What maps this WROLPi has: the downloaded map files (regions) and whether each has a place'
|
||||
' search index, the map regions subscribed for updates, and the user\'s saved pins with a'
|
||||
' link to each. Use search_places to look up a town or landmark.',
|
||||
)
|
||||
@openapi.operation('get_map_overview')
|
||||
async def get_map_overview(request: Request):
|
||||
files = [dict(name=i['name'], path=i['path'], size=i['size'], has_search_index=i['has_search_index'])
|
||||
for i in map_lib.get_pmtiles_files()]
|
||||
pins = [dict(id=i.get('id'), label=i.get('label'), lat=i.get('lat'), lon=i.get('lon'),
|
||||
link=_map_link(i.get('lat'), i.get('lon')))
|
||||
for i in get_map_pins_config().pins]
|
||||
return json_response(dict(
|
||||
files=files,
|
||||
subscriptions=map_lib.get_map_subscriptions(request.ctx.session),
|
||||
pins=pins,
|
||||
search_indexes=map_search.get_search_status(),
|
||||
))
|
||||
|
||||
|
||||
@ai_bp.get('/map/search')
|
||||
@openapi.definition(
|
||||
summary='Search places on the map',
|
||||
description='Find towns, cities, and landmarks by name in the downloaded maps\' place index. Each result'
|
||||
' has coordinates and a WROLPi map link. Pass lat and lon to rank nearest first. Results are'
|
||||
' empty when no map has a search index (see get_map_overview).',
|
||||
)
|
||||
@openapi.operation('search_places')
|
||||
@openapi.parameter('q', str, 'query')
|
||||
@openapi.parameter('limit', int, 'query')
|
||||
@openapi.parameter('offset', int, 'query')
|
||||
@openapi.parameter('lat', float, 'query')
|
||||
@openapi.parameter('lon', float, 'query')
|
||||
async def search_places(request: Request):
|
||||
q = (request.args.get('q') or '').strip()
|
||||
if not q:
|
||||
raise ValidationError('q query parameter is required')
|
||||
try:
|
||||
limit = ai_limiter(int(request.args.get('limit', 0)) or None)
|
||||
lat = float(request.args.get('lat')) if request.args.get('lat') is not None else None
|
||||
lon = float(request.args.get('lon')) if request.args.get('lon') is not None else None
|
||||
except ValueError:
|
||||
raise ValidationError('limit, lat, and lon must be numbers')
|
||||
data = await asyncio.to_thread(map_search.search_places, q, limit=limit, offset=_offset(request),
|
||||
lat=lat, lon=lon)
|
||||
results = [{k: v for k, v in dict(
|
||||
name=i.get('name'),
|
||||
kind=i.get('kind_detail') or i.get('kind'),
|
||||
region=i.get('region'),
|
||||
population=i.get('population'),
|
||||
lat=i.get('lat'),
|
||||
lon=i.get('lon'),
|
||||
link=_map_link(i.get('lat'), i.get('lon')),
|
||||
).items() if v is not None} for i in data.get('results') or []]
|
||||
return json_response(dict(results=results, total=data.get('total', len(results))))
|
||||
|
||||
|
||||
@ai_bp.post('/search/suggestions')
|
||||
@openapi.definition(
|
||||
summary='Preview what a search would find',
|
||||
description='Before searching, learn which channels, domains (archived sites), authors, and subjects match'
|
||||
' a term, and how many files and Zim entries a search would return. Use the channel id with'
|
||||
' search_videos, the domain name with search_archives, the author/subject with search_docs.'
|
||||
' tag_names narrows the file estimate to tagged items.',
|
||||
body=schema.AISearchSuggestionsRequest,
|
||||
)
|
||||
@openapi.operation('search_suggestions')
|
||||
@validate(schema.AISearchSuggestionsRequest)
|
||||
async def search_suggestions(request: Request, body: schema.AISearchSuggestionsRequest):
|
||||
if not body.search_str and not body.tag_names:
|
||||
raise ValidationError('search_str or tag_names is required')
|
||||
session = request.ctx.session
|
||||
search_str = body.search_str or ''
|
||||
|
||||
channels = await search_channels_by_name(session, search_str, order_by_video_count=True)
|
||||
domains = await search_domains_by_name(session, search_str)
|
||||
authors = await search_authors_by_name(session, search_str)
|
||||
subjects = await search_subjects_by_name(session, search_str)
|
||||
|
||||
counts = await search_file_suggestion_count(body.search_str, body.tag_names, [])
|
||||
zims = []
|
||||
if body.search_str:
|
||||
def estimate():
|
||||
with_counts = Zims.estimate(session, body.search_str)
|
||||
return [dict(id=zim.id, title=lib.zim_metadata_dict(zim.zim_metadata).get('title') or zim.path.name,
|
||||
estimate=count) for zim, count in with_counts.items()]
|
||||
|
||||
try:
|
||||
zims = await asyncio.to_thread(estimate)
|
||||
except Exception as e:
|
||||
logger.debug('AI suggestions could not estimate Zims', exc_info=e)
|
||||
|
||||
return json_response(dict(
|
||||
channels=[dict(id=i.id, name=i.name, link=f'/videos/channel/{i.id}/video') for i in channels],
|
||||
domains=[dict(id=i['id'], name=i['domain']) for i in domains],
|
||||
authors=authors,
|
||||
subjects=subjects,
|
||||
estimates=dict(file_groups=counts['file_groups'], file_groups_deep=counts['file_groups_deep'], zims=zims),
|
||||
))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Manage endpoints — for the AI page's Manage tab, NOT part of the model's tool catalog.
|
||||
# The agent loop excludes everything under /api/ai/manage/ when building tool definitions.
|
||||
|
|
|
|||
|
|
@ -60,6 +60,10 @@ MODES = {
|
|||
'get_inventory',
|
||||
'list_files',
|
||||
'read_file',
|
||||
'list_tags',
|
||||
'search_suggestions',
|
||||
'get_map_overview',
|
||||
'search_places',
|
||||
),
|
||||
),
|
||||
'system': dict(
|
||||
|
|
|
|||
|
|
@ -83,3 +83,9 @@ class AIListFilesResponse:
|
|||
files: list = dataclasses.field(default_factory=list)
|
||||
total: int = 0
|
||||
next_offset: Optional[int] = None
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class AISearchSuggestionsRequest:
|
||||
search_str: Optional[str] = None
|
||||
tag_names: List[str] = dataclasses.field(default_factory=list)
|
||||
|
|
|
|||
|
|
@ -383,3 +383,139 @@ async def test_ai_list_files_refuses_escapes(async_client, test_directory):
|
|||
(test_directory / 'config').mkdir(exist_ok=True)
|
||||
request, response = await async_client.get('/api/ai/files/list?path=config')
|
||||
assert response.status_code == HTTPStatus.BAD_REQUEST
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ai_list_tags(async_client, test_session, tag_factory, video_factory):
|
||||
"""Tags are listed with what they are applied to, plus the recently used names."""
|
||||
one = await tag_factory('one')
|
||||
await tag_factory('two')
|
||||
video = video_factory(title='tagged video')
|
||||
video.file_group.add_tag(test_session, one.id)
|
||||
test_session.commit()
|
||||
|
||||
request, response = await async_client.get('/api/ai/tags')
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.json['total'] == 2
|
||||
assert [i['name'] for i in response.json['results']] == ['one', 'two']
|
||||
assert response.json['results'][0]['file_groups'] == 1
|
||||
assert response.json['results'][1]['file_groups'] == 0
|
||||
assert response.json['recent'] == ['one']
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ai_list_downloads(async_client, test_session, test_download_manager, test_downloader,
|
||||
test_directory):
|
||||
"""The download queue is summarized with lean, relative-path entries; status can filter."""
|
||||
test_download_manager.create_download(test_session, 'https://example.com/once', test_downloader.name,
|
||||
destination=test_directory / 'archive/example.com')
|
||||
test_download_manager.recurring_download(test_session, 'https://example.com/feed', 86400,
|
||||
test_downloader.name)
|
||||
test_session.commit()
|
||||
|
||||
request, response = await async_client.get('/api/ai/downloads')
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.json['summary']['recurring'] == 1
|
||||
assert 'disabled' in response.json['summary']
|
||||
once, = response.json['once']
|
||||
assert once['url'] == 'https://example.com/once'
|
||||
assert once['destination'] == 'archive/example.com'
|
||||
assert once['status'] == 'new'
|
||||
assert once['downloader'] == test_downloader.name
|
||||
recurring, = response.json['recurring']
|
||||
assert recurring['url'] == 'https://example.com/feed'
|
||||
assert recurring['frequency'] == 86400
|
||||
|
||||
# Filter by status.
|
||||
request, response = await async_client.get('/api/ai/downloads?status=complete')
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.json['once'] == [] and response.json['recurring'] == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ai_map(async_client, test_session, test_directory, make_files_structure):
|
||||
"""The map overview reports map files, subscriptions, pins, and search indexes; places can be searched."""
|
||||
from modules.map.pins import get_map_pins_config
|
||||
from modules.map.test.test_search import _create_test_search_db
|
||||
|
||||
make_files_structure(['map/oregon.pmtiles'])
|
||||
_create_test_search_db(test_directory / 'map/oregon.search.db', [
|
||||
('Portland', 'city', 45.5, -122.6, 6, 'places', 'city', 650000, 'Oregon'),
|
||||
('Port Orford', 'town', 42.7, -124.5, 9, 'places', 'town', 1100, 'Oregon'),
|
||||
])
|
||||
get_map_pins_config().add_pin(45.1, -122.2, 'Home')
|
||||
|
||||
request, response = await async_client.get('/api/ai/map')
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
files, = response.json['files']
|
||||
assert files['name'] == 'oregon.pmtiles' and files['has_search_index'] is True
|
||||
assert response.json['subscriptions'] == []
|
||||
pin, = response.json['pins']
|
||||
assert pin['label'] == 'Home' and pin['link'] == '/map?lat=45.1&lon=-122.2&z=12'
|
||||
|
||||
request, response = await async_client.get('/api/ai/map/search?q=port')
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.json['total'] == 2
|
||||
first = response.json['results'][0]
|
||||
assert first['name'] == 'Portland'
|
||||
assert first['region'] == 'Oregon'
|
||||
assert first['link'] == '/map?lat=45.5&lon=-122.6&z=12'
|
||||
|
||||
# Proximity ranking within an importance tier: equal min_zoom, nearest first.
|
||||
_create_test_search_db(test_directory / 'map/coast.search.db', [
|
||||
('Port A', 'town', 42.0, -124.0, 9, 'places', 'town', 100, 'Oregon'),
|
||||
('Port B', 'town', 44.0, -124.0, 9, 'places', 'town', 100, 'Oregon'),
|
||||
])
|
||||
request, response = await async_client.get('/api/ai/map/search?q=port%20b&lat=44.0&lon=-124.0')
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.json['results'][0]['name'] == 'Port B'
|
||||
request, response = await async_client.get('/api/ai/map/search?q=port&lat=42.0&lon=-124.0&limit=2')
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert len(response.json['results']) == 2 and response.json['total'] == 4
|
||||
|
||||
# q is required.
|
||||
request, response = await async_client.get('/api/ai/map/search')
|
||||
assert response.status_code == HTTPStatus.BAD_REQUEST
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ai_search_suggestions(async_client, test_session, video_factory, archive_factory, channel_factory,
|
||||
tag_factory):
|
||||
"""Suggestions name the channels/domains/authors/subjects matching a term and estimate result counts."""
|
||||
channel = channel_factory(name='Wild Cooking')
|
||||
video_factory(channel_id=channel.id, title='cooking rice')
|
||||
archive_factory(domain='cooking.example.com', title='cooking beans', contents='beans')
|
||||
tag = await tag_factory('recipes')
|
||||
test_session.commit()
|
||||
|
||||
content = dict(search_str='cooking')
|
||||
request, response = await async_client.post('/api/ai/search/suggestions', content=json.dumps(content))
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
ch, = response.json['channels']
|
||||
assert ch['name'] == 'Wild Cooking' and ch['id'] == channel.id and ch['link'].startswith('/videos/channel/')
|
||||
dom, = response.json['domains']
|
||||
assert dom['name'] == 'cooking.example.com'
|
||||
assert response.json['authors'] == [] and response.json['subjects'] == []
|
||||
assert response.json['estimates']['file_groups'] == 2
|
||||
assert 'file_groups_deep' in response.json['estimates']
|
||||
assert response.json['estimates']['zims'] == []
|
||||
|
||||
# Tag names narrow the estimate.
|
||||
content = dict(search_str='cooking', tag_names=[tag.name])
|
||||
request, response = await async_client.post('/api/ai/search/suggestions', content=json.dumps(content))
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.json['estimates']['file_groups'] == 0
|
||||
|
||||
# Something to search for is required.
|
||||
request, response = await async_client.post('/api/ai/search/suggestions', content=json.dumps({}))
|
||||
assert response.status_code == HTTPStatus.BAD_REQUEST
|
||||
|
||||
|
||||
def test_ai_format_download_error_tail():
|
||||
"""Download errors are tracebacks; the model gets the end (the exception message), not the start."""
|
||||
from modules.ai.api import _format_download, DOWNLOAD_ERROR_LENGTH
|
||||
error = 'Traceback (most recent call last):\n' + (' File "x.py"\n' * 100) + 'ValueError: the real reason'
|
||||
formatted = _format_download(dict(id=1, url='https://example.com', status='failed', error=error))
|
||||
assert formatted['error'].endswith('ValueError: the real reason')
|
||||
assert len(formatted['error']) == DOWNLOAD_ERROR_LENGTH + 1
|
||||
assert 'error' not in _format_download(dict(id=2, url='https://example.com', status='new', error=None))
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ EXPECTED_TOOLS = {
|
|||
'get_doc', 'get_zim_entry',
|
||||
'list_collections', 'list_zim_files', 'list_files',
|
||||
'get_statistics', 'get_inventory', 'get_inventory_items', 'get_status',
|
||||
'list_tags', 'search_suggestions', 'list_downloads', 'get_map_overview', 'search_places',
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -498,6 +498,164 @@ async def list_zim_files() -> str:
|
|||
return "\n\n".join(lines)
|
||||
|
||||
|
||||
@mcp.tool(annotations={"readOnlyHint": True})
|
||||
async def list_tags() -> str:
|
||||
"""List every tag in the library with how many files, Zim entries, channels, and domains carry it.
|
||||
|
||||
Tag names can be passed as tag_names to the search tools. Users tag what they care about, so this
|
||||
is a good map of their interests. Also returns the most recently used tag names.
|
||||
"""
|
||||
data = await api_get("/api/ai/tags")
|
||||
results = data.get("results") or []
|
||||
if not results:
|
||||
return "No tags found."
|
||||
lines = []
|
||||
for tag in results:
|
||||
counts = ", ".join(f"{tag.get(k, 0)} {k.replace('_', ' ')}" for k in
|
||||
("file_groups", "zim_entries", "channels", "domains") if tag.get(k))
|
||||
lines.append(f" {tag['name']}" + (f" ({counts})" if counts else " (unused)"))
|
||||
text = f"Tags ({data.get('total', len(results))}):\n" + "\n".join(lines)
|
||||
if recent := data.get("recent"):
|
||||
text += f"\n\nRecently used: {', '.join(recent)}"
|
||||
return text
|
||||
|
||||
|
||||
@mcp.tool(annotations={"readOnlyHint": True})
|
||||
async def search_suggestions(query: str, tag_names: list[str] | None = None) -> str:
|
||||
"""Preview what a search would find before running it.
|
||||
|
||||
Returns the channels, domains (archived sites), authors, and subjects whose names match the query,
|
||||
and estimates of how many files and Zim entries the search would return. Use the channel ID with
|
||||
search_videos, the domain with search_archives, the author/subject with search_docs.
|
||||
|
||||
Args:
|
||||
query: The term the user wants to search for.
|
||||
tag_names: Narrow the file estimate to items with these tags.
|
||||
"""
|
||||
body = {"search_str": query}
|
||||
if tag_names:
|
||||
body["tag_names"] = tag_names
|
||||
data = await api_post("/api/ai/search/suggestions", json=body)
|
||||
lines = []
|
||||
for channel in data.get("channels") or []:
|
||||
lines.append(f" Channel: {channel['name']} (ID: {channel['id']}) LINK: {_link(channel.get('link'))}")
|
||||
for domain in data.get("domains") or []:
|
||||
lines.append(f" Domain: {domain['name']}")
|
||||
for author in data.get("authors") or []:
|
||||
lines.append(f" Author: {author['name']}")
|
||||
for subject in data.get("subjects") or []:
|
||||
lines.append(f" Subject: {subject['name']}")
|
||||
if not lines:
|
||||
lines.append(" No matching channels, domains, authors, or subjects.")
|
||||
estimates = data.get("estimates") or {}
|
||||
lines.append(f"\nEstimated matching files: {estimates.get('file_groups', 0)}"
|
||||
f" (deep content search: {estimates.get('file_groups_deep', 0)})")
|
||||
for zim in estimates.get("zims") or []:
|
||||
lines.append(f" Zim '{zim.get('title')}' (ID {zim.get('id')}): ~{zim.get('estimate', 0)} entries")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
@mcp.tool(annotations={"readOnlyHint": True})
|
||||
async def list_downloads(status: str | None = None, limit: int = DEFAULT_LIMIT) -> str:
|
||||
"""Read the WROLPi download queue: summary, recurring downloads (channels, feeds), and one-time downloads.
|
||||
|
||||
Use this when the user asks what is downloading, what failed and why, or what is scheduled.
|
||||
Downloads cannot be started, stopped, or retried from here.
|
||||
|
||||
Args:
|
||||
status: Only show downloads with this status: new, pending, failed, deferred, or complete.
|
||||
limit: Maximum recurring and one-time downloads to show (each).
|
||||
"""
|
||||
params = {"limit": limit}
|
||||
if status:
|
||||
params["status"] = status
|
||||
data = await api_get("/api/ai/downloads", params=params)
|
||||
summary = data.get("summary") or {}
|
||||
lines = ["Summary: " + ", ".join(f"{k}={v}" for k, v in summary.items())]
|
||||
|
||||
def render(download: dict) -> str:
|
||||
parts = [f" [{download.get('status')}] {download.get('url')} (ID: {download.get('id')},"
|
||||
f" {download.get('downloader')})"]
|
||||
if download.get("frequency"):
|
||||
parts.append(f" every {download['frequency']}s, next: {download.get('next_download') or 'unscheduled'}")
|
||||
if download.get("destination"):
|
||||
parts.append(f" destination: {download['destination']}")
|
||||
if download.get("tag_names"):
|
||||
parts.append(f" tags: {', '.join(download['tag_names'])}")
|
||||
if download.get("error"):
|
||||
parts.append(f" error: {download['error']}")
|
||||
return "\n".join(parts)
|
||||
|
||||
recurring = data.get("recurring") or []
|
||||
once = data.get("once") or []
|
||||
lines.append(f"\nRecurring downloads ({len(recurring)} shown):")
|
||||
lines.extend(render(i) for i in recurring) if recurring else lines.append(" none")
|
||||
lines.append(f"\nOne-time downloads ({len(once)} shown, {data.get('pending_once', 0)} pending):")
|
||||
lines.extend(render(i) for i in once) if once else lines.append(" none")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
@mcp.tool(annotations={"readOnlyHint": True})
|
||||
async def get_map_overview() -> str:
|
||||
"""Describe the maps on this WROLPi: downloaded map regions, whether each has a place-search index,
|
||||
subscribed regions, and the user's saved pins (with links)."""
|
||||
data = await api_get("/api/ai/map")
|
||||
lines = ["Map files:"]
|
||||
files = data.get("files") or []
|
||||
for file in files:
|
||||
size_gb = (file.get("size") or 0) / (1024 ** 3)
|
||||
index = "searchable" if file.get("has_search_index") else "no search index"
|
||||
lines.append(f" {file['name']} ({size_gb:,.2f} GB, {index})")
|
||||
if not files:
|
||||
lines.append(" none")
|
||||
subscriptions = data.get("subscriptions") or []
|
||||
names = [i.get("name") or i.get("region") if isinstance(i, dict) else str(i) for i in subscriptions]
|
||||
lines.append(f"\nSubscribed regions: {', '.join(names) if names else 'none'}")
|
||||
lines.append("\nPins:")
|
||||
pins = data.get("pins") or []
|
||||
for pin in pins:
|
||||
lines.append(f" {pin.get('label') or 'Unlabeled'} ({pin.get('lat')}, {pin.get('lon')})"
|
||||
f" LINK: {_link(pin.get('link'))}")
|
||||
if not pins:
|
||||
lines.append(" none")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
@mcp.tool(annotations={"readOnlyHint": True})
|
||||
async def search_places(
|
||||
query: str,
|
||||
limit: int = DEFAULT_LIMIT,
|
||||
offset: int = 0,
|
||||
lat: float | None = None,
|
||||
lon: float | None = None,
|
||||
) -> str:
|
||||
"""Find towns, cities, and landmarks by name in the downloaded maps. Each result links to the WROLPi map.
|
||||
|
||||
Args:
|
||||
query: Place name (prefix match), e.g. "Portland".
|
||||
limit: Maximum results.
|
||||
offset: Pagination offset.
|
||||
lat: Latitude to rank nearby places first.
|
||||
lon: Longitude to rank nearby places first.
|
||||
"""
|
||||
params = {"q": query, "limit": limit, "offset": offset}
|
||||
if lat is not None and lon is not None:
|
||||
params.update(lat=lat, lon=lon)
|
||||
data = await api_get("/api/ai/map/search", params=params)
|
||||
results = data.get("results") or []
|
||||
if not results:
|
||||
return "No places found. The maps may have no search index (see get_map_overview)."
|
||||
lines = []
|
||||
for i, place in enumerate(results, 1):
|
||||
detail = ", ".join(str(place[k]) for k in ("kind", "region") if place.get(k))
|
||||
if place.get("population"):
|
||||
detail += f", pop. {place['population']:,}"
|
||||
lines.append(f"{i}. {place['name']} ({detail}) at {place.get('lat')}, {place.get('lon')}"
|
||||
f" LINK: {_link(place.get('link'))}")
|
||||
lines.append(f"\nTotal matching: {data.get('total', len(results))}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
@mcp.tool(annotations={"readOnlyHint": True})
|
||||
async def get_statistics() -> str:
|
||||
"""Get an overview of what content is stored in the WROLPi library.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue