"""Vault listing and directory browsing services. Single source of truth consumed by both the REST routes (``/api/vaults``, ``/api/browse/{vault}``) and the AI tool layer (``list_vaults``, ``list_directory``). """ from __future__ import annotations from pathlib import Path from typing import Any from backend.services.errors import ServiceError from backend.services.paths import resolve_safe_path def list_accessible_vaults(user: dict[str, Any]) -> list[dict[str, Any]]: """Return the vaults *user* may access, with summary metadata.""" from backend.auth.middleware import check_vault_access from backend.indexer import index result: list[dict[str, Any]] = [] for name, data in index.items(): if not check_vault_access(name, user): continue result.append({ "name": name, "file_count": len(data.get("files", [])), "tag_count": len(data.get("tags", {})), "type": data.get("config", {}).get("type", "VAULT"), }) return result def get_vault_root(vault_name: str) -> Path: """Return the filesystem root of *vault_name* or raise ``not_found``.""" from backend.indexer import get_vault_data data = get_vault_data(vault_name) if not data: raise ServiceError( f"Vault '{vault_name}' not found", code="not_found", status=404, details={"vault": vault_name}, ) return Path(data["path"]) def browse_directory(vault_name: str, path: str = "") -> dict[str, Any]: """Return the direct children of a vault directory (directories first).""" from backend.indexer import SUPPORTED_EXTENSIONS from backend.vault_settings import get_vault_setting root = get_vault_root(vault_name) target = resolve_safe_path(root, path) if path else root.resolve() if not target.exists(): raise ServiceError( f"Path not found: {path}", code="not_found", status=404, details={"vault": vault_name, "path": path}, ) hide_hidden = (get_vault_setting(vault_name) or {}).get("hideHiddenFiles", False) items: list[dict[str, Any]] = [] try: for entry in sorted(target.iterdir(), key=lambda e: (not e.is_dir(), e.name.lower())): if hide_hidden and entry.name.startswith("."): continue rel = str(entry.relative_to(root)).replace("\\", "/") if entry.is_dir(): # Count only direct children (files and subdirs) for performance. try: file_count = sum( 1 for child in entry.iterdir() if (not hide_hidden or not child.name.startswith(".")) and (child.is_file() and (child.suffix.lower() in SUPPORTED_EXTENSIONS or child.name.lower() in ("dockerfile", "makefile")) or child.is_dir()) ) except PermissionError: file_count = 0 items.append({ "name": entry.name, "path": rel, "type": "directory", "children_count": file_count, }) elif entry.suffix.lower() in SUPPORTED_EXTENSIONS or entry.name.lower() in ("dockerfile", "makefile"): items.append({ "name": entry.name, "path": rel, "type": "file", "size": entry.stat().st_size, "extension": entry.suffix.lower(), }) except PermissionError: raise ServiceError("Permission denied", code="permission_denied", status=403) from None return {"vault": vault_name, "path": path, "items": items} def _indexed_tags(vault_name: str, rel_path: str) -> list[str]: """Tags of a file as stored in the search index (#158). Empty list when the file is not indexed yet (binary formats, index still building) — the listing itself comes from the filesystem, tags are a decoration used by the navigation page facets/filters. Args: vault_name: Vault name. rel_path: Path of the file relative to the vault root (``/`` separated). Returns: The file's tags, or an empty list when unknown. """ try: from backend.search import get_inverted_index info = get_inverted_index().doc_info.get(f"{vault_name}::{rel_path}") except Exception: return [] return list((info or {}).get("tags") or []) def list_all_files( vault_name: str, dir: str = "", limit: int = 200, recursive: bool = True, ) -> dict[str, Any]: """List files in a vault directory sorted by modification time (newest first). Unlike :func:`browse_directory`, this returns files only (with metadata: size, mtime, extension) and can recurse into subdirectories. Hidden files and ignored directories are skipped. Args: vault_name: Name of the vault. dir: Relative directory path within the vault (empty = root). limit: Maximum number of files to return. recursive: Recurse into subdirectories when True. Returns: Dict with ``vault``, ``directory``, ``recursive``, ``count`` and ``files``. Raises: ServiceError: ``not_found`` (404) if the directory does not exist, ``permission_denied`` (403) if it cannot be read. """ from datetime import datetime, timezone from backend.indexer import IGNORED_DIRS root = get_vault_root(vault_name) dir_path = resolve_safe_path(root, dir) if dir else root if not dir_path.exists() or not dir_path.is_dir(): raise ServiceError( f"Directory not found: {dir}", code="not_found", status=404, details={"vault": vault_name, "path": dir}, ) files: list[dict[str, Any]] = [] dir_prefix = (dir or "").strip("/") try: iterator = dir_path.rglob("*") if recursive else dir_path.iterdir() for entry in iterator: if not entry.is_file(): continue if entry.name.startswith("."): continue if entry.name in IGNORED_DIRS: continue if recursive and dir_prefix: skip = False try: for part in entry.relative_to(dir_path).parts[:-1]: if part.startswith(".") or part in IGNORED_DIRS: skip = True break except ValueError: pass if skip: continue stat = entry.stat() rel_path = str(entry.relative_to(root)).replace("\\", "/") if recursive and dir_prefix: try: rel_to_dir = str(entry.parent.relative_to(dir_path)).replace("\\", "/") except ValueError: rel_to_dir = "" else: rel_to_dir = "" ext = entry.suffix.lower() if entry.suffix else "" file_entry = { "name": entry.name, "path": rel_path, "vault": vault_name, "size": stat.st_size, "modified": stat.st_mtime, "modified_iso": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(), "extension": ext.lstrip(".") if ext else "", "tags": _indexed_tags(vault_name, rel_path), } if rel_to_dir and rel_to_dir != ".": file_entry["rel_dir"] = rel_to_dir files.append(file_entry) except PermissionError: raise ServiceError( "Permission denied reading directory", code="permission_denied", status=403, details={"vault": vault_name, "path": dir}, ) from None files.sort(key=lambda f: f["modified"], reverse=True) files = files[:limit] return { "vault": vault_name, "directory": dir, "recursive": recursive, "count": len(files), "files": files, }