fix,refac: Properly clean up thumbnails

- Fix: Cleanup of thumbnails upon interruption (by the user or an
  exception).
- Add: `.__main__.TEMP_DIR`.
- Change: Create a session-specific temporary data directory and clean
  it up at the topmost execution level.
- Change: Disable thumbnailing if unable to create the temporary data
  directory.
- Change: Refactor `generate_grid_thumbnails()` and
  `manage_grid_thumbnails()` in `.tui.render` accordingly.

Complements #13, #16.

src/termvisage/__main__.py

@@ -22,6 +22,19 @@ def main() -> int:
 
     from . import cli, logging, notify, tui
 
+    def cleanup_temp_dir():
+        if not TEMP_DIR:
+            return
+
+        from shutil import rmtree
+
+        try:
+            rmtree(TEMP_DIR)
+        except OSError:
+            logging.log_exception(
+                f"Failed to delete the temporary data directory {TEMP_DIR!r}", logger
+            )
+
     def finish_loading():
         if notify.loading_initialized:
             notify.end_loading()  # End the current phase (may be CLI or TUI)
@@ -58,6 +71,7 @@ def finish_multi_logging():
         cli.interrupted.set()  # Signal interruption to subprocesses and other threads.
         finish_loading()
         finish_multi_logging()
+        cleanup_temp_dir()
         logging.log(
             "Session interrupted",
             logger,
@@ -76,6 +90,7 @@ def finish_multi_logging():
         cli.interrupted.set()  # Signal interruption to subprocesses and other threads.
         finish_loading()
         finish_multi_logging()
+        cleanup_temp_dir()
         if logging.initialized:
             logger.exception("Session terminated due to:")
         logging.log(
@@ -91,6 +106,7 @@ def finish_multi_logging():
     else:
         finish_loading()
         finish_multi_logging()
+        cleanup_temp_dir()
         logger.info(f"Session ended with return-code {exit_code} ({codes[exit_code]})")
         return exit_code
     finally:
@@ -101,5 +117,9 @@ def finish_multi_logging():
             image_w._ti_image.close()
 
 
+# Session-specific temporary data directory.
+# Updated from `.cli.main()`.
+TEMP_DIR: str | None = None
+
 if __name__ == "__main__":
     sys.exit(main())

src/termvisage/cli.py

@@ -13,6 +13,7 @@
 from operator import mul, setitem
 from os.path import abspath, basename, exists, isdir, isfile, islink, realpath
 from queue import Empty, Queue
+from tempfile import mkdtemp
 from threading import Event, current_thread
 from time import sleep
 from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Union
@@ -651,7 +652,7 @@ def main() -> None:
     """CLI execution sub-entry-point"""
     # Importing these (in isort order) at module-level results in circular imports
     # # Prevents circular import for program execution
-    from . import config
+    from . import __main__, config
 
     # # Prevents circular import for docs `autoprogram` (isort order or not)
     from .parsers import parser, style_parsers
@@ -719,6 +720,13 @@ def main() -> None:
             )
             setattr(args, var_name, option.value)
 
+    try:
+        __main__.TEMP_DIR = mkdtemp(prefix="termvisage-")
+    except OSError:
+        logging.log_exception("Failed to create the temporary data directory", logger)
+    else:
+        logger.debug(f"Created the temporary data directory {__main__.TEMP_DIR!r}")
+
     set_query_timeout(args.query_timeout)
     if args.swap_win_size:
         enable_win_size_swap()

src/termvisage/tui/__init__.py

@@ -30,6 +30,7 @@ def init(
     ImageClass: type,
 ) -> None:
     """Initializes the TUI"""
+    from ..__main__ import TEMP_DIR
     from . import keys
 
     global active, initialized
@@ -44,7 +45,7 @@ def init(
     main.NO_ANIMATION = args.no_anim
     main.RECURSIVE = args.recursive
     main.SHOW_HIDDEN = args.all
-    main.THUMBNAIL = args.thumbnail
+    main.THUMBNAIL = args.thumbnail and TEMP_DIR
     main.THUMBNAIL_SIZE_PRODUCT = config_options.thumbnail_size**2
     main.ImageClass = ImageClass
     main.loop = Loop(

src/termvisage/tui/render.py

@@ -7,7 +7,6 @@
 from multiprocessing import Event as mp_Event, Lock as mp_Lock, Queue as mp_Queue
 from os import remove
 from queue import Empty, Queue
-from tempfile import mkdtemp, mkstemp
 from threading import Event, Lock
 from typing import Union
 
@@ -74,14 +73,17 @@ def generate_grid_thumbnails(
     thumbnail_size: int,
     not_generating: Event | mp_Event,
     deduplication_lock: Lock | mp_Lock,
+    temp_dir: str,
 ) -> None:
     from glob import iglob
     from os import fdopen, mkdir, scandir
-    from shutil import copyfile, rmtree
+    from shutil import copyfile
     from sys import hash_info
+    from tempfile import mkstemp
 
     from PIL.Image import Resampling, open as Image_open
 
+    THUMBNAIL_DIR = temp_dir + "/thumbnails"
     THUMBNAIL_FRAME_SIZE = (thumbnail_size,) * 2
     BOX = Resampling.BOX
     THUMBNAIL_MODES = {"RGB", "RGBA"}
@@ -94,122 +96,104 @@ def generate_grid_thumbnails(
     deduplicated_to_be_deleted: set[str] = set()
 
     try:
-        THUMBNAIL_DIR = (TEMP_DIR := mkdtemp(prefix="termvisage-")) + "/thumbnails"
         mkdir(THUMBNAIL_DIR)
     except OSError:
-        logging.log_exception(
-            "Failed to create thumbnail directory", logger, fatal=True
-        )
+        logging.log_exception("Failed to create the thumbnail directory", logger)
         raise
+    logger.debug(f"Created the thumbnail directory {THUMBNAIL_DIR!r}")
 
-    logging.log(
-        f"Created thumbnail directory {THUMBNAIL_DIR!r}",
-        logger,
-        _logging.DEBUG,
-        direct=False,
-    )
-
-    try:
-        while True:
-            not_generating.set()
-            try:
-                if not (source := input.get()):
-                    break  # Quitting
-            finally:
-                not_generating.clear()
-
-            if deduplicated_to_be_deleted:
-                # Retain only the files that still exist;
-                # remove files that have been deleted.
-                deduplicated_to_be_deleted.intersection_update(
-                    entry.path for entry in scandir(THUMBNAIL_DIR)
-                )
+    while True:
+        not_generating.set()
+        try:
+            if not (source := input.get()):
+                break  # Quitting
+        finally:
+            not_generating.clear()
+
+        if deduplicated_to_be_deleted:
+            # Retain only the files that still exist;
+            # remove files that have been deleted.
+            deduplicated_to_be_deleted.intersection_update(
+                entry.path for entry in scandir(THUMBNAIL_DIR)
+            )
 
-            # Make source image into a thumbnail
-            try:
-                img = Image_open(source)
-                has_transparency = img.has_transparency_data
-                if img.mode not in THUMBNAIL_MODES:
-                    with img:
-                        img = img.convert("RGBA" if has_transparency else "RGB")
-                img.thumbnail(THUMBNAIL_FRAME_SIZE, BOX)
-            except Exception:
-                output.put((source, None, None))
-                logging.log_exception(
-                    f"Failed to generate thumbnail for {source!r}", logger
-                )
-                continue
+        # Make source image into a thumbnail
+        try:
+            img = Image_open(source)
+            has_transparency = img.has_transparency_data
+            if img.mode not in THUMBNAIL_MODES:
+                with img:
+                    img = img.convert("RGBA" if has_transparency else "RGB")
+            img.thumbnail(THUMBNAIL_FRAME_SIZE, BOX)
+        except Exception:
+            output.put((source, None, None))
+            logging.log_exception(
+                f"Failed to generate thumbnail for {source!r}", logger
+            )
+            continue
 
-            img_bytes = img.tobytes()
-            # The hash is interpreted as an unsigned integer, represented in hex and
-            # zero-extended to fill up the platform-specific hash integer width.
-            img_hash = f"{hash(img_bytes) & UINT_HASH_WIDTH_MAX:0{HEX_HASH_WIDTH}x}"
+        img_bytes = img.tobytes()
+        # The hash is interpreted as an unsigned integer, represented in hex and
+        # zero-extended to fill up the platform-specific hash integer width.
+        img_hash = f"{hash(img_bytes) & UINT_HASH_WIDTH_MAX:0{HEX_HASH_WIDTH}x}"
 
-            # Create thumbnail file
-            try:
-                thumbnail_fd, thumbnail = mkstemp("", f"{img_hash}-", THUMBNAIL_DIR)
-            except Exception:
-                output.put((source, None, None))
-                logging.log_exception(
-                    f"Failed to create thumbnail file for {source!r}", logger
-                )
-                del img_bytes  # Possibly relatively large
-                img.close()
-                continue
+        # Create thumbnail file
+        try:
+            thumbnail_fd, thumbnail = mkstemp("", f"{img_hash}-", THUMBNAIL_DIR)
+        except Exception:
+            output.put((source, None, None))
+            logging.log_exception(
+                f"Failed to create thumbnail file for {source!r}", logger
+            )
+            del img_bytes  # Possibly relatively large
+            img.close()
+            continue
 
-            # Deduplication
-            deduplicated = None
-            with deduplication_lock:
-                for other_thumbnail in iglob(
-                    f"{THUMBNAIL_DIR}/{img_hash}-*", root_dir=THUMBNAIL_DIR
+        # Deduplication
+        deduplicated = None
+        with deduplication_lock:
+            for other_thumbnail in iglob(
+                f"{THUMBNAIL_DIR}/{img_hash}-*", root_dir=THUMBNAIL_DIR
+            ):
+                if (
+                    other_thumbnail == thumbnail
+                    or other_thumbnail in deduplicated_to_be_deleted
                 ):
-                    if (
-                        other_thumbnail == thumbnail
-                        or other_thumbnail in deduplicated_to_be_deleted
-                    ):
-                        continue
+                    continue
 
-                    with Image_open(other_thumbnail) as other_img:
-                        if other_img.tobytes() != img_bytes:
-                            continue
+                with Image_open(other_thumbnail) as other_img:
+                    if other_img.tobytes() != img_bytes:
+                        continue
 
-                    try:
-                        copyfile(other_thumbnail, thumbnail)
-                    except Exception:
-                        logging.log_exception(
-                            f"Failed to deduplicate {other_thumbnail!r} for "
-                            "{source!r}",
-                            logger,
-                        )
-                    else:
-                        deduplicated_to_be_deleted.add(deduplicated := other_thumbnail)
+                try:
+                    copyfile(other_thumbnail, thumbnail)
+                except Exception:
+                    logging.log_exception(
+                        f"Failed to deduplicate {other_thumbnail!r} for {source!r}",
+                        logger,
+                    )
+                else:
+                    deduplicated_to_be_deleted.add(deduplicated := other_thumbnail)
 
-                    break
+                break
 
-            del img_bytes  # Possibly relatively large
+        del img_bytes  # Possibly relatively large
 
-            # Save thumbnail, if deduplication didn't work out
-            if not deduplicated:
-                with img, fdopen(thumbnail_fd, "wb") as thumbnail_file:
-                    try:
-                        img.save(thumbnail_file, "PNG")
-                    except Exception:
-                        output.put((source, None, None))
-                        thumbnail_file.close()  # Close before deleting the file
-                        delete_thumbnail(thumbnail)
-                        logging.log_exception(
-                            f"Failed to save thumbnail for {source!r}", logger
-                        )
-                        continue
+        # Save thumbnail, if deduplication didn't work out
+        if not deduplicated:
+            with img, fdopen(thumbnail_fd, "wb") as thumbnail_file:
+                try:
+                    img.save(thumbnail_file, "PNG")
+                except Exception:
+                    output.put((source, None, None))
+                    thumbnail_file.close()  # Close before deleting the file
+                    delete_thumbnail(thumbnail)
+                    logging.log_exception(
+                        f"Failed to save thumbnail for {source!r}", logger
+                    )
+                    continue
 
-            output.put((source, thumbnail, deduplicated))
-    finally:
-        try:
-            rmtree(TEMP_DIR, ignore_errors=True)
-        except OSError:
-            logging.log_exception(
-                f"Failed to delete thumbnail directory {THUMBNAIL_DIR!r}", logger
-            )
+        output.put((source, thumbnail, deduplicated))
 
     clear_queue(output)
 
@@ -576,6 +560,7 @@ def mark_thumbnail_rendered(source: str, thumbnail: str) -> None:
 
 
 def manage_grid_thumbnails(thumbnail_size: int) -> None:
+    from ..__main__ import TEMP_DIR
     from .main import grid_active, quitting
 
     # NOTE:
@@ -657,6 +642,7 @@ def cache_thumbnail(source: str, thumbnail: str, deduplicated: str | None) -> No
             thumbnail_size,
             not_generating,
             deduplication_lock,
+            TEMP_DIR,
         ),
         name="GridThumbnailer",
         redirect_notifs=True,