fix: clear thumbnails across all configured gallery roots
This commit is contained in:
Vito Sansevero
+84
-96
@@ -1251,119 +1251,107 @@ class ImageRoutesMixin:
|
||||
)
|
||||
return False
|
||||
|
||||
async def clear_thumbnails(self, request):
|
||||
"""Safely clear only our generated thumbnails, never touch original images."""
|
||||
try:
|
||||
async def clear_thumbnails(self, request: web.Request) -> web.Response:
|
||||
"""Clear generated thumbnails across gallery roots, preserving originals.
|
||||
|
||||
# Find ComfyUI output directory
|
||||
output_dir = self._find_comfyui_output_dir()
|
||||
if not output_dir:
|
||||
Invalid thumbnail directories return 400; unavailable gallery roots
|
||||
return 404, and unexpected filesystem failures return 500.
|
||||
"""
|
||||
try:
|
||||
output_dirs = self._get_all_output_dirs()
|
||||
if not output_dirs:
|
||||
return web.json_response(
|
||||
{"success": False, "error": "ComfyUI output directory not found"},
|
||||
status=404,
|
||||
)
|
||||
|
||||
output_path = Path(output_dir)
|
||||
thumbnails_dir = output_path / "thumbnails"
|
||||
|
||||
if not thumbnails_dir.exists():
|
||||
return web.json_response(
|
||||
{
|
||||
"success": True,
|
||||
"message": "No thumbnails directory found - nothing to clear",
|
||||
"cleared_files": 0,
|
||||
}
|
||||
)
|
||||
|
||||
# Verify this is actually our thumbnails directory: a real child
|
||||
# of the output root (symlinks resolved), not merely a path that
|
||||
# shares its prefix such as "<root>2/thumbnails".
|
||||
try:
|
||||
if (
|
||||
not _is_within(thumbnails_dir, output_path)
|
||||
or thumbnails_dir.name != "thumbnails"
|
||||
):
|
||||
self.logger.error(
|
||||
"Safety check failed: thumbnails directory path invalid: "
|
||||
f"{thumbnails_dir}"
|
||||
)
|
||||
return web.json_response(
|
||||
{
|
||||
"success": False,
|
||||
"error": "Safety check failed: invalid thumbnails "
|
||||
"directory path",
|
||||
},
|
||||
status=400,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Path validation failed: {e}")
|
||||
return web.json_response(
|
||||
{"success": False, "error": "Path validation failed"}, status=500
|
||||
)
|
||||
|
||||
# Count files before deletion
|
||||
cleared_count = 0
|
||||
cleared_size = 0
|
||||
|
||||
for root, dirs, files in os.walk(thumbnails_dir):
|
||||
for file in files:
|
||||
file_path = Path(root) / file
|
||||
|
||||
if "_thumb" in file.lower() and any(
|
||||
file.lower().endswith(ext)
|
||||
for ext in [".png", ".jpg", ".jpeg", ".webp", ".gif"]
|
||||
):
|
||||
try:
|
||||
file_size = file_path.stat().st_size
|
||||
file_path.unlink()
|
||||
cleared_count += 1
|
||||
cleared_size += file_size
|
||||
self.logger.debug(f"Cleared thumbnail: {file_path}")
|
||||
except Exception as e:
|
||||
self.logger.warning(
|
||||
f"Failed to delete thumbnail {file_path}: {e}"
|
||||
)
|
||||
|
||||
# Remove empty directories within thumbnails folder
|
||||
try:
|
||||
for root, dirs, files in os.walk(thumbnails_dir, topdown=False):
|
||||
if root != str(thumbnails_dir):
|
||||
try:
|
||||
Path(root).rmdir()
|
||||
except OSError:
|
||||
pass
|
||||
except Exception as e:
|
||||
self.logger.debug(f"Directory cleanup info: {e}")
|
||||
|
||||
def format_size(bytes_size):
|
||||
for unit in ["B", "KB", "MB", "GB"]:
|
||||
if bytes_size < 1024.0:
|
||||
return f"{bytes_size:.1f} {unit}"
|
||||
bytes_size /= 1024.0
|
||||
return f"{bytes_size:.1f} TB"
|
||||
|
||||
self.logger.info(
|
||||
f"Thumbnail cleanup: cleared {cleared_count} files "
|
||||
f"({format_size(cleared_size)})"
|
||||
cleared_count, cleared_size = await self._run_in_executor(
|
||||
self._clear_gallery_thumbnails_sync, output_dirs
|
||||
)
|
||||
self.invalidate_gallery_cache()
|
||||
|
||||
def format_size(bytes_size: int) -> str:
|
||||
size = float(bytes_size)
|
||||
for unit in ["B", "KB", "MB", "GB"]:
|
||||
if size < 1024.0:
|
||||
return f"{size:.1f} {unit}"
|
||||
size /= 1024.0
|
||||
return f"{size:.1f} TB"
|
||||
|
||||
size_formatted = format_size(cleared_size)
|
||||
self.logger.info(
|
||||
f"Thumbnail cleanup: cleared {cleared_count} files ({size_formatted})"
|
||||
)
|
||||
return web.json_response(
|
||||
{
|
||||
"success": True,
|
||||
"cleared_files": cleared_count,
|
||||
"cleared_size": cleared_size,
|
||||
"cleared_size_formatted": format_size(cleared_size),
|
||||
"cleared_size_formatted": size_formatted,
|
||||
"message": f"Cleared {cleared_count} thumbnail files "
|
||||
"({format_size(cleared_size)})",
|
||||
f"({size_formatted})",
|
||||
}
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Clear thumbnails error: {e}")
|
||||
except ValueError as exc:
|
||||
self.logger.error(f"Thumbnail path validation failed: {exc}")
|
||||
return web.json_response(
|
||||
{"success": False, "error": _safe_error(e)}, status=500
|
||||
{"success": False, "error": _safe_error(exc)}, status=400
|
||||
)
|
||||
except Exception as exc:
|
||||
self.logger.error(f"Clear thumbnails error: {exc}")
|
||||
return web.json_response(
|
||||
{"success": False, "error": _safe_error(exc)}, status=500
|
||||
)
|
||||
|
||||
def _clear_gallery_thumbnails_sync(
|
||||
self, output_dirs: list[Path]
|
||||
) -> tuple[int, int]:
|
||||
"""Validate all thumbnail directories before deleting generated files.
|
||||
|
||||
Raises ValueError for symlinked or out-of-root thumbnail directories.
|
||||
Filesystem errors during path validation propagate to the route.
|
||||
"""
|
||||
directories = []
|
||||
for output_path in output_dirs:
|
||||
output_path = Path(output_path)
|
||||
thumbnails_dir = output_path / "thumbnails"
|
||||
if not thumbnails_dir.exists() and not thumbnails_dir.is_symlink():
|
||||
continue
|
||||
if (
|
||||
thumbnails_dir.is_symlink()
|
||||
or not _is_within(thumbnails_dir, output_path)
|
||||
or _canonical(thumbnails_dir) == _canonical(output_path)
|
||||
):
|
||||
raise ValueError(
|
||||
"Safety check failed: invalid thumbnails directory path"
|
||||
)
|
||||
directories.append(thumbnails_dir)
|
||||
|
||||
cleared_count = 0
|
||||
cleared_size = 0
|
||||
for thumbnails_dir in directories:
|
||||
for root, _dirs, files in os.walk(thumbnails_dir):
|
||||
for name in files:
|
||||
path = Path(root) / name
|
||||
if (
|
||||
"_thumb" not in name.lower()
|
||||
or path.suffix.lower() not in GALLERY_IMAGE_EXTENSIONS
|
||||
or path.is_symlink()
|
||||
):
|
||||
continue
|
||||
try:
|
||||
size = path.stat().st_size
|
||||
path.unlink()
|
||||
cleared_count += 1
|
||||
cleared_size += size
|
||||
except OSError as exc:
|
||||
self.logger.warning(f"Failed to delete thumbnail {path}: {exc}")
|
||||
for root, _dirs, _files in os.walk(thumbnails_dir, topdown=False):
|
||||
if Path(root) != thumbnails_dir:
|
||||
try:
|
||||
Path(root).rmdir()
|
||||
except OSError:
|
||||
pass
|
||||
return cleared_count, cleared_size
|
||||
|
||||
@staticmethod
|
||||
def _resolve_client_media_path(raw_path, allowed_dirs):
|
||||
|
||||
@@ -1536,7 +1536,7 @@ class TestClearThumbnails(ImageRouteCoverageCase):
|
||||
URL = "/prompt_manager/images/clear-thumbnails"
|
||||
|
||||
async def test_missing_output_dir_is_404(self):
|
||||
self.api._find_comfyui_output_dir = lambda: None
|
||||
self.api._get_all_output_dirs = lambda: []
|
||||
status, _ = await self._json("POST", self.URL)
|
||||
self.assertEqual(status, 404)
|
||||
|
||||
@@ -1569,7 +1569,7 @@ class TestClearThumbnails(ImageRouteCoverageCase):
|
||||
self.assertEqual(data["cleared_files"], 0)
|
||||
|
||||
async def test_internal_failure_is_500(self):
|
||||
self.api._find_comfyui_output_dir = _raise
|
||||
self.api._get_all_output_dirs = _raise
|
||||
status, _ = await self._json("POST", self.URL)
|
||||
self.assertEqual(status, 500)
|
||||
|
||||
@@ -1581,7 +1581,7 @@ class TestClearThumbnails(ImageRouteCoverageCase):
|
||||
out.mkdir()
|
||||
sibling = make_png(self.comfy_dir / "out2" / "thumbnails" / "a_thumb.png")
|
||||
self._make_symlink(sibling.parent, out / "thumbnails")
|
||||
self.api._find_comfyui_output_dir = lambda: str(out)
|
||||
self.api._get_all_output_dirs = lambda: [out]
|
||||
|
||||
status, data = await self._json("POST", self.URL)
|
||||
|
||||
@@ -1589,6 +1589,79 @@ class TestClearThumbnails(ImageRouteCoverageCase):
|
||||
self.assertFalse(data["success"])
|
||||
self.assertTrue(sibling.is_file())
|
||||
|
||||
async def test_clears_all_roots_and_combines_counts_without_touching_originals(
|
||||
self,
|
||||
):
|
||||
second = self.comfy_dir / "extra-output"
|
||||
roots = [self.output_dir, second]
|
||||
self.api._get_all_output_dirs = lambda: roots
|
||||
expected_size = 0
|
||||
for root in roots:
|
||||
make_png(root / "original_thumb.png")
|
||||
thumb = make_png(root / "thumbnails" / "sub" / "generated_thumb.png")
|
||||
expected_size += thumb.stat().st_size
|
||||
(root / "thumbnails" / "notes.txt").write_text("keep")
|
||||
status, data = await self._json("POST", self.URL)
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(data["cleared_files"], 2)
|
||||
self.assertEqual(data["cleared_size"], expected_size)
|
||||
self.assertNotIn("{format_size", data["message"])
|
||||
for root in roots:
|
||||
self.assertTrue((root / "original_thumb.png").is_file())
|
||||
self.assertTrue((root / "thumbnails" / "notes.txt").is_file())
|
||||
self.assertFalse((root / "thumbnails" / "sub").exists())
|
||||
|
||||
async def test_missing_thumbnails_in_first_root_does_not_skip_later_roots(self):
|
||||
second = self.comfy_dir / "extra-output"
|
||||
thumb = make_png(second / "thumbnails" / "a_thumb.png")
|
||||
self.api._get_all_output_dirs = lambda: [self.output_dir, second]
|
||||
status, data = await self._json("POST", self.URL)
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(data["cleared_files"], 1)
|
||||
self.assertFalse(thumb.exists())
|
||||
|
||||
async def test_invalid_later_root_is_rejected_before_deleting_any_files(self):
|
||||
first_thumb = make_png(self.output_dir / "thumbnails" / "a_thumb.png")
|
||||
second = self.comfy_dir / "extra-output"
|
||||
second.mkdir()
|
||||
outside = make_png(self.comfy_dir / "elsewhere" / "a_thumb.png")
|
||||
self._make_symlink(outside.parent, second / "thumbnails")
|
||||
self.api._get_all_output_dirs = lambda: [self.output_dir, second]
|
||||
status, data = await self._json("POST", self.URL)
|
||||
self.assertEqual(status, 400)
|
||||
self.assertFalse(data["success"])
|
||||
self.assertTrue(first_thumb.is_file())
|
||||
self.assertTrue(outside.is_file())
|
||||
|
||||
async def test_thumbnail_symlink_back_to_root_cannot_delete_originals(self):
|
||||
original = make_png(self.output_dir / "original_thumb.png")
|
||||
self._make_symlink(self.output_dir, self.output_dir / "thumbnails")
|
||||
status, data = await self._json("POST", self.URL)
|
||||
self.assertEqual(status, 400)
|
||||
self.assertFalse(data["success"])
|
||||
self.assertTrue(original.is_file())
|
||||
|
||||
async def test_cleanup_runs_off_loop_and_invalidates_gallery_cache(self):
|
||||
import threading
|
||||
|
||||
make_png(self.output_dir / "thumbnails" / "a_thumb.png")
|
||||
loop_thread = threading.get_ident()
|
||||
deletion_threads = []
|
||||
original_unlink = Path.unlink
|
||||
|
||||
def unlink(path, *args, **kwargs):
|
||||
deletion_threads.append(threading.get_ident())
|
||||
return original_unlink(path, *args, **kwargs)
|
||||
|
||||
self.api._gallery_cache = {"previous": object()}
|
||||
with patch.object(Path, "unlink", unlink):
|
||||
status, data = await self._json("POST", self.URL)
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(data["cleared_files"], 1)
|
||||
self.assertTrue(deletion_threads)
|
||||
self.assertNotIn(loop_thread, deletion_threads)
|
||||
self.assertEqual(self.api._gallery_cache, {})
|
||||
|
||||
|
||||
class TestImagePromptLookup(ImageRouteCoverageCase):
|
||||
|
||||
|
||||
Reference in New Issue
Block a user