From ad392a99a77f90d4e6d2cf8c7314c73769946cfd Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Thu, 29 Oct 2026 10:03:49 -0700 Subject: [PATCH] feat(images): review and apply Delete images as a reviewed plan job (#480) POST /prompt_manager/images/delete/preview reviews both modes; the maintenance apply takes {token, mode}. Library mode removes the records and writes the removal mark; trash mode moves the file first and removes its records only after a successful move. The workspace listing, the output scan, the duplicate find and Tag output images skip marked files. --- py/api/__init__.py | 3 + py/api/admin.py | 17 +- py/api/image_delete.py | 465 ++++++++++++++++++++++++++++++ py/api/image_tag_jobs.py | 6 +- py/api/images.py | 4 + py/api/jobs.py | 2 + py/api/maintenance.py | 13 +- tests/fake_trash.py | 52 ++++ tests/test_image_delete_routes.py | 281 ++++++++++++++++++ 9 files changed, 839 insertions(+), 4 deletions(-) create mode 100644 py/api/image_delete.py create mode 100644 tests/fake_trash.py create mode 100644 tests/test_image_delete_routes.py diff --git a/py/api/__init__.py b/py/api/__init__.py index b2376662..2b54b9c9 100644 --- a/py/api/__init__.py +++ b/py/api/__init__.py @@ -74,6 +74,7 @@ from .autotag_routes import AutotagRoutesMixin from .lora_integration import LoraIntegrationMixin from .lora_import_routes import LoraImportRoutesMixin from .image_categories import ImageCategoryRoutesMixin +from .image_delete import ImageDeleteMixin from .source_rescope import SourceRescopeMixin try: @@ -343,6 +344,7 @@ class PromptManagerAPI( LoraIntegrationMixin, LoraImportRoutesMixin, ImageCategoryRoutesMixin, + ImageDeleteMixin, SourceRescopeMixin, ): """REST API handler for PromptManager operations and web interface. @@ -710,6 +712,7 @@ class PromptManagerAPI( self._register_lora_routes(routes) self._register_lora_import_routes(routes) self._register_image_category_routes(routes) + self._register_image_delete_routes(routes) self._register_rescope_routes(routes) # Register gzip compression middleware (once) diff --git a/py/api/admin.py b/py/api/admin.py index a7b3bac7..466f6ebe 100644 --- a/py/api/admin.py +++ b/py/api/admin.py @@ -883,6 +883,8 @@ class AdminRoutesMixin: extensions = IMAGE_EXTENSIONS + VIDEO_EXTENSIONS media_files = [] seen = set() + # #480: a file removed from the library is never a copy or a keeper. + marks = self.db.removal_marks() for slot, root in enumerate(roots): for path in _collect_media_files([root], extensions): try: @@ -891,6 +893,12 @@ class AdminRoutesMixin: stat = os.stat(path) except OSError: continue + mark = marks.get(os.path.normcase(os.path.realpath(str(path)))) + if mark is not None and tuple(mark) == ( + stat.st_size, + stat.st_mtime_ns, + ): + continue key = _physical_file_key(path, stat) if key not in seen: seen.add(key) @@ -1551,8 +1559,13 @@ class AdminRoutesMixin: if job.cancelled: return batch = media_files[batch_start : batch_start + SCAN_BATCH_SIZE] - skipped = [f for f in batch if str(f) in known] - to_read = [f for f in batch if str(f) not in known] + # #480: a file removed from the library is not read or linked + # again while its bytes are unchanged. + removed = await self._run_in_executor(self._removed_paths, batch) + skipped = [f for f in batch if str(f) in known or str(f) in removed] + to_read = [ + f for f in batch if str(f) not in known and str(f) not in removed + ] outcomes = [_scan_outcome(f, roots, "skipped") for f in skipped] counts["processed"] += len(skipped) counts["unchanged"] += len(skipped) diff --git a/py/api/image_delete.py b/py/api/image_delete.py new file mode 100644 index 00000000..3e74c3e1 --- /dev/null +++ b/py/api/image_delete.py @@ -0,0 +1,465 @@ +"""Delete selected images: remove them from the library, or trash them (#480). + +``POST /prompt_manager/images/delete/preview {items | filter}`` reviews the +selection for both modes at once and answers a reviewed plan (the System › +Maintenance contract, ``py/api/maintenance.py``): the token, the counts of +the records each file loses, the refused files with a coded reason (both +modes, and the trash-only ones), the prompts that lose their last image and +the safety backup. It writes nothing. ``items`` are Images listing identities +(at most 500); ``filter`` is "Select all N matching". Each target is matched +to its entry in the current listing (``not-listed`` otherwise), so every +later step uses the walked spelling, never the request's text. + +``POST /prompt_manager/maintenance/apply {token, mode}`` runs the plan as the +``image_delete`` job (admission ``records``, not cancellable): the sources +are checked, the promised safety backup is taken, then each file is checked +again (containment and fingerprint) and + +- ``mode="library"``: its records are deleted and its removal mark written + in one transaction; the file stays on disk, hidden; +- ``mode="trash"``: it is moved to the trash first (``utils/trash.py``); a + refused move leaves the file and its records alone; after a move its + records are deleted. + +No path here unlinks a media file. A failed records write stops the job; the +files after it are not attempted. +""" + +import hashlib +import json +import os +import stat as stat_module +from pathlib import Path + +from .admin import DELETABLE_MEDIA_EXTENSIONS, _remove_thumbnail +from .duplicate_review import ( + CHANGED_CONTENT, + GONE, + LINK, + OUTSIDE_SOURCE, + UNREADABLE, + file_problem, +) +from .images import _favorite_relative_key, _favorite_root_key +from .jobs import IMAGE_DELETE_GROUP_NAMES, IMAGE_DELETE_JOB_KIND, JobBusy +from .job_routes import busy_response +from .maintenance import MAINTENANCE_PREVIEW_LEASE, _fail_job, _failure, backup_note + +try: + from ...database.annotation_schema import MAX_LOCATOR_LENGTH + from ...utils import trash +except ImportError: + from database.annotation_schema import MAX_LOCATOR_LENGTH + from utils import trash + +IMAGE_DELETE = "delete_images" +MODES = ("library", "trash") +TITLES = { + "library": "Delete images (remove from the library)", + "trash": "Delete images (move to the trash)", +} +MAX_DETAIL_ROWS = 50 +TRASH_FOLDERS = frozenset({".trash", ".trashes", "$recycle.bin"}) + +# Reason codes (design §4.4) and the sentence each shows. +REASONS = { + "not-listed": "No longer listed in Images.", + "outside": "Outside the output folders.", + "link": "A link, not a file.", + "gone": "The file no longer exists.", + "unreadable": "The file can't be read.", + "thumbnail": "A generated preview, not an image.", + "not-media": "Not a media file.", + "trash-folder": "Inside a trash folder.", + "path-too-long": "The path is too long.", + "changed": "The file changed after the review.", + trash.NO_TRASH: "This drive has no trash.", + trash.NETWORK: "Network drives can't use the trash.", + trash.TRASH_IN_ROOT: "This drive's trash is inside an output folder.", + trash.LOCKED: "The file is open in another program. Close it and try again.", + trash.FAILED: "The file could not be moved to the trash.", + "not-attempted": "Not attempted: the job stopped after an error.", +} +_PROBLEM_CODES = { + OUTSIDE_SOURCE: "outside", + LINK: "link", + GONE: "gone", + UNREADABLE: "unreadable", + CHANGED_CONTENT: "changed", +} +SOURCES_CHANGED = "The sources changed since the review; nothing was changed." +RECORDS_FAILED = "The image records could not be removed; the job stopped." + + +def _is_trash_folder(name: str) -> bool: + folded = name.casefold() + return folded in TRASH_FOLDERS or ( + folded.startswith(".trash-") and folded[7:].isdigit() + ) + + +def lexical_problem(relative: str) -> str | None: + """The refusal code a root-relative POSIX path earns before any file access.""" + parts = relative.split("/") + if "thumbnails" in parts[:-1]: + return "thumbnail" + if any(_is_trash_folder(part) for part in parts[:-1]): + return "trash-folder" + if os.path.splitext(parts[-1])[1].lower() not in DELETABLE_MEDIA_EXTENSIONS: + return "not-media" + return None + + +def file_check(path, root, relative, *, expected=None): + """``(code, lstat)``: why *path* may not be deleted, or ``(None, lstat)``. + + Text first (inside the root, not a preview or trash folder, a media + file), then ``duplicate_review.file_problem``: a regular file, not a + link, whose resolved path is still inside the resolved root. With + *expected* (``{size, mtime_ns, inode}`` from the review) the file must + still be those bytes. Every handle is closed when this returns. + """ + code = lexical_problem(relative) + if code: + return code, None + try: + info = os.lstat(path) + except FileNotFoundError: + return "gone", None + except OSError: + return "unreadable", None + if getattr(info, "st_file_attributes", 0) & getattr( + stat_module, "FILE_ATTRIBUTE_REPARSE_POINT", 0 + ): + return "link", info + if expected is None: + problem = file_problem(path, root, None, info.st_size, hash_check=False) + else: + problem = file_problem( + path, + root, + None, + expected["size"], + hash_check=False, + mtime_ns=expected["mtime_ns"], + inode=expected["inode"], + ) + if problem: + return _PROBLEM_CODES.get(problem, "outside"), info + if len(os.path.normcase(os.path.realpath(path))) > MAX_LOCATOR_LENGTH: + return "path-too-long", info + return None, info + + +def _row(target, code): + return { + "path": target["rel"], + "root_label": target["root_label"], + "code": code, + "detail": REASONS[code], + } + + +def _summary(rows): + counts = {} + for row in rows: + counts[row["code"]] = counts.get(row["code"], 0) + 1 + return {"count": len(rows), "counts": counts, "rows": rows[:MAX_DETAIL_ROWS]} + + +def _fingerprint(targets) -> str: + material = [ + (t["root_index"], t["rel"], t["size"], t["mtime_ns"], t["inode"]) + for t in targets + ] + return hashlib.sha256(json.dumps(material).encode("utf-8")).hexdigest() + + +class ImageDeleteMixin: + """The Images Delete… review and its apply job (#480).""" + + def _register_image_delete_routes(self, routes): + # A fixed path beside the dynamic /images/{image_id} routes: aiohttp + # (3.10+) looks fixed paths up first (see image_categories.py). + @routes.post("/prompt_manager/images/delete/preview") + async def preview_image_delete_route(request): + return await self.preview_image_delete(request) + + # ── review ──────────────────────────────────────────────────────── + + async def preview_image_delete(self, request): + """Review the selection under the maintenance preview lease.""" + try: + async with self._jobs.lease(MAINTENANCE_PREVIEW_LEASE): + return await self._preview_image_delete(request) + except JobBusy as busy: + return busy_response(busy) + + async def _preview_image_delete(self, request): + _data, targets, roots, error = await self._category_request( + request, frozenset() + ) + if error is not None: + return error + db = self.db + identity = self._database_identity(db) + try: + files = {} + for index in dict.fromkeys(t[0] for t in targets): + files[index] = await self._get_gallery_files(roots[index]) + review = await self._run_in_executor( + self._review_image_delete, db, roots, targets, files + ) + except Exception: + self.logger.exception("Image delete review failed") + return _failure("The review failed. Check server logs for details.", 500) + return await self._plan_response( + db, + identity, + IMAGE_DELETE, + _fingerprint(review["targets"]), + review["public"], + {"targets": review["targets"]}, + ) + + def _review_image_delete(self, db, roots, targets, files): + """Every target checked now (blocking); ``{targets, public}``.""" + listings, seen = {}, set() + acted, refused, trash_refused = [], [], [] + spellings = self._source_spellings() + root_texts = [str(root) for root in roots] + for index, _scope, relative in targets: + root = roots[index] + if index not in listings: + listings[index] = self._favorite_listing(root, files[index]) + spelled = listings[index].get(_favorite_relative_key(relative)) + target = { + "root_index": index, + "root_scope": _scope, + "root": str(root), + "root_label": self._public_path(root), + "rel": spelled or relative, + } + if spelled is None: + refused.append(_row(target, "not-listed")) + continue + absolute = os.path.join(str(root), *spelled.split("/")) + code, info = file_check(absolute, str(root), spelled) + if code: + refused.append(_row(target, code)) + continue + identity = (info.st_dev, info.st_ino) + if identity in seen: + continue # one physical file named twice (nested roots, hardlinks) + seen.add(identity) + target.update(self._removal_of(absolute, info, spellings)) + target["trash"] = trash.trash_available(absolute, root_texts) + if target["trash"]: + trash_refused.append(_row(target, target["trash"])) + acted.append(target) + counted = db.count_removal_records([t["removal"] for t in acted]) + library_count = len(acted) + public = { + "operation": IMAGE_DELETE, + "scope": {"count": library_count, "label": "images"}, + "library": {"count": library_count}, + "trash": {"count": library_count - len(trash_refused)}, + "protected": {"count": len(refused), "reasons": []}, + "refused": _summary(refused), + "trash_refused": _summary(trash_refused), + "records": counted["records"], + "orphaned_prompts": counted["orphaned_prompts"], + "media": {"deletes_files": False, "label": "Chosen in the review"}, + "details": [], + } + return {"targets": acted, "public": public} + + @staticmethod + def _removal_of(absolute, info, spellings): + favorites = [ + (_favorite_root_key(root), _favorite_relative_key(relative)) + for root, relative in spellings.favorites(absolute).items() + ] + return { + "abs": absolute, + "size": info.st_size, + "mtime_ns": info.st_mtime_ns, + "inode": [info.st_dev, info.st_ino], + "removal": { + "spellings": list(spellings.of(absolute).values()), + "locator": os.path.normcase(os.path.realpath(absolute)), + "favorites": favorites, + "category": spellings.category_key(absolute), + }, + } + + # ── apply ───────────────────────────────────────────────────────── + + def _start_image_delete(self, db, plan, mode, backup_on): + """Register the delete job for a reviewed plan (on the loop).""" + return self._jobs.start( + IMAGE_DELETE_JOB_KIND, + TITLES[mode], + admission="records", + cancellable=False, + scope={"operation": IMAGE_DELETE, "mode": mode}, + groups=IMAGE_DELETE_GROUP_NAMES, + runner=lambda job: self._run_image_delete(job, db, plan, mode, backup_on), + ) + + async def _run_image_delete(self, job, db, plan, mode, backup_on): + targets = plan.payload["targets"] + unchanged = await self._run_in_executor(self._sources_unchanged, targets) + if not unchanged: + return _fail_job(job, "sources-changed", SOURCES_CHANGED) + ok, backup = await self._safety_backup_step(job, db, backup_on) + if not ok: + return + try: + await self._delete_images(job, db, targets, mode, backup) + finally: + if backup: + await self._prune_backup_pool(job, db) + + async def _delete_images(self, job, db, targets, mode, backup): + job.update(total=len(targets), status_line=TITLES[mode]) + roots = [str(root) for root in self._get_all_output_dirs()] + done_group = "trashed" if mode == "trash" else "removed" + changed, failed = 0, False + for done, target in enumerate(targets, start=1): + if failed: + self._record_delete(job, "not_changed", target, "not-attempted") + continue + group, code, label = await self._run_in_executor( + self._delete_one_image, db, target, mode, roots + ) + if group == "errors": + failed = True + job.record( + "errors", + { + "path": target["rel"], + "root": target["root_label"], + "root_index": target["root_index"], + "root_scope": target["root_scope"], + "detail": RECORDS_FAILED, + }, + f"! {target['root_label']}/{target['rel']} · {RECORDS_FAILED}", + ) + elif group == done_group: + changed += 1 + detail = ( + f"Moved to the {label}" if label else "Removed from the library" + ) + self._record_delete(job, group, target, None, detail) + else: + self._record_delete(job, group, target, code) + job.update(processed=done, progress=int(done / len(targets) * 100)) + if changed: + self.invalidate_gallery_cache() + verb = "Moved" if mode == "trash" else "Removed" + where = "to the trash" if mode == "trash" else "from the library" + line = f"{verb} {changed:,} of {len(targets):,} images {where}" + job.log_line(line) + job.update(progress=100, status_line=f"{line} · {backup_note(backup)}") + if failed: + job.scope["failure"] = "records-failed" + job.publish({"type": "error", "code": "records-failed", "message": line}) + job.finish("failed") + + @staticmethod + def _record_delete(job, group, target, code, detail=None): + text = detail or REASONS[code] + row = { + "path": target["rel"], + "root": target["root_label"], + "root_index": target["root_index"], + "root_scope": target["root_scope"], + "code": code, + "detail": text, + } + mark = {"removed": "-", "trashed": "-"}.get(group, "=") + job.record( + group, row, f"{mark} {target['root_label']}/{target['rel']} · {text}" + ) + + def _delete_one_image(self, db, target, mode, roots): + """``(group, code, trash label)`` for one reviewed file (blocking).""" + code, _info = file_check( + target["abs"], + target["root"], + target["rel"], + expected={ + "size": target["size"], + "mtime_ns": target["mtime_ns"], + "inode": target["inode"], + }, + ) + if code: + return "not_changed", code, None + label = None + if mode == "trash": + try: + label = trash.move_to_trash(target["abs"], roots) + except trash.TrashRefused as refused: + code = refused.code if refused.code in REASONS else trash.FAILED + return "not_changed", code, None + mark = None + else: + mark = (target["size"], target["mtime_ns"]) + try: + db.remove_image_records(target["removal"], mark=mark) + except Exception: + self.logger.exception("Removing an image's records failed") + return "errors", None, label + _remove_thumbnail(Path(target["root"]), Path(target["rel"]), self.logger) + return ("trashed" if mode == "trash" else "removed"), None, label + + # ── the workspace listing ───────────────────────────────────────── + + def _hidden_entries(self, listings) -> set: + """Normalized paths of listed files a removal mark hides (blocking). + + Only marked locators are looked at, and only those files are + ``lstat``'ed: a mark hides a file while its size and ``mtime_ns`` are + still the ones it was removed with. + """ + marks = self.db.removal_marks() + if not marks: + return set() + located = self._workspace_locators(listings) + hidden = set() + for locator, (size, mtime_ns) in marks.items(): + entry = located.get(locator) + if entry is None: + continue + try: + info = os.lstat(entry.path) + except OSError: + continue + if (info.st_size, info.st_mtime_ns) == (size, mtime_ns): + hidden.add(entry.normalized) + return hidden + + def _removed_paths(self, paths) -> set: + """The absolute paths among *paths* a removal mark hides (blocking).""" + marks = self.db.removal_marks() + if not marks: + return set() + hidden = set() + for path in paths: + mark = marks.get(os.path.normcase(os.path.realpath(str(path)))) + if mark is None: + continue + try: + info = os.lstat(path) + except OSError: + continue + if (info.st_size, info.st_mtime_ns) == tuple(mark): + hidden.add(str(path)) + return hidden + + +def parse_mode(value): + """The apply's ``mode``, or None when it is not one of MODES.""" + return value if isinstance(value, str) and value in MODES else None diff --git a/py/api/image_tag_jobs.py b/py/api/image_tag_jobs.py index 5d1f9547..c16751b0 100644 --- a/py/api/image_tag_jobs.py +++ b/py/api/image_tag_jobs.py @@ -488,7 +488,11 @@ class ImageTagJobsMixin: _still_targets, self, index, root, files, seen ) ) - return found + # #480: a file removed from the library is never tagged by a batch. + hidden = await self._run_in_executor( + self._removed_paths, [target.path for target in found] + ) + return [target for target in found if target.path not in hidden] async def _start_autotag( self, *, model_type, existing, options, keep_in_memory, title, owner_kind diff --git a/py/api/images.py b/py/api/images.py index 01b27e3f..38286e2e 100644 --- a/py/api/images.py +++ b/py/api/images.py @@ -2450,6 +2450,10 @@ class ImageRoutesMixin: def key(entry): return _workspace_key(entry, order) + # #480: files removed from the library stay on disk, hidden here. + hidden = self._hidden_entries(listings) + if hidden: + entries = [entry for entry in entries if entry.normalized not in hidden] return ( entries, key, diff --git a/py/api/jobs.py b/py/api/jobs.py index de92fba8..52f016be 100644 --- a/py/api/jobs.py +++ b/py/api/jobs.py @@ -123,6 +123,8 @@ MODEL_DOWNLOAD_JOB_KIND = "model_download" DUPLICATES_JOB_KIND = "duplicates" DUPLICATE_GROUP_NAMES = ("groups_found",) DUPLICATE_CLEANUP_GROUP_NAMES = ("removed", "kept", "errors") +IMAGE_DELETE_JOB_KIND = "image_delete" # #480 +IMAGE_DELETE_GROUP_NAMES = ("trashed", "removed", "not_changed", "errors") # The System health check (py/api/system.py): a read-only ``read`` job with # no file outcomes, so it may run beside a scan. HEALTH_JOB_KIND = "health" diff --git a/py/api/maintenance.py b/py/api/maintenance.py index dd103b85..4a518efd 100644 --- a/py/api/maintenance.py +++ b/py/api/maintenance.py @@ -488,7 +488,9 @@ class MaintenanceRoutesMixin: # ── apply ───────────────────────────────────────────────────────── async def apply_maintenance_plan(self, request): - data, error = await self._read_maintenance_body(request, ("token",)) + from .image_delete import IMAGE_DELETE, parse_mode + + data, error = await self._read_maintenance_body(request, ("token", "mode")) if error is not None: return error token = data.get("token") @@ -497,6 +499,13 @@ class MaintenanceRoutesMixin: plan, refused = self._peek_plan(token) if refused is not None: return refused + mode = None + if plan.operation == IMAGE_DELETE: + mode = parse_mode(data.get("mode")) + if mode is None: + return _invalid('mode must be "library" or "trash"') + elif "mode" in data: + return _invalid("Unknown field: mode") store = self._maintenance_plans if plan.operation == ROTATED_LOGS: # Log files only: no database identity, backup or disk check. @@ -544,6 +553,8 @@ class MaintenanceRoutesMixin: try: if plan.operation == DUPLICATE_FILES: job = self._start_duplicate_cleanup(db, plan, backup) + elif plan.operation == IMAGE_DELETE: + job = self._start_image_delete(db, plan, mode, backup) elif plan.operation == WORKFLOW_STORAGE: job = self._start_workflow_storage(db, backup) else: diff --git a/tests/fake_trash.py b/tests/fake_trash.py new file mode 100644 index 00000000..0420f92a --- /dev/null +++ b/tests/fake_trash.py @@ -0,0 +1,52 @@ +"""A fake system trash for tests (#480): renames into a folder outside every root. + +``install(folder)`` replaces ``utils.trash.BACKEND`` and returns the fake; +``uninstall`` puts the real one back. ``refuse[path] = code`` refuses one +file with that code. The fake never copies or unlinks; it records each call. +The disposable browser server installs it when ``PM_TEST_FAKE_TRASH=1``, so +no test touches a real trash. +""" + +import os + +from utils import trash + + +class FakeTrash: + def __init__(self, folder): + self.folder = str(folder) + os.makedirs(self.folder, exist_ok=True) + self.refuse = {} + self.moved = [] # (source, destination) + self.calls = [] + + def available(self, path, roots): + self.calls.append(("available", str(path))) + return self.refuse.get(str(path)) + + def move(self, path, roots): + self.calls.append(("move", str(path))) + code = self.refuse.get(str(path)) + if code: + raise trash.TrashRefused(code) + destination = os.path.join( + self.folder, f"{len(self.moved)}-{os.path.basename(path)}" + ) + if os.path.lexists(destination): + raise trash.TrashRefused(trash.FAILED) + os.rename(path, destination) + self.moved.append((str(path), destination)) + return "fake trash" + + +_ORIGINAL = trash.BACKEND + + +def install(folder): + fake = FakeTrash(folder) + trash.BACKEND = fake + return fake + + +def uninstall(): + trash.BACKEND = _ORIGINAL diff --git a/tests/test_image_delete_routes.py b/tests/test_image_delete_routes.py new file mode 100644 index 00000000..d88dc0b8 --- /dev/null +++ b/tests/test_image_delete_routes.py @@ -0,0 +1,281 @@ +"""Delete selected images: the review and the apply job (#480). + +``POST /prompt_manager/images/delete/preview`` reviews both modes; +``POST /prompt_manager/maintenance/apply {token, mode}`` removes the files' +records (library mode, with a removal mark) or moves the files to the trash +first (trash mode, the fake trash here). No path unlinks a media file. +""" + +import asyncio +import os +import sqlite3 +from pathlib import Path +from unittest.mock import patch + +try: + from tests import fake_trash + from tests.disk_waits import DISK_BOUND_SECONDS + from tests.test_api_duplicate_files import DuplicateFilesTestCase +except ImportError: # discovered with tests/ as the top-level directory + import fake_trash + from disk_waits import DISK_BOUND_SECONDS + from test_api_duplicate_files import DuplicateFilesTestCase + +from py.api import image_delete +from py.api.image_delete import file_check, lexical_problem + +PREVIEW = "/prompt_manager/images/delete/preview" +APPLY = "/prompt_manager/maintenance/apply" +LISTING = "/prompt_manager/images/output" + + +class ImageDeleteTestCase(DuplicateFilesTestCase): + async def get_application(self): + app = await super().get_application() + self.fake = fake_trash.install(Path(self.tmpdir) / "fake-trash") + self.addCleanup(fake_trash.uninstall) + return app + + def target(self, path, index=0): + roots = self.api._get_all_output_dirs() + return { + "root_index": index, + "root_scope": self.api._gallery_root_scope(roots[index], index), + "path": path, + } + + async def review(self, paths, status=200): + items = [p if isinstance(p, dict) else self.target(p) for p in paths] + return await self.post(PREVIEW, {"items": items}, status) + + async def delete(self, paths, mode): + plan = await self.review(paths) + body = await self.post(APPLY, {"token": plan["token"], "mode": mode}, 202) + job = self.api._jobs.get(body["job"]["id"]) + await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS) + return plan, job + + async def listed(self): + response = await self.client.get(LISTING, params={"mode": "workspace"}) + page = await response.json() + return [image["path"] for image in page["images"]] + + def link(self, name, text="a prompt"): + prompt_id = self.save(text) + return self.api.db.link_image_to_prompt( + str(prompt_id), str(self.output_dir / name) + ) + + def count(self, sql): + conn = sqlite3.connect(self.api.db.model.db_path) + try: + return conn.execute(sql).fetchone()[0] + finally: + conn.close() + + @staticmethod + def details(job, group): + return [row["detail"] for row in job.detail()["outcomes"][group]] + + +class TestReview(ImageDeleteTestCase): + async def test_review_counts_records_and_refuses_what_it_cannot_delete(self): + self.write("a.png", b"aaa", mtime=1_000_000) + self.write("b.png", b"bbb", mtime=2_000_000) + self.link("a.png", "only a") + self.fake.refuse[str(self.output_dir / "b.png")] = "no-trash" + plan = await self.review(["a.png", "b.png", "missing.png"]) + self.assertEqual(plan["library"], {"count": 2}) + self.assertEqual(plan["trash"], {"count": 1}) + self.assertEqual(plan["refused"]["counts"], {"not-listed": 1}) + self.assertEqual(plan["trash_refused"]["counts"], {"no-trash": 1}) + self.assertEqual(plan["records"]["prompt_links"], 1) + self.assertEqual(plan["orphaned_prompts"]["count"], 1) + self.assertIn(plan["backup"]["policy"], ("on", "off")) + self.assertNotIn(self.tmpdir, str(plan)) + # The review wrote nothing. + self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 1) + self.assertEqual(self.fake.moved, []) + + async def test_body_shapes_and_the_cap(self): + await self.post(PREVIEW, {}, 400) + await self.post(PREVIEW, {"items": [], "mode": "trash"}, 400) + too_many = [self.target(f"{n}.png") for n in range(501)] + await self.post(PREVIEW, {"items": too_many}, 400) + + async def test_mode_is_required_and_only_for_this_plan(self): + self.write("a.png", b"aaa") + plan = await self.review(["a.png"]) + await self.post(APPLY, {"token": plan["token"]}, 400) + await self.post(APPLY, {"token": plan["token"], "mode": "delete"}, 400) + self.assertTrue((self.output_dir / "a.png").exists()) + + +class TestLibraryMode(ImageDeleteTestCase): + async def test_records_go_and_the_file_stays_hidden(self): + self.write("a.png", b"aaa", mtime=1_000_000) + self.write("b.png", b"bbb", mtime=2_000_000) + self.link("a.png") + self.assertEqual(sorted(await self.listed()), ["a.png", "b.png"]) + before = os.stat(self.output_dir / "a.png") + _plan, job = await self.delete(["a.png"], "library") + self.assertEqual(job.state, "finished", job.summary()) + self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 0) + after = os.stat(self.output_dir / "a.png") + self.assertEqual( + (before.st_size, before.st_mtime_ns), (after.st_size, after.st_mtime_ns) + ) + self.assertEqual(await self.listed(), ["b.png"]) + # A later scan or link never brings it back. + self.assertEqual(self.link("a.png", "rescanned"), 0) + self.assertEqual( + self.api._removed_paths([self.output_dir / "a.png"]), + {str(self.output_dir / "a.png")}, + ) + self.assertEqual(self.fake.moved, []) + + async def test_a_modified_file_shows_again(self): + self.write("a.png", b"aaa", mtime=1_000_000) + await self.delete(["a.png"], "library") + self.assertEqual(await self.listed(), []) + self.write("a.png", b"new and longer", mtime=3_000_000) + self.api.invalidate_gallery_cache() + self.assertEqual(await self.listed(), ["a.png"]) + + async def test_a_safety_backup_is_taken_first(self): + self.write("a.png", b"aaa") + with patch.object(self.api, "_safety_backup_policy", return_value="on"): + await self.delete(["a.png"], "library") + self.assertEqual(len(self.backups()), 1) + + +class TestTrashMode(ImageDeleteTestCase): + async def test_the_file_moves_first_then_its_records_go(self): + self.write("a.png", b"aaa") + self.link("a.png") + seen = {} + original = self.fake.move + + def move(path, roots): + # The records still exist while the file moves. + seen["links"] = self.count("SELECT COUNT(*) FROM generated_images") + return original(path, roots) + + self.fake.move = move + _plan, job = await self.delete(["a.png"], "trash") + self.assertEqual(job.state, "finished", job.summary()) + self.assertEqual(seen["links"], 1) + self.assertFalse((self.output_dir / "a.png").exists()) + [(source, destination)] = self.fake.moved + self.assertTrue(os.path.exists(destination)) + self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 0) + self.assertEqual(self.api.db.removal_marks(), {}) + + async def test_a_refused_move_changes_nothing(self): + self.write("a.png", b"aaa") + self.link("a.png") + plan = await self.review(["a.png"]) + self.fake.refuse[str(self.output_dir / "a.png")] = "locked" + body = await self.post(APPLY, {"token": plan["token"], "mode": "trash"}, 202) + job = self.api._jobs.get(body["job"]["id"]) + await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS) + self.assertTrue((self.output_dir / "a.png").exists()) + self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 1) + self.assertEqual(self.api.db.removal_marks(), {}) + self.assertEqual(job.counters["not_changed"], 1) + self.assertEqual( + self.details(job, "not_changed"), [image_delete.REASONS["locked"]] + ) + + async def test_a_file_changed_after_the_review_is_not_changed(self): + self.write("a.png", b"aaa", mtime=1_000_000) + plan = await self.review(["a.png"]) + self.write("a.png", b"changed bytes", mtime=2_000_000) + body = await self.post(APPLY, {"token": plan["token"], "mode": "trash"}, 202) + job = self.api._jobs.get(body["job"]["id"]) + await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS) + self.assertTrue((self.output_dir / "a.png").exists()) + self.assertEqual(self.fake.moved, []) + self.assertEqual( + self.details(job, "not_changed"), [image_delete.REASONS["changed"]] + ) + + async def test_a_records_failure_stops_the_job(self): + self.write("a.png", b"aaa", mtime=1_000_000) + self.write("b.png", b"bbb", mtime=2_000_000) + with patch.object( + type(self.api.db), + "remove_image_records", + side_effect=sqlite3.OperationalError("locked"), + ): + _plan, job = await self.delete(["a.png", "b.png"], "library") + self.assertEqual(job.state, "failed") + self.assertEqual(job.scope.get("failure"), "records-failed") + self.assertEqual(job.counters["errors"], 1) + self.assertEqual( + self.details(job, "not_changed"), [image_delete.REASONS["not-attempted"]] + ) + + async def test_nothing_is_ever_unlinked(self): + self.write("a.png", b"aaa") + self.write("b.png", b"bbb") + media = str(self.output_dir) + real_remove, real_unlink = os.remove, os.unlink + + def guard(real): + def call(path, *args, **kwargs): + text = os.path.realpath(str(path)) + if ( + text.startswith(os.path.realpath(media)) + and "thumbnails" not in text + ): + raise AssertionError(f"unlinked a media file: {path}") + return real(path, *args, **kwargs) + + return call + + with ( + patch("os.remove", guard(real_remove)), + patch("os.unlink", guard(real_unlink)), + ): + await self.delete(["a.png"], "trash") + await self.delete(["b.png"], "library") + self.assertTrue((self.output_dir / "b.png").exists()) + + +class TestFileCheck(ImageDeleteTestCase): + def test_lexical_codes(self): + self.assertEqual(lexical_problem("thumbnails/a.png"), "thumbnail") + self.assertEqual(lexical_problem(".Trash-1000/files/a.png"), "trash-folder") + self.assertEqual(lexical_problem("x/$RECYCLE.BIN/a.png"), "trash-folder") + self.assertEqual(lexical_problem("a.txt"), "not-media") + self.assertIsNone(lexical_problem(".Trashy/a.png")) + self.assertIsNone(lexical_problem("sub/a.png")) + + def test_links_and_folders_leading_outside_are_refused(self): + root = str(self.output_dir) + outside = Path(self.tmpdir) / "elsewhere" + outside.mkdir() + (outside / "x.png").write_bytes(b"x") + os.symlink(outside / "x.png", self.output_dir / "link.png") + os.symlink(outside, self.output_dir / "linked") + self.assertEqual( + file_check(str(self.output_dir / "link.png"), root, "link.png")[0], "link" + ) + self.assertEqual( + file_check(str(self.output_dir / "linked/x.png"), root, "linked/x.png")[0], + "outside", + ) + self.assertEqual( + file_check(str(self.output_dir / "gone.png"), root, "gone.png")[0], "gone" + ) + self.write("ok.png", b"ok") + self.assertIsNone( + file_check(str(self.output_dir / "ok.png"), root, "ok.png")[0] + ) + + def test_reasons_cover_every_code(self): + from utils import trash + + for code in trash.CODES: + self.assertIn(code, image_delete.REASONS)