116 lines
3.4 KiB
Python
116 lines
3.4 KiB
Python
"""Search services shared by REST routes and the AI tool layer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
def search_vaults(
|
|
q: str,
|
|
vault: str = "all",
|
|
tag: str | None = None,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> dict[str, Any]:
|
|
"""Full-text search with pagination, returned as the API response payload.
|
|
|
|
No permission filtering is applied here: callers that need it (the tool
|
|
layer) filter the ``results`` list themselves.
|
|
"""
|
|
from backend.search import search
|
|
|
|
all_results = search(q, vault_filter=vault, tag_filter=tag)
|
|
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}
|