315 lines
11 KiB
Python
315 lines
11 KiB
Python
"""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
|
|
|
|
import fnmatch
|
|
from datetime import UTC
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from backend.services.errors import ServiceError
|
|
from backend.services.paths import resolve_safe_path
|
|
|
|
|
|
def chemin_exclu_dans_config(config: dict[str, Any] | None, rel_path: str) -> bool:
|
|
"""True si *rel_path* est exclu par la config d'exclusions d'un vault.
|
|
|
|
Deux mécanismes, lus dans le dict ``config`` du vault (alimenté par
|
|
``DIR_N_IGNORED_DIRS`` / ``DIR_N_IGNORED_PATHS``, vaults DIR seulement) :
|
|
|
|
- ``ignoredDirs`` : le nom d'un segment du chemin (à toute profondeur)
|
|
figure dans l'ensemble — ex. ``.home`` cache ``a/.home/x`` ;
|
|
- ``ignoredPaths`` : motifs fnmatch testés contre le chemin relatif
|
|
POSIX **et chacun de ses préfixes** — un dossier qui matche cache
|
|
donc tout son sous-arbre, ex. ``Executeurs/*/data`` cache aussi
|
|
``Executeurs/Hermes/data/config.md``.
|
|
|
|
Tolérant : config absente/vide ou chemin vide → False.
|
|
"""
|
|
if not config or not rel_path:
|
|
return False
|
|
rel = str(rel_path).replace("\\", "/").strip("/")
|
|
if not rel:
|
|
return False
|
|
segments = rel.split("/")
|
|
ignored_dirs = config.get("ignoredDirs") or []
|
|
if any(seg in ignored_dirs for seg in segments):
|
|
return True
|
|
prefixes = ["/".join(segments[:i]) for i in range(1, len(segments) + 1)]
|
|
for pattern in config.get("ignoredPaths") or []:
|
|
pat = str(pattern).replace("\\", "/").strip("/")
|
|
if pat and any(fnmatch.fnmatchcase(prefix, pat) for prefix in prefixes):
|
|
return True
|
|
return False
|
|
|
|
|
|
def chemin_exclu(vault_name: str, rel_path: str) -> bool:
|
|
"""True si *rel_path* est exclu dans le vault *vault_name*.
|
|
|
|
Lit la config du vault via l'index (``get_vault_data``). Tolérant :
|
|
vault inconnu ou sans config d'exclusions → False.
|
|
"""
|
|
try:
|
|
from backend.indexer import get_vault_data
|
|
|
|
data = get_vault_data(vault_name)
|
|
except Exception:
|
|
return False
|
|
if not data:
|
|
return False
|
|
return chemin_exclu_dans_config(data.get("config") or {}, rel_path)
|
|
|
|
|
|
def vault_readonly(vault_name: str) -> bool:
|
|
"""True si le vault est déclaré en lecture seule (DIR_N_READONLY)."""
|
|
try:
|
|
from backend.indexer import get_vault_data
|
|
|
|
data = get_vault_data(vault_name)
|
|
except Exception:
|
|
return False
|
|
if not data:
|
|
return False
|
|
return bool((data.get("config") or {}).get("readonly", False))
|
|
|
|
|
|
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"),
|
|
"readonly": bool(data.get("config", {}).get("readonly", False)),
|
|
})
|
|
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 datetime import datetime
|
|
|
|
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() or (path and chemin_exclu(vault_name, path)):
|
|
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 chemin_exclu(vault_name, rel):
|
|
continue
|
|
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 not chemin_exclu(vault_name, f"{rel}/{child.name}")
|
|
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"):
|
|
stat = entry.stat()
|
|
items.append({
|
|
"name": entry.name,
|
|
"path": rel,
|
|
"type": "file",
|
|
"size": stat.st_size,
|
|
"modified": stat.st_mtime,
|
|
"modified_iso": datetime.fromtimestamp(stat.st_mtime, tz=UTC).isoformat(),
|
|
"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
|
|
|
|
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 chemin_exclu(vault_name, rel_path):
|
|
continue
|
|
|
|
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=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,
|
|
}
|