diff --git a/py/api/__init__.py b/py/api/__init__.py index cf2fa251..1a7d5b60 100644 --- a/py/api/__init__.py +++ b/py/api/__init__.py @@ -278,6 +278,8 @@ class PromptManagerAPI( self._long_jobs: dict = {} # The current output scan (py/api/scan_job.ScanJob), if one was started. self._scan_job = None + # The current duplicate scan (also a ScanJob), if one was started. + self._duplicate_job = None # Run cleanup on initialization to remove any existing duplicates try: diff --git a/py/api/admin.py b/py/api/admin.py index 6fbf9241..1af21360 100644 --- a/py/api/admin.py +++ b/py/api/admin.py @@ -8,6 +8,7 @@ import json import os import sqlite3 import tempfile +import threading from pathlib import Path from aiohttp import web @@ -298,6 +299,14 @@ class AdminRoutesMixin: async def scan_duplicates_route(request): return await self.scan_duplicates_endpoint(request) + @routes.post("/prompt_manager/scan_duplicates/stream") + async def scan_duplicates_stream_route(request): + return await self.scan_duplicates_stream(request) + + @routes.get("/prompt_manager/scan_duplicates/status") + async def scan_duplicates_status_route(request): + return await self.scan_duplicates_status(request) + @routes.post("/prompt_manager/delete_duplicate_images") async def delete_duplicate_images_route(request): return await self.delete_duplicate_images_endpoint(request) @@ -375,6 +384,89 @@ class AdminRoutesMixin: status=500, ) + async def scan_duplicates_stream(self, request): + """Run the duplicate scan as a background job and stream its progress. + + A request while a scan is running attaches to it. The job holds the + duplicates lock, so a delete is refused (409) until it finishes. + """ + job = self._duplicate_job + if job is None or job.finished: + guard = self._acquire_long_job(self.LONG_JOB_DUPLICATES) + if guard is None: + return self._long_job_busy_response("Duplicate scan or delete") + job = ScanJob() + self._duplicate_job = job + job.task = asyncio.ensure_future(self._run_duplicate_job(job, guard)) + return await self._stream_scan_job(request, job) + + async def scan_duplicates_status(self, request): + """Whether a duplicate scan is running, and its latest event.""" + job = self._duplicate_job + return web.json_response( + { + "success": True, + "running": bool(job and job.running), + "last_event": job.last_event if job else None, + } + ) + + async def _run_duplicate_job(self, job, guard): + loop = asyncio.get_running_loop() + counts = {"processed": 0} + + def report(done, total): + # Called from worker threads; hand the event to the loop. + counts["processed"] = done + pct = int(done / total * 100) if total else 100 + loop.call_soon_threadsafe( + job.publish, + { + "type": "progress", + "progress": pct, + "status": f"Hashing file {done}/{total}...", + "processed": done, + "found": 0, + }, + ) + + try: + async with guard: + job.publish( + { + "type": "progress", + "progress": 0, + "status": "Collecting media files...", + "processed": 0, + "found": 0, + } + ) + output_dir = await self._run_in_executor(self._find_comfyui_output_dir) + groups = await self._run_in_executor( + self._find_duplicate_images_sync, output_dir, report + ) + job.publish( + { + "type": "complete", + "processed": counts["processed"], + "found": len(groups), + "duplicates": groups, + "message": f"Found {len(groups)} groups of duplicate images", + } + ) + except Exception: + self.logger.exception("Duplicate scan error") + job.publish( + { + "type": "error", + "message": ( + "An internal error occurred. Check server logs for details." + ), + } + ) + finally: + job.finish() + async def cleanup_duplicates_endpoint(self, request): """Cleanup duplicate prompts endpoint.""" try: @@ -400,8 +492,15 @@ class AdminRoutesMixin: output_dir = await self._run_in_executor(self._find_comfyui_output_dir) return await self._run_in_executor(self._find_duplicate_images_sync, output_dir) - def _find_duplicate_images_sync(self, output_dir): - """Group media files under ``output_dir`` by content hash (blocking).""" + def _find_duplicate_images_sync(self, output_dir, progress=None): + """Group media files under ``output_dir`` by content hash (blocking). + + Files are hashed on ``PromptManagerConfig.WORKER_THREADS`` threads. + ``progress(done, total)`` is called from those threads about every + percent and once at the end. + """ + from ..config import PromptManagerConfig + if not output_dir or not os.path.isdir(output_dir): self.logger.warning("ComfyUI output directory not found") return [] @@ -409,21 +508,35 @@ class AdminRoutesMixin: output_path = Path(output_dir) extensions = IMAGE_EXTENSIONS + VIDEO_EXTENSIONS media_files = _collect_media_files([output_path], extensions) - self.logger.info(f"Found {len(media_files)} media files to analyze") + total = len(media_files) + self.logger.info(f"Found {total} media files to analyze") + step = max(1, total // 100) + counter = {"done": 0} + lock = threading.Lock() - file_hashes = {} - for index, media_path in enumerate(media_files, start=1): + def describe(media_path): try: info = self._describe_media_file(media_path, output_path) except Exception as e: - self.logger.error(f"Error processing file {media_path}: {e}") - continue - file_hashes.setdefault(info["hash"], []).append(info) - if index % 100 == 0: + self.logger.error(f"Error processing file {media_path.name}: {e}") + info = None + with lock: + counter["done"] += 1 + done = counter["done"] + if done % 100 == 0: self.logger.info( - f"Processed {index}/{len(media_files)} files " - "for duplicate detection" + f"Processed {done}/{total} files " "for duplicate detection" ) + if progress and (done % step == 0 or done == total): + progress(done, total) + return info + + infos = map_parallel(describe, media_files, PromptManagerConfig.WORKER_THREADS) + + file_hashes = {} + for info in infos: + if info is not None: + file_hashes.setdefault(info["hash"], []).append(info) duplicates = [] for file_hash, images in file_hashes.items(): diff --git a/tests/js/gallery-settings.test.js b/tests/js/gallery-settings.test.js index 44098a53..c3681a29 100644 --- a/tests/js/gallery-settings.test.js +++ b/tests/js/gallery-settings.test.js @@ -55,3 +55,9 @@ test("gridColumnsStyle produces an explicit grid template", () => { assert.equal(gridColumnsStyle("6"), "repeat(6, minmax(0, 1fr))"); assert.equal(gridColumnsStyle("x"), "repeat(8, minmax(0, 1fr))"); }); + +test("infinite scrolling is off by default and must be a boolean", () => { + assert.equal(DEFAULT_SETTINGS.infiniteScroll, false); + assert.equal(normalizeGallerySettings({ infiniteScroll: true }).infiniteScroll, true); + assert.equal(normalizeGallerySettings({ infiniteScroll: "yes" }).infiniteScroll, false); +}); diff --git a/tests/js/list-state.test.js b/tests/js/list-state.test.js index 0c2186ed..3aa11d20 100644 --- a/tests/js/list-state.test.js +++ b/tests/js/list-state.test.js @@ -1,7 +1,7 @@ // Run with: node --test tests/js/ const test = require("node:test"); const assert = require("node:assert/strict"); -const { nextPageState, shouldSaveEdit, hasMorePages } = require("../../web/js/list-state.js"); +const { nextPageState, shouldSaveEdit, hasMorePages, shouldLoadMore } = require("../../web/js/list-state.js"); test("hasMorePages follows pagination.has_more only while pages keep arriving", () => { assert.equal(hasMorePages({ pagination: { has_more: true } }, 5000), true); @@ -49,3 +49,17 @@ test("shouldSaveEdit is true only for a non-empty text that changed", () => { assert.equal(shouldSaveEdit("old", " "), false, "an emptied prompt is not saved"); assert.equal(shouldSaveEdit("old", null), false); }); + +test("shouldLoadMore asks for the next page only near the bottom with pages left", () => { + const base = { + infiniteScroll: true, loading: false, page: 1, limit: 100, total: 1000, + scrollTop: 0, clientHeight: 800, scrollHeight: 4000, threshold: 600, + }; + assert.equal(shouldLoadMore(base), false, "far from the bottom"); + assert.equal(shouldLoadMore({ ...base, scrollTop: 2700 }), true, "within the threshold"); + assert.equal(shouldLoadMore({ ...base, scrollTop: 2700, loading: true }), false, "a load is in flight"); + assert.equal(shouldLoadMore({ ...base, scrollTop: 2700, infiniteScroll: false }), false, "setting off"); + assert.equal(shouldLoadMore({ ...base, scrollTop: 2700, page: 10 }), false, "last page reached"); + assert.equal(shouldLoadMore({ ...base, scrollHeight: 700 }), true, "content shorter than the viewport"); + assert.equal(shouldLoadMore({ ...base, total: 0, scrollHeight: 700 }), false, "nothing to load"); +}); diff --git a/tests/test_api_admin.py b/tests/test_api_admin.py index 06bfcc35..180d09c2 100644 --- a/tests/test_api_admin.py +++ b/tests/test_api_admin.py @@ -1788,5 +1788,134 @@ class TestScanReadsVideos(AdminAPITestCase): self.assertEqual([line for line in cm.output if line.startswith("ERROR")], []) +def _sse_payloads(body): + return [ + json.loads(line[6:]) for line in body.splitlines() if line.startswith("data: ") + ] + + +class TestDuplicateScanIsABackgroundJob(AdminAPITestCase): + """The duplicate scan streams progress, survives the client, and is single-flight + with the duplicate delete.""" + + STREAM = "/prompt_manager/scan_duplicates/stream" + STATUS = "/prompt_manager/scan_duplicates/status" + + def _two_dups_one_single(self): + (self.output_dir / "a.png").write_bytes(b"same bytes") + (self.output_dir / "b.png").write_bytes(b"same bytes") + (self.output_dir / "c.png").write_bytes(b"other bytes") + GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir)] + + def _gate_hashing(self): + import threading + + gate = threading.Event() + calls = [] + original = self.api._calculate_file_hash + + def gated(path): + calls.append(path) + gate.wait(5) + return original(path) + + self.api._calculate_file_hash = gated + return gate, calls + + async def test_stream_reports_progress_then_the_groups(self): + self._two_dups_one_single() + + body = await (await self.client.request("POST", self.STREAM)).text() + events = _sse_payloads(body) + + types = [e["type"] for e in events] + self.assertIn("progress", types) + self.assertEqual(types[-1], "complete") + done = events[-1] + self.assertEqual(done["processed"], 3) + self.assertEqual(done["found"], 1) + self.assertEqual(done["duplicates"][0]["count"], 2) + self.assertEqual( + sorted(i["filename"] for i in done["duplicates"][0]["images"]), + ["a.png", "b.png"], + ) + self.assertNotIn(self.tmpdir, body) + + async def test_second_request_attaches_and_delete_is_refused_meanwhile(self): + self._two_dups_one_single() + gate, calls = self._gate_hashing() + + first = asyncio.ensure_future(self.client.request("POST", self.STREAM)) + await asyncio.sleep(0.1) + second = await asyncio.wait_for(self.client.request("POST", self.STREAM), 5) + self.assertEqual(second.status, 200) + delete = await self.client.request( + "POST", + "/prompt_manager/delete_duplicate_images", + json={"image_paths": ["a.png"]}, + ) + self.assertEqual(delete.status, 409) + running = await (await self.client.request("GET", self.STATUS)).json() + self.assertTrue(running["running"]) + + gate.set() + first_body = await (await first).text() + second_body = await second.text() + finished = await (await self.client.request("GET", self.STATUS)).json() + + self.assertIn('"type": "complete"', first_body) + self.assertIn('"type": "complete"', second_body) + self.assertEqual(len(calls), 3) # one scan, each file hashed once + self.assertFalse(finished["running"]) + self.assertEqual(finished["last_event"]["found"], 1) + + async def test_status_is_idle_before_any_scan(self): + data = await (await self.client.request("GET", self.STATUS)).json() + self.assertEqual(data, {"success": True, "running": False, "last_event": None}) + + async def test_failed_scan_streams_an_error_and_releases_the_lock(self): + GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir)] + self.api._find_comfyui_output_dir = _raise + + body = await (await self.client.request("POST", self.STREAM)).text() + self.api._find_comfyui_output_dir = lambda: str(self.output_dir) + legacy = await self.client.request("GET", "/prompt_manager/scan_duplicates") + + self.assertEqual(_sse_payloads(body)[-1]["type"], "error") + self.assertNotIn("boom", body) + self.assertEqual(legacy.status, 200) + + def test_hashing_runs_on_worker_threads_and_reports_progress(self): + import threading + import time + + for i in range(8): + (self.output_dir / f"f{i}.png").write_bytes(f"bytes {i % 4}".encode()) + PromptManagerConfig.WORKER_THREADS = 4 + seen = set() + lock = threading.Lock() + original = self.api._calculate_file_hash + + def recording(path): + with lock: + seen.add(threading.current_thread().name) + time.sleep(0.05) + return original(path) + + self.api._calculate_file_hash = recording + reports = [] + + groups = self.api._find_duplicate_images_sync( + str(self.output_dir), + progress=lambda done, total: reports.append((done, total)), + ) + + self.assertEqual(len(groups), 4) + self.assertGreaterEqual(len(seen), 2) + self.assertEqual(reports[-1], (8, 8)) + self.assertEqual([t for _, t in reports], [8] * len(reports)) + self.assertEqual(sorted(d for d, _ in reports), [d for d, _ in reports]) + + if __name__ == "__main__": unittest.main() diff --git a/web/gallery.html b/web/gallery.html index be411101..d494247e 100644 --- a/web/gallery.html +++ b/web/gallery.html @@ -8,6 +8,7 @@ + @@ -120,7 +121,7 @@ -
+
@@ -295,6 +296,13 @@
+
+
+ +

Load the next page as you scroll instead of showing page controls

+
+ +
@@ -398,6 +406,12 @@
+ + +
-
-
- - - - - Scanning for duplicate images... +
+
+
+ Scanning for duplicate images + Starting... +
+
+
+
+

Every file is hashed, so this takes a while on a large library. You can close this window; the scan keeps running and the results will be waiting.