Files
ObsiGate/backend/services/search.py
T
bruno 9274dbd49f
CI / lint (push) Successful in 1m11s
CI / security (push) Successful in 47s
CI / test (push) Successful in 2m12s
CI / build (push) Successful in 45s
CI / e2e (push) Successful in 10m21s
feat(ai): menu @ instantane, selecteurs agrandis et BooksLM racine (#82)
Nouvel endpoint GET /api/vault/{vault}/paths + prechargement et filtrage client du menu @ (affichage instantane des fichiers/repertoires). Selecteurs Fournisseur/Modele agrandis (0,8rem / 34px) pour s'aligner sur le reste du site. Ajout de BooksLM au menu contextuel de la racine des vaults.
2026-09-12 20:09:08 -04:00

145 lines
4.2 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}
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}