ObsiGate/backend/services/search.py

"""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}