ObsiGate/backend/services/recent.py

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