mirror of
https://github.com/willmiao/ComfyUI-Lora-Manager.git
synced 2026-09-21 03:01:27 -03:00
feat(settings): directory picker and live validation for path settings
Add a reusable directory-picker modal backed by a new generic POST /api/lm/browse-directory endpoint (browse logic extracted from the recipe batch-import handler into py/utils/directory_browser.py) and wire a browse button plus advisory validate-path feedback (POST /api/lm/validate-path) into the settings path inputs: recipes path, example images path/local root, and the extra-folder/model-path rows. The browse button insets into the right edge of static inputs so narrow settings rows keep their single-control layout. Translations for the new settings.directoryPicker and settings.pathValidation keys are filled in for all 9 locales.
This commit is contained in:
Will Miao
@@ -61,6 +61,7 @@ from .agent_handlers import AgentHandler
|
||||
from .download_routing_handlers import DownloadRoutingHandler
|
||||
from .model_handlers import ModelCivitaiHandler
|
||||
from ...utils.civitai_utils import rewrite_preview_url
|
||||
from ...utils.directory_browser import browse_directory
|
||||
from ...utils.example_images_paths import (
|
||||
find_non_compliant_items_in_example_images_root,
|
||||
is_valid_example_images_root,
|
||||
@@ -3497,6 +3498,76 @@ class FileSystemHandler:
|
||||
logger.error("Failed to open wildcards location: %s", exc, exc_info=True)
|
||||
return web.json_response({"success": False, "error": str(exc)}, status=500)
|
||||
|
||||
async def browse_directory(self, request: web.Request) -> web.Response:
|
||||
"""Browse a directory for the settings-UI directory picker."""
|
||||
try:
|
||||
data = await request.json()
|
||||
payload, status = browse_directory(data.get("path", ""))
|
||||
return web.json_response(payload, status=status)
|
||||
except json.JSONDecodeError:
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Invalid JSON"}, status=400
|
||||
)
|
||||
except Exception as exc: # pragma: no cover - defensive logging
|
||||
logger.error("Failed to browse directory: %s", exc, exc_info=True)
|
||||
return web.json_response({"success": False, "error": str(exc)}, status=500)
|
||||
|
||||
async def validate_path(self, request: web.Request) -> web.Response:
|
||||
"""Validate a filesystem path for the settings UI.
|
||||
|
||||
A well-formed request always returns HTTP 200; invalid paths are
|
||||
reported via ``error_code`` in the payload. HTTP 400 is reserved for
|
||||
malformed requests (missing path, invalid JSON).
|
||||
"""
|
||||
try:
|
||||
data = await request.json()
|
||||
raw_path = data.get("path")
|
||||
expect = data.get("expect", "directory")
|
||||
|
||||
if not raw_path or not isinstance(raw_path, str):
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Missing path parameter"}, status=400
|
||||
)
|
||||
|
||||
# Business path convention: abspath only, never realpath.
|
||||
path = os.path.abspath(os.path.expanduser(raw_path))
|
||||
|
||||
exists = os.path.exists(path)
|
||||
is_directory = os.path.isdir(path) if exists else False
|
||||
readable = bool(exists and os.access(path, os.R_OK))
|
||||
writable = bool(exists and os.access(path, os.W_OK))
|
||||
|
||||
error_code = None
|
||||
if not exists:
|
||||
error_code = "path_not_found"
|
||||
elif expect == "directory" and not is_directory:
|
||||
error_code = "not_a_directory"
|
||||
elif expect == "file" and not os.path.isfile(path):
|
||||
error_code = "not_a_file"
|
||||
elif not readable:
|
||||
error_code = "not_readable"
|
||||
elif not writable:
|
||||
error_code = "not_writable"
|
||||
|
||||
return web.json_response(
|
||||
{
|
||||
"success": True,
|
||||
"path": path,
|
||||
"exists": exists,
|
||||
"is_directory": is_directory,
|
||||
"readable": readable,
|
||||
"writable": writable,
|
||||
"error_code": error_code,
|
||||
}
|
||||
)
|
||||
except json.JSONDecodeError:
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Invalid JSON"}, status=400
|
||||
)
|
||||
except Exception as exc: # pragma: no cover - defensive logging
|
||||
logger.error("Failed to validate path: %s", exc, exc_info=True)
|
||||
return web.json_response({"success": False, "error": str(exc)}, status=500)
|
||||
|
||||
|
||||
class CustomWordsHandler:
|
||||
"""Handler for autocomplete via TagFTSIndex."""
|
||||
@@ -4141,6 +4212,8 @@ class MiscHandlerSet:
|
||||
"open_settings_location": self.filesystem.open_settings_location,
|
||||
"open_backup_location": self.filesystem.open_backup_location,
|
||||
"open_wildcards_location": self.filesystem.open_wildcards_location,
|
||||
"browse_directory": self.filesystem.browse_directory,
|
||||
"validate_path": self.filesystem.validate_path,
|
||||
"search_custom_words": self.custom_words.search_custom_words,
|
||||
"search_wildcards": self.wildcards.search_wildcards,
|
||||
"get_supporters": self.supporters.get_supporters,
|
||||
|
||||
@@ -9,7 +9,6 @@ import re
|
||||
import asyncio
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Awaitable, Callable, Dict, List, Mapping, Optional, Protocol, Tuple
|
||||
|
||||
from aiohttp import web
|
||||
@@ -34,6 +33,7 @@ from ...utils.civitai_utils import (
|
||||
rewrite_preview_url,
|
||||
)
|
||||
from ...utils.constants import NSFW_LEVELS
|
||||
from ...utils.directory_browser import WINDOWS_DRIVES_TOKEN, browse_directory
|
||||
from ...utils.exif_utils import ExifUtils
|
||||
from ...utils.recipe_open_stats import RecipeOpenStats
|
||||
from ...recipes.merger import GenParamsMerger
|
||||
@@ -3124,11 +3124,10 @@ class RecipeWorkflowHandler:
|
||||
class BatchImportHandler:
|
||||
"""Handle batch import operations for recipes."""
|
||||
|
||||
# Virtual path token for the Windows drive list. Browsing up from a drive
|
||||
# root (e.g. C:\) lands here so users can switch drives without typing a
|
||||
# path. Only meaningful on Windows; elsewhere it falls through to normal
|
||||
# path handling and fails the existence check.
|
||||
WINDOWS_DRIVES_TOKEN = "__drives__"
|
||||
# Virtual path token for the Windows drive list. Kept as a class
|
||||
# attribute for backwards compatibility; the canonical definition lives
|
||||
# in py/utils/directory_browser.py.
|
||||
WINDOWS_DRIVES_TOKEN = WINDOWS_DRIVES_TOKEN
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -3301,131 +3300,8 @@ class BatchImportHandler:
|
||||
"""Browse a directory and return its contents (subdirectories and files)."""
|
||||
try:
|
||||
data = await request.json()
|
||||
directory_path = data.get("path", "")
|
||||
|
||||
if os.name == "nt" and directory_path == self.WINDOWS_DRIVES_TOKEN:
|
||||
return self._windows_drives_response()
|
||||
|
||||
# Default to the user's home directory. The frontend previously
|
||||
# sent "/" as the initial path, which is POSIX-only: on Windows it
|
||||
# resolves to the current drive root and then fails the access
|
||||
# check below.
|
||||
if not directory_path:
|
||||
path = Path.home()
|
||||
else:
|
||||
path = Path(directory_path).expanduser().resolve()
|
||||
|
||||
# Access check: browsing intentionally covers the whole server
|
||||
# filesystem (the server operator browses their own machine). On
|
||||
# POSIX every absolute path is under "/", but Path("/") has no
|
||||
# drive letter on Windows and can never anchor a drive-qualified
|
||||
# path in relative_to(), so test for a drive there instead.
|
||||
if os.name == "nt":
|
||||
is_allowed = bool(path.drive)
|
||||
else:
|
||||
is_allowed = path.is_absolute()
|
||||
|
||||
if not is_allowed:
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Access denied to this directory"},
|
||||
status=403,
|
||||
)
|
||||
|
||||
if not path.exists():
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Directory does not exist"},
|
||||
status=404,
|
||||
)
|
||||
|
||||
if not path.is_dir():
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Path is not a directory"},
|
||||
status=400,
|
||||
)
|
||||
|
||||
# List directory contents
|
||||
directories = []
|
||||
image_files = []
|
||||
|
||||
image_extensions = {
|
||||
".jpg",
|
||||
".jpeg",
|
||||
".png",
|
||||
".gif",
|
||||
".webp",
|
||||
".bmp",
|
||||
".tiff",
|
||||
".tif",
|
||||
}
|
||||
|
||||
try:
|
||||
for item in path.iterdir():
|
||||
try:
|
||||
if item.is_dir():
|
||||
# Skip hidden directories and common system folders
|
||||
if not item.name.startswith(".") and item.name not in [
|
||||
"__pycache__",
|
||||
"node_modules",
|
||||
]:
|
||||
directories.append(
|
||||
{
|
||||
"name": item.name,
|
||||
"path": str(item),
|
||||
"is_parent": False,
|
||||
}
|
||||
)
|
||||
elif item.is_file() and item.suffix.lower() in image_extensions:
|
||||
image_files.append(
|
||||
{
|
||||
"name": item.name,
|
||||
"path": str(item),
|
||||
"size": item.stat().st_size,
|
||||
}
|
||||
)
|
||||
except (PermissionError, OSError):
|
||||
# Skip files/directories we can't access
|
||||
continue
|
||||
|
||||
# Sort directories and files alphabetically
|
||||
directories.sort(key=lambda x: x["name"].lower())
|
||||
image_files.sort(key=lambda x: x["name"].lower())
|
||||
|
||||
# Parent directory. A filesystem root is its own parent
|
||||
# (parent == path): POSIX "/" gets no parent, while a Windows
|
||||
# drive root (C:\) links up to the virtual drive list so users
|
||||
# can switch drives. The previous str(path) != str(path.root)
|
||||
# check misfired on Windows, where a drive root's parent is
|
||||
# itself, producing an infinite self-loop.
|
||||
if path.parent == path:
|
||||
parent_path = (
|
||||
self.WINDOWS_DRIVES_TOKEN if os.name == "nt" else None
|
||||
)
|
||||
else:
|
||||
parent_path = str(path.parent)
|
||||
|
||||
return web.json_response(
|
||||
{
|
||||
"success": True,
|
||||
"current_path": str(path),
|
||||
"parent_path": parent_path,
|
||||
"directories": directories,
|
||||
"image_files": image_files,
|
||||
"image_count": len(image_files),
|
||||
"directory_count": len(directories),
|
||||
}
|
||||
)
|
||||
|
||||
except PermissionError:
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Permission denied"},
|
||||
status=403,
|
||||
)
|
||||
except OSError as exc:
|
||||
return web.json_response(
|
||||
{"success": False, "error": f"Error reading directory: {str(exc)}"},
|
||||
status=500,
|
||||
)
|
||||
|
||||
payload, status = browse_directory(data.get("path", ""))
|
||||
return web.json_response(payload, status=status)
|
||||
except json.JSONDecodeError:
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Invalid JSON"},
|
||||
@@ -3434,30 +3310,3 @@ class BatchImportHandler:
|
||||
except Exception as exc:
|
||||
self._logger.error("Error browsing directory: %s", exc, exc_info=True)
|
||||
return web.json_response({"success": False, "error": str(exc)}, status=500)
|
||||
|
||||
def _windows_drives_response(self) -> web.Response:
|
||||
"""List available drive letters as a virtual directory (Windows only)."""
|
||||
try:
|
||||
drives = os.listdrives()
|
||||
except AttributeError: # Python < 3.12
|
||||
drives = [
|
||||
f"{letter}:\\"
|
||||
for letter in "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
|
||||
if os.path.exists(f"{letter}:\\")
|
||||
]
|
||||
directories = [
|
||||
{"name": drive, "path": drive, "is_parent": False} for drive in drives
|
||||
]
|
||||
return web.json_response(
|
||||
{
|
||||
"success": True,
|
||||
# Empty current_path marks the virtual level; the frontend
|
||||
# disables folder selection there.
|
||||
"current_path": "",
|
||||
"parent_path": None,
|
||||
"directories": directories,
|
||||
"image_files": [],
|
||||
"image_count": 0,
|
||||
"directory_count": len(directories),
|
||||
}
|
||||
)
|
||||
|
||||
@@ -37,6 +37,8 @@ MISC_ROUTE_DEFINITIONS: tuple[RouteDefinition, ...] = (
|
||||
RouteDefinition("GET", "/api/lm/wildcards/search", "search_wildcards"),
|
||||
RouteDefinition("POST", "/api/lm/wildcards/open-location", "open_wildcards_location"),
|
||||
RouteDefinition("POST", "/api/lm/open-file-location", "open_file_location"),
|
||||
RouteDefinition("POST", "/api/lm/browse-directory", "browse_directory"),
|
||||
RouteDefinition("POST", "/api/lm/validate-path", "validate_path"),
|
||||
RouteDefinition("POST", "/api/lm/update-usage-stats", "update_usage_stats"),
|
||||
RouteDefinition("GET", "/api/lm/get-usage-stats", "get_usage_stats"),
|
||||
RouteDefinition("POST", "/api/lm/update-lora-code", "update_lora_code"),
|
||||
|
||||
@@ -0,0 +1,152 @@
|
||||
"""Shared directory-browsing logic for HTTP directory pickers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Tuple
|
||||
|
||||
# Virtual path token for the Windows drive list. Browsing up from a drive
|
||||
# root (e.g. C:\) lands here so users can switch drives without typing a
|
||||
# path. Only meaningful on Windows; elsewhere it falls through to normal
|
||||
# path handling and fails the existence check.
|
||||
WINDOWS_DRIVES_TOKEN = "__drives__"
|
||||
|
||||
_IMAGE_EXTENSIONS = {
|
||||
".jpg",
|
||||
".jpeg",
|
||||
".png",
|
||||
".gif",
|
||||
".webp",
|
||||
".bmp",
|
||||
".tiff",
|
||||
".tif",
|
||||
}
|
||||
|
||||
|
||||
def browse_directory(directory_path: str) -> Tuple[Dict[str, Any], int]:
|
||||
"""Browse a directory and return (payload, http_status).
|
||||
|
||||
The payload shape matches the JSON responses historically produced by
|
||||
``BatchImportHandler.browse_directory``: on success a dict with
|
||||
``success``, ``current_path``, ``parent_path``, ``directories``,
|
||||
``image_files``, ``image_count`` and ``directory_count``; on failure a
|
||||
``{"success": False, "error": ...}`` dict with a 400/403/404/500 status.
|
||||
"""
|
||||
if os.name == "nt" and directory_path == WINDOWS_DRIVES_TOKEN:
|
||||
return _windows_drives_payload(), 200
|
||||
|
||||
# Default to the user's home directory. The frontend previously
|
||||
# sent "/" as the initial path, which is POSIX-only: on Windows it
|
||||
# resolves to the current drive root and then fails the access
|
||||
# check below.
|
||||
if not directory_path:
|
||||
path = Path.home()
|
||||
else:
|
||||
path = Path(directory_path).expanduser().resolve()
|
||||
|
||||
# Access check: browsing intentionally covers the whole server
|
||||
# filesystem (the server operator browses their own machine). On
|
||||
# POSIX every absolute path is under "/", but Path("/") has no
|
||||
# drive letter on Windows and can never anchor a drive-qualified
|
||||
# path in relative_to(), so test for a drive there instead.
|
||||
if os.name == "nt":
|
||||
is_allowed = bool(path.drive)
|
||||
else:
|
||||
is_allowed = path.is_absolute()
|
||||
|
||||
if not is_allowed:
|
||||
return {"success": False, "error": "Access denied to this directory"}, 403
|
||||
|
||||
if not path.exists():
|
||||
return {"success": False, "error": "Directory does not exist"}, 404
|
||||
|
||||
if not path.is_dir():
|
||||
return {"success": False, "error": "Path is not a directory"}, 400
|
||||
|
||||
directories = []
|
||||
image_files = []
|
||||
|
||||
try:
|
||||
for item in path.iterdir():
|
||||
try:
|
||||
if item.is_dir():
|
||||
# Skip hidden directories and common system folders
|
||||
if not item.name.startswith(".") and item.name not in [
|
||||
"__pycache__",
|
||||
"node_modules",
|
||||
]:
|
||||
directories.append(
|
||||
{
|
||||
"name": item.name,
|
||||
"path": str(item),
|
||||
"is_parent": False,
|
||||
}
|
||||
)
|
||||
elif item.is_file() and item.suffix.lower() in _IMAGE_EXTENSIONS:
|
||||
image_files.append(
|
||||
{
|
||||
"name": item.name,
|
||||
"path": str(item),
|
||||
"size": item.stat().st_size,
|
||||
}
|
||||
)
|
||||
except (PermissionError, OSError):
|
||||
# Skip files/directories we can't access
|
||||
continue
|
||||
|
||||
directories.sort(key=lambda x: x["name"].lower())
|
||||
image_files.sort(key=lambda x: x["name"].lower())
|
||||
|
||||
# Parent directory. A filesystem root is its own parent
|
||||
# (parent == path): POSIX "/" gets no parent, while a Windows
|
||||
# drive root (C:\) links up to the virtual drive list so users
|
||||
# can switch drives. The previous str(path) != str(path.root)
|
||||
# check misfired on Windows, where a drive root's parent is
|
||||
# itself, producing an infinite self-loop.
|
||||
if path.parent == path:
|
||||
parent_path = WINDOWS_DRIVES_TOKEN if os.name == "nt" else None
|
||||
else:
|
||||
parent_path = str(path.parent)
|
||||
|
||||
return (
|
||||
{
|
||||
"success": True,
|
||||
"current_path": str(path),
|
||||
"parent_path": parent_path,
|
||||
"directories": directories,
|
||||
"image_files": image_files,
|
||||
"image_count": len(image_files),
|
||||
"directory_count": len(directories),
|
||||
},
|
||||
200,
|
||||
)
|
||||
|
||||
except PermissionError:
|
||||
return {"success": False, "error": "Permission denied"}, 403
|
||||
except OSError as exc:
|
||||
return {"success": False, "error": f"Error reading directory: {str(exc)}"}, 500
|
||||
|
||||
|
||||
def _windows_drives_payload() -> Dict[str, Any]:
|
||||
"""List available drive letters as a virtual directory (Windows only)."""
|
||||
try:
|
||||
drives = os.listdrives()
|
||||
except AttributeError: # Python < 3.12
|
||||
drives = [
|
||||
f"{letter}:\\"
|
||||
for letter in "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
|
||||
if os.path.exists(f"{letter}:\\")
|
||||
]
|
||||
directories = [{"name": drive, "path": drive, "is_parent": False} for drive in drives]
|
||||
return {
|
||||
"success": True,
|
||||
# Empty current_path marks the virtual level; the frontend
|
||||
# disables folder selection there.
|
||||
"current_path": "",
|
||||
"parent_path": None,
|
||||
"directories": directories,
|
||||
"image_files": [],
|
||||
"image_count": 0,
|
||||
"directory_count": len(directories),
|
||||
}
|
||||
Reference in New Issue
Block a user