"""Search services shared by REST routes and the AI tool layer.""" from __future__ import annotations from collections.abc import Callable from typing import Any def search_vaults( q: str, vault: str = "all", tag: str | None = None, limit: int = 50, offset: int = 0, is_allowed: Callable[[str], bool] | None = None, ) -> dict[str, Any]: """Full-text search with pagination, returned as the API response payload. ``is_allowed`` (#194) filters the raw hits **before** pagination, so a restricted vault set neither distorts ``total`` nor the returned page. The tool layer filters on its own and leaves it as ``None``. """ from backend.search import search all_results = search(q, vault_filter=vault, tag_filter=tag) if is_allowed is not None: all_results = [r for r in all_results if is_allowed(r.get("vault", ""))] total = len(all_results) page = all_results[offset: offset + limit] return { "query": q, "vault_filter": vault, "tag_filter": tag, "count": len(page), "total": total, "offset": offset, "limit": limit, "results": page, } def list_tags(vault: str | None = None) -> dict[str, int]: """Return tag → count, optionally restricted to a single vault.""" from backend.search import get_all_tags return get_all_tags(vault_filter=vault) def advanced_search_vaults( query: str = "", vault: str = "all", tag: str | None = None, limit: int = 50, offset: int = 0, sort: str = "relevance", case_sensitive: bool = False, whole_word: bool = False, regex: bool = False, include_paths: str | None = None, exclude_paths: str | None = None, created: str | None = None, modified: str | None = None, size: str | None = None, semantic: bool = False, ) -> dict[str, Any]: """Advanced full-text search (TF-IDF, facets, operators). When ``semantic`` is True, the TF-IDF ranking is fused with the embedding (semantic) ranking via RRF. No permission filtering is applied: callers that need it (the tool layer) filter the ``results`` list themselves. """ from backend.search import advanced_search return advanced_search( query, vault_filter=vault, tag_filter=tag, limit=limit, offset=offset, sort_by=sort, case_sensitive=case_sensitive, whole_word=whole_word, regex=regex, include_paths=include_paths, exclude_paths=exclude_paths, created=created, modified=modified, size=size, semantic=semantic, ) def search_paths(q: str, vault: str = "all") -> dict[str, Any]: """Search files and directories by path substring using the path index. No permission filtering is applied: callers that need it (the tool layer) filter the ``results`` list themselves. """ from backend.indexer import path_index if not q: return {"query": q, "vault_filter": vault, "results": []} query_lower = q.lower() results: list[dict[str, Any]] = [] vaults_to_search = [vault] if vault != "all" else list(path_index.keys()) for vault_name in vaults_to_search: for entry in path_index.get(vault_name, []): if query_lower in entry["name"].lower() or query_lower in entry["path"].lower(): results.append({ "vault": vault_name, "path": entry["path"], "name": entry["name"], "type": entry["type"], "matched_path": entry["path"], }) return {"query": q, "vault_filter": vault, "results": results} def list_paths(vault: str, limit: int = 5000) -> dict[str, Any]: """Return a flat, capped list of every indexed path in a vault. Backs the AI assistant ``@`` mention menu: fetching the whole path index once lets the client filter files/directories instantly instead of issuing a request per keystroke. Args: vault: Vault name. limit: Maximum number of entries returned. Returns: ``{"vault", "count", "results": [{vault, path, name, type}]}``. """ from backend.indexer import path_index entries = path_index.get(vault, []) results = [ { "vault": vault, "path": entry["path"], "name": entry["name"], "type": entry["type"], } for entry in entries[: max(0, limit)] ] return {"vault": vault, "count": len(results), "results": results}