"""Recent-files services shared by the REST route and the AI tool layer. Provides ``list_recent`` (last opened files, falling back to last modified) and ``humanize_mtime``. The route ``/api/recent`` and the tool ``list_recent`` both delegate here. Permission filtering is applied against the caller's vault list. """ from __future__ import annotations import time from datetime import datetime from typing import Any from backend.auth.middleware import is_home_vault from backend.history import get_recent_opened, is_bookmarked from backend.indexer import find_file_in_index, index def humanize_mtime(mtime: float) -> str: """Return a short, human-friendly French rendering of a timestamp.""" delta = time.time() - mtime if delta < 60: return "à l'instant" if delta < 3600: return f"il y a {int(delta / 60)} min" if delta < 86400: return f"il y a {int(delta / 3600)} h" if delta < 604800: return f"il y a {int(delta / 86400)} j" return datetime.fromtimestamp(mtime).strftime("%d %b %Y") def _can_access(vault: str, user_vaults: list[str]) -> bool: # #194 : un dossier perso ne bénéficie jamais de "*" (même règle que # backend.auth.middleware.check_vault_access). if is_home_vault(vault): return vault in user_vaults return "*" in user_vaults or vault in user_vaults def list_recent( username: str | None, user_vaults: list[str] | None = None, *, vault: str | None = None, limit: int = 20, mode: str = "opened", ) -> dict[str, Any]: """Return the caller's recent files. Args: username: Caller username (needed for the "opened" history mode). user_vaults: Vaults the caller may access (``["*"]`` = all). vault: Optional single-vault filter. limit: Maximum number of files to return. mode: ``"opened"`` to use the open history, anything else for the last-modified fallback. Returns: Dict with ``files``, ``total``, ``limit`` and ``mode``. """ user_vaults = user_vaults or [] if mode == "opened" and username: history = get_recent_opened(username, vault_filter=vault, limit=limit) files_resp: list[dict[str, Any]] = [] for item in history: v_name = item["vault"] if not _can_access(v_name, user_vaults): continue f_idx = find_file_in_index(item["path"], v_name) if f_idx: files_resp.append({ "path": f_idx["path"], "title": f_idx.get("title") or item["path"].split("/")[-1], "vault": v_name, "mtime": item["opened_at"], "mtime_human": humanize_mtime(item["opened_at"]), "size_bytes": f_idx.get("size", 0), "tags": [f"#{t}" for t in f_idx.get("tags", [])][:5], "preview": f_idx.get("content_preview", "")[:120], "bookmarked": is_bookmarked(username, v_name, f_idx["path"]), }) else: files_resp.append({ "path": item["path"], "title": item.get("title") or item["path"].split("/")[-1], "vault": v_name, "mtime": item["opened_at"], "mtime_human": humanize_mtime(item["opened_at"]), "tags": [], "preview": "", "bookmarked": is_bookmarked(username, v_name, item["path"]), }) return { "files": files_resp, "total": len(files_resp), "limit": limit, "mode": "opened", } all_files: list[tuple[str, dict[str, Any]]] = [] for v_name, v_data in index.items(): if vault and v_name != vault: continue if not _can_access(v_name, user_vaults): continue for f in v_data.get("files", []): all_files.append((v_name, f)) all_files.sort(key=lambda x: x[1].get("modified", ""), reverse=True) recent = all_files[:limit] files_resp = [] for v_name, f in recent: iso_modified = f.get("modified", "") try: mtime_dt = datetime.fromisoformat(iso_modified.replace("Z", "+00:00")) mtime_val = mtime_dt.timestamp() except Exception: mtime_val = time.time() files_resp.append({ "path": f["path"], "title": f["title"], "vault": v_name, "mtime": mtime_val, "mtime_human": humanize_mtime(mtime_val), "mtime_iso": iso_modified, "size_bytes": f.get("size", 0), "tags": [f"#{t}" for t in f.get("tags", [])][:5], "preview": f.get("content_preview", "")[:120], "bookmarked": is_bookmarked(username, v_name, f["path"]) if username else False, }) return { "files": files_resp, "total": len(all_files), "limit": limit, "mode": "modified", }