249 lines
8.1 KiB
Python
249 lines
8.1 KiB
Python
"""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,
|
|
username: str | 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``.
|
|
|
|
``username`` (#196) : les documents **reçus** par cet utilisateur via un
|
|
partage dirigé sont ajoutés aux résultats (lecture seule, sans indexer
|
|
les fichiers d'autrui — le contenu est lu à la volée et mis en cache
|
|
TF-IDF côté index, pas sur disque).
|
|
"""
|
|
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", ""))]
|
|
|
|
if username is not None:
|
|
all_results = all_results + _shared_results(username, q, 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 _shared_results(username: str, q: str, vault: str) -> list[dict[str, Any]]:
|
|
"""Search results for documents shared TO *username* (#196).
|
|
|
|
Reads the shared file content on the fly (read-only, source vault on
|
|
disk) and returns hits shaped exactly like ordinary search results so
|
|
the frontend can open them via the normal share page.
|
|
"""
|
|
from pathlib import Path
|
|
|
|
from backend.indexer import get_vault_data
|
|
from backend.services.paths import resolve_safe_path
|
|
from backend.share import list_shares
|
|
|
|
out: list[dict[str, Any]] = []
|
|
if not q:
|
|
return out
|
|
q_lower = q.lower()
|
|
|
|
for s in list_shares(user=username):
|
|
if s.get("created_by") == username:
|
|
continue # own share — already indexed in its own vault
|
|
if vault not in ("all", f"home-{username}"):
|
|
continue
|
|
data = get_vault_data(s["vault"])
|
|
if not data:
|
|
continue
|
|
try:
|
|
fp = resolve_safe_path(Path(data["path"]), s["path"])
|
|
if not fp.exists() or fp.suffix.lower() != ".md":
|
|
continue
|
|
raw = fp.read_text(encoding="utf-8", errors="replace")
|
|
except Exception: # nosec B112 — partage illisible : ignoré, on continue
|
|
continue
|
|
title = fp.stem
|
|
occurrences = raw.lower().count(q_lower)
|
|
if q_lower in title.lower():
|
|
occurrences += 1
|
|
if occurrences == 0:
|
|
continue
|
|
out.append({
|
|
"vault": f"home-{username}",
|
|
"path": f"Partage/{fp.name}",
|
|
"title": title,
|
|
"tags": [],
|
|
"score": min(occurrences, 10),
|
|
"snippet": _snippet(raw, q_lower),
|
|
"modified": None,
|
|
"share_token": s["token"],
|
|
})
|
|
return out
|
|
|
|
|
|
def _snippet(text: str, q_lower: str, width: int = 160) -> str:
|
|
"""Small excerpt around the first match (#196, shared-file search)."""
|
|
idx = text.lower().find(q_lower)
|
|
if idx < 0:
|
|
return text[:width]
|
|
start = max(0, idx - width // 2)
|
|
return "…" + text[start:start + width].replace("\n", " ") + "…"
|
|
|
|
|
|
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.
|
|
|
|
#197 (synchro TikTak) : chaque entrée fichier expose aussi ``size``,
|
|
``mtime`` (epoch) et ``mtime_iso`` — un client externe peut détecter les
|
|
fichiers modifiés depuis un horodatage en **un seul appel**, sans relire
|
|
chaque fichier. Les valeurs viennent de l'index (source unique, même
|
|
contrat que ``list_all_files``) ; les répertoires portent ``None``.
|
|
|
|
Args:
|
|
vault: Vault name.
|
|
limit: Maximum number of entries returned.
|
|
|
|
Returns:
|
|
``{"vault", "count", "results": [{vault, path, name, type, size?,
|
|
mtime?, mtime_iso?}]}``.
|
|
"""
|
|
from datetime import datetime
|
|
|
|
from backend.indexer import get_vault_data, path_index
|
|
|
|
entries = path_index.get(vault, [])
|
|
meta: dict[str, dict[str, Any]] = {}
|
|
data = get_vault_data(vault)
|
|
if data:
|
|
for info in data.get("files", []):
|
|
meta[info["path"]] = info
|
|
|
|
results = []
|
|
for entry in entries[: max(0, limit)]:
|
|
item: dict[str, Any] = {
|
|
"vault": vault,
|
|
"path": entry["path"],
|
|
"name": entry["name"],
|
|
"type": entry["type"],
|
|
"size": None,
|
|
"mtime": None,
|
|
"mtime_iso": None,
|
|
}
|
|
info = meta.get(entry["path"])
|
|
if info is not None:
|
|
item["size"] = info.get("size")
|
|
iso = info.get("modified")
|
|
item["mtime_iso"] = iso
|
|
if iso:
|
|
try:
|
|
item["mtime"] = datetime.fromisoformat(iso).timestamp()
|
|
except (ValueError, TypeError):
|
|
item["mtime"] = None
|
|
results.append(item)
|
|
return {"vault": vault, "count": len(results), "results": results}
|