bruno
4c4e415975
CI / lint (push) Successful in 58s
CI / security (push) Successful in 40s
CI / test (push) Successful in 1m15s
CI / build (push) Successful in 37s
CI / e2e (push) Successful in 10m15s
Desktop Build / build-windows (push) Canceled after 0s
Desktop Build / build-linux (push) Canceled after 0s
103 lines
3.7 KiB
Python
103 lines
3.7 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
|
|
|
|
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}
|