From dc8b86751e2afc37da30b7dbefef3bb8fd27373e Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sat, 24 Oct 2026 12:41:06 -0700 Subject: [PATCH] feat(thumbnails): thumbnail format, quality and size as server settings Thumbnail format (webp default, jxl, jpeg), quality (1-100, default 80) and size limit (128-2048 px, default 512) are server settings in the config.json thumbnails section, saved with the settings revision. The never-read gallery thumbnail_size is gone: older saves wrote its unused default 256 into config.json. One module names previews (_thumb.webp/.jxl/.jpg) and finds the preview a source owns, preferring the configured format; legacy names are still found. The encoder module writes every preview atomically and verifies it decodes. JPEG XL needs no new dependency: Pillow (natively or with pillow_jxl), ffmpeg with libjxl, or cjxl; without one the settings refuse jxl (400 jxl-unavailable) and GET says why. Encoder processes are tracked, bounded and killed at shutdown. --- py/api/settings_routes.py | 37 +++ py/api/thumbnail_encode.py | 403 +++++++++++++++++++++++++++++++ py/api/thumbnail_settings.py | 186 ++++++++++++++ py/config.py | 37 ++- tests/test_api_admin.py | 97 ++++++++ tests/test_config.py | 3 +- tests/test_thumbnail_encode.py | 327 +++++++++++++++++++++++++ tests/test_thumbnail_settings.py | 214 ++++++++++++++++ 8 files changed, 1298 insertions(+), 6 deletions(-) create mode 100644 py/api/thumbnail_encode.py create mode 100644 py/api/thumbnail_settings.py create mode 100644 tests/test_thumbnail_encode.py create mode 100644 tests/test_thumbnail_settings.py diff --git a/py/api/settings_routes.py b/py/api/settings_routes.py index 17637f7e..97814f07 100644 --- a/py/api/settings_routes.py +++ b/py/api/settings_routes.py @@ -23,6 +23,8 @@ from ..config_store import ( get_config_store, ) from . import sources +from . import thumbnail_encode +from . import thumbnail_settings from .admin import _current_image_monitor from .job_routes import busy_response from .jobs import JobBusy @@ -41,6 +43,9 @@ _SETTINGS_FILE_KEYS = { "WORKER_THREADS": ("performance", "worker_threads"), "MONITORING_ENABLED": ("gallery", "monitoring", "enabled"), "SAFETY_BACKUP": ("maintenance", "safety_backup"), + "THUMBNAIL_FORMAT": ("thumbnails", "format"), + "THUMBNAIL_QUALITY": ("thumbnails", "quality"), + "THUMBNAIL_EDGE": ("thumbnails", "edge"), } # Attributes that live on GalleryConfig; the others on PromptManagerConfig. _GALLERY_SETTINGS = frozenset({"MONITORING_ENABLED"}) @@ -176,12 +181,15 @@ class SettingsRoutesMixin: ] revision = await self._run_in_executor(self._config_revision) source_status, monitor = await self._run_in_executor(self._source_status) + # #317: which thumbnail formats can be written (a cached probe). + encoders = await self._run_in_executor(thumbnail_encode.encoder_status) return web.json_response( { "success": True, **revision, "sources": source_status, "monitor": monitor, + "thumbnail_encoders": encoders, "settings": { "monitoring_enabled": GalleryConfig.MONITORING_ENABLED, "result_timeout": PromptManagerConfig.RESULT_TIMEOUT, @@ -189,6 +197,9 @@ class SettingsRoutesMixin: "infinite_scroll": PromptManagerConfig.INFINITE_SCROLL, "worker_threads": PromptManagerConfig.WORKER_THREADS, "safety_backup": PromptManagerConfig.SAFETY_BACKUP, + "thumbnail_format": PromptManagerConfig.THUMBNAIL_FORMAT, + "thumbnail_quality": PromptManagerConfig.THUMBNAIL_QUALITY, + "thumbnail_edge": PromptManagerConfig.THUMBNAIL_EDGE, "cpu_count": PromptManagerConfig.max_worker_threads(), "gallery_root_paths": root_paths, "gallery_root_path": root_paths[0] if root_paths else "", @@ -295,6 +306,20 @@ class SettingsRoutesMixin: return web.json_response( {"success": False, "error": str(ve)}, status=400 ) + if updates.get("THUMBNAIL_FORMAT") == "jxl": + # #317: JPEG XL only with an encoder (a cached probe). + encoder = await self._run_in_executor( + thumbnail_encode.detect_jxl_encoder + ) + if encoder is None: + return web.json_response( + { + "success": False, + "code": "jxl-unavailable", + "error": thumbnail_encode.JXL_UNAVAILABLE, + }, + status=400, + ) # Off the loop: an accepted root is resolved (filesystem access). new_roots, error = await self._run_in_executor( @@ -491,6 +516,18 @@ class SettingsRoutesMixin: updates["WORKER_THREADS"] = validate_worker_threads( data["worker_threads"], PromptManagerConfig.max_worker_threads() ) + if "thumbnail_format" in data: + updates["THUMBNAIL_FORMAT"] = thumbnail_settings.validate_format( + data["thumbnail_format"] + ) + if "thumbnail_quality" in data: + updates["THUMBNAIL_QUALITY"] = thumbnail_settings.validate_quality( + data["thumbnail_quality"] + ) + if "thumbnail_edge" in data: + updates["THUMBNAIL_EDGE"] = thumbnail_settings.validate_edge( + data["thumbnail_edge"] + ) return updates def _parse_gallery_roots(self, data): diff --git a/py/api/thumbnail_encode.py b/py/api/thumbnail_encode.py new file mode 100644 index 00000000..960f763e --- /dev/null +++ b/py/api/thumbnail_encode.py @@ -0,0 +1,403 @@ +"""Thumbnail encoders (#317): every preview file is written here. + +``write_thumbnail(frame, dst, settings)`` encodes a prepared frame +(``thumbnail_frame``: the first frame, RGB, fitted inside the edge) in the +configured format into a hidden temp sibling, closes it, verifies that it +decodes, and renames it over ``dst`` (``os.replace``), so a reader never sees +a partial file and a failure keeps the old preview. ``guard()`` is asked just +before the rename (automatic previews' holds, #249). + +Formats: WebP (Pillow, ``method=4``) and JPEG (Pillow, ``optimize``) always +work. JPEG XL needs no new dependency; the first encoder found is used: + +1. Pillow, natively or through the ``pillow_jxl`` plugin (imported when + installed: ``pip install pillow-jxl-plugin``), given the quality itself; +2. ffmpeg built with ``libjxl`` (``-distance``), fed a PNG on stdin; +3. ``cjxl`` from libjxl (``-d``), fed a temp PNG that is closed first. + +Quality maps to a JPEG XL distance with libjxl's own +``JxlEncoderDistanceFromQuality``: ``0.1 + (100 - q) * 0.09`` from 30 up +(q80 → d1.9, q85 → d1.45, q90 → d1.0), the quadratic below 30, and q100 → +d0.1 (a preview is never lossless). Pillow's plugin applies the same mapping +inside libjxl. Detection is cached (``detect_jxl_encoder(refresh=True)`` +probes again). If JPEG XL is configured but no encoder is left at write +time, that file is written as WebP and the fallback is logged once. + +Encoder processes (and ffmpeg's video frames, images.py) go through +``run_encoder``: tracked, bounded by a timeout, and killed by +``shutdown_encoders()`` at the app's cleanup, after which none starts. +""" + +import importlib +import io +import os +import shutil +import subprocess # nosec B404 - fixed argv lists, never a shell +import threading +from pathlib import Path + +from PIL import Image + +try: + from ...utils.logging_config import get_logger +except ImportError: # tests import the package top-level + from utils.logging_config import get_logger + +logger = get_logger("prompt_manager.thumbnails") + +PILLOW_JXL_FORMAT = "JXL" +JXL_UNAVAILABLE = ( + "No JPEG XL encoder found. Install pillow-jxl-plugin " + "(pip install pillow-jxl-plugin), ffmpeg built with libjxl, or cjxl from " + "libjxl, then restart ComfyUI." +) +# Seconds an encoder process may run (ffmpeg's video frame shares it). +ENCODE_TIMEOUT = 30 +PROBE_TIMEOUT = 10 +JXL_EFFORT = 7 +WEBP_METHOD = 4 +JXL_SIGNATURES = (b"\xff\x0a", b"\x00\x00\x00\x0cJXL \r\n\x87\n") +_NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0) + + +class PreviewWriteHeld(Exception): + """A preview write stopped before its rename: ``guard()`` said no (#249).""" + + +# The result error of a write held before its rename (never remembered). +PREVIEW_HELD = "Preview writes are held while another job runs" + + +class ThumbnailVerifyError(RuntimeError): + """The encoded preview did not decode; it was never renamed in.""" + + +class EncoderFailed(RuntimeError): + """An encoder process failed or wrote nothing.""" + + +class EncoderClosed(RuntimeError): + """The app is shutting down: no encoder process runs any more.""" + + +# ── temp files ─────────────────────────────────────────────────────── + + +def temp_sibling(dst, tag="part"): + """Hidden temp path next to *dst* that keeps its extension.""" + dst = Path(dst) + token = f"{os.getpid()}-{threading.get_ident()}" + return dst.with_name(f".{dst.stem}.{tag}-{token}{dst.suffix}") + + +def replace_unless_held(tmp, dst, guard=None): + """Rename *tmp* over *dst*, unless ``guard()`` is false (PreviewWriteHeld).""" + if guard is not None and not guard(): + raise PreviewWriteHeld(PREVIEW_HELD) + os.replace(tmp, dst) + + +def _unlink_quietly(path): + try: + Path(path).unlink() + except FileNotFoundError: + pass + except OSError as exc: + logger.debug(f"Could not remove temp file {Path(path).name}: {exc}") + + +# ── encoder processes ──────────────────────────────────────────────── + +_process_lock = threading.Lock() +_processes = set() +_closed = False + + +def run_encoder(cmd, *, input=None, timeout=ENCODE_TIMEOUT, track=None): + """Run *cmd* (a fixed argv) with *input* on stdin; returns a + CompletedProcess with bytes output. + + The process is tracked so ``shutdown_encoders`` can kill it; past + *timeout* it is killed and TimeoutExpired raised. After the shutdown + nothing starts (EncoderClosed). ``track(process)`` sees it too. + """ + with _process_lock: + if _closed: + raise EncoderClosed("Encoders are shut down") + process = subprocess.Popen( # nosec B603 - fixed argv, no shell + cmd, + stdin=subprocess.PIPE if input is not None else subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + creationflags=_NO_WINDOW, + ) + with _process_lock: + _processes.add(process) + closed = _closed + try: + if closed: + process.kill() + if track is not None: + track(process) + try: + out, err = process.communicate(input=input, timeout=timeout) + except subprocess.TimeoutExpired: + process.kill() + process.communicate() + raise + finally: + with _process_lock: + _processes.discard(process) + closed = _closed + if closed: + raise EncoderClosed("Encoder stopped by the shutdown") + return subprocess.CompletedProcess(cmd, process.returncode, out, err) + + +def shutdown_encoders(): + """Kill every running encoder process; none starts afterwards.""" + global _closed + with _process_lock: + _closed = True + running = list(_processes) + for process in running: + try: + process.kill() + except OSError: + pass + + +def reset_encoders(): + """Allow encoders again and forget the detection (tests, a reload).""" + global _closed, _detected, _fallback_logged + with _process_lock: + _closed = False + with _detect_lock: + _detected = _UNSET + _fallback_logged = False + + +# ── JPEG XL detection ──────────────────────────────────────────────── + +_UNSET = object() +_detect_lock = threading.Lock() +_detected = _UNSET +_fallback_logged = False + + +def _pillow_jxl_format(): + """Pillow's JPEG XL format id when it can save one, else None.""" + try: + native = Image.registered_extensions().get(".jxl") + except Exception: + native = None + if native and native in Image.SAVE: + return native + return PILLOW_JXL_FORMAT if PILLOW_JXL_FORMAT in Image.SAVE else None + + +def _pillow_can_save_jxl(): + if _pillow_jxl_format(): + return True + try: + importlib.import_module("pillow_jxl") + except Exception: + return False + return _pillow_jxl_format() is not None + + +def _pillow_decodes_jxl(): + try: + native = Image.registered_extensions().get(".jxl") + except Exception: + native = None + return bool(native and native in Image.OPEN) or PILLOW_JXL_FORMAT in Image.OPEN + + +def _ffmpeg_has_libjxl(): + """Whether the ffmpeg on PATH lists the ``libjxl`` encoder.""" + if not shutil.which("ffmpeg"): + return False + try: + done = run_encoder( + ["ffmpeg", "-hide_banner", "-encoders"], timeout=PROBE_TIMEOUT + ) + except (OSError, subprocess.SubprocessError, EncoderClosed): + return False + listing = done.stdout + if isinstance(listing, bytes): + listing = listing.decode("utf-8", "replace") + return any( + len(line.split()) > 1 and line.split()[1] == "libjxl" + for line in listing.splitlines() + ) + + +def _detect(): + if _pillow_can_save_jxl(): + return "pillow" + if _ffmpeg_has_libjxl(): + return "ffmpeg" + if shutil.which("cjxl"): + return "cjxl" + return None + + +def detect_jxl_encoder(refresh=False): + """``pillow``, ``ffmpeg``, ``cjxl`` or None (cached; blocking on a probe).""" + global _detected + with _detect_lock: + if refresh or _detected is _UNSET: + _detected = _detect() + return _detected + + +def encoder_status(): + """Which formats can be written now (``GET /prompt_manager/settings``).""" + encoder = detect_jxl_encoder() + return { + "webp": {"available": True}, + "jpeg": {"available": True}, + "jxl": { + "available": encoder is not None, + "encoder": encoder, + "reason": None if encoder is not None else JXL_UNAVAILABLE, + }, + } + + +def jxl_distance(quality): + """libjxl's distance for a 1–100 quality (q100 gives 0.1, never lossless).""" + if quality >= 100: + return 0.1 + if quality >= 30: + return 0.1 + (100 - quality) * 0.09 + return 53.0 / 3000.0 * quality * quality - 23.0 / 20.0 * quality + 25.0 + + +# ── frames and writers ─────────────────────────────────────────────── + + +def thumbnail_frame(img, edge): + """The first frame of *img* as RGB, fitted inside ``edge`` x ``edge``.""" + try: + img.seek(0) + except (EOFError, AttributeError, ValueError): + pass + try: + img.draft("RGB", (edge, edge)) # JPEG: decode at a reduced scale + except (AttributeError, ValueError, OSError): + pass + frame = img.convert("RGB") if img.mode != "RGB" else img.copy() + frame.thumbnail((edge, edge), Image.Resampling.LANCZOS) + return frame + + +def _png_bytes(img): + buffer = io.BytesIO() + img.save(buffer, "PNG", compress_level=1) + return buffer.getvalue() + + +def _write_webp(img, tmp, settings, track): + img.save(tmp, "WEBP", quality=settings.quality, method=WEBP_METHOD) + + +def _write_jpeg(img, tmp, settings, track): + img.save(tmp, "JPEG", quality=settings.quality, optimize=True) + + +def _check(done, tmp, name): + if done.returncode != 0 or not Path(tmp).is_file(): + detail = (done.stderr or b"")[-300:] + if isinstance(detail, bytes): + detail = detail.decode("utf-8", "replace") + raise EncoderFailed(f"{name} could not encode the preview: {detail.strip()}") + + +def _write_jxl(img, tmp, settings, track): + encoder = detect_jxl_encoder() + distance = f"{jxl_distance(settings.quality):.2f}" + if encoder == "pillow": + img.save( + tmp, + _pillow_jxl_format() or PILLOW_JXL_FORMAT, + quality=min(settings.quality, 99), + effort=JXL_EFFORT, + num_threads=1, + ) + return + if encoder == "ffmpeg": + cmd = [ + "ffmpeg", "-v", "error", "-f", "png_pipe", "-i", "-", + "-frames:v", "1", "-c:v", "libjxl", "-distance", distance, + "-effort", str(JXL_EFFORT), "-f", "image2", "-y", str(tmp), + ] # fmt: skip + _check(run_encoder(cmd, input=_png_bytes(img), track=track), tmp, "ffmpeg") + return + if encoder == "cjxl": + source = temp_sibling(Path(tmp).with_suffix(".png"), "src") + try: + with open(source, "wb") as handle: + handle.write(_png_bytes(img)) + cmd = ["cjxl", str(source), str(tmp), "-d", distance, "-e", str(JXL_EFFORT)] + _check(run_encoder(cmd, track=track), tmp, "cjxl") + finally: + _unlink_quietly(source) + return + raise EncoderFailed(JXL_UNAVAILABLE) + + +_WRITERS = {"webp": _write_webp, "jpeg": _write_jpeg, "jxl": _write_jxl} + + +def verify_thumbnail(path, fmt): + """Raise ThumbnailVerifyError unless *path* decodes as a preview. + + A JPEG XL file Pillow cannot read (no plugin) is checked by signature. + """ + try: + if fmt == "jxl" and not _pillow_decodes_jxl(): + with open(path, "rb") as handle: + head = handle.read(12) + if not head.startswith(JXL_SIGNATURES): + raise ThumbnailVerifyError("The JPEG XL preview has no signature") + return + with Image.open(path) as written: + written.load() + except ThumbnailVerifyError: + raise + except Exception as exc: + raise ThumbnailVerifyError(f"The new preview does not decode: {exc}") from exc + + +def _jxl_fallback(): + """Log once that JPEG XL previews are being written as WebP.""" + global _fallback_logged + with _detect_lock: + first, _fallback_logged = not _fallback_logged, True + if first: + logger.warning( + "JPEG XL is the thumbnail format but no JPEG XL encoder is available " + "any more; previews are written as WebP until one is installed" + ) + + +def write_thumbnail(frame, dst, settings, *, guard=None, track=None): + """Encode *frame* into *dst* in ``settings``' format; returns the path + written (``.webp`` when JPEG XL fell back). Atomic and verified; raises + PreviewWriteHeld, ThumbnailVerifyError, EncoderFailed, EncoderClosed or + the OSError of the write.""" + dst = Path(dst) + fmt = settings.format + if fmt == "jxl" and detect_jxl_encoder() is None: + _jxl_fallback() + fmt, dst = "webp", dst.with_suffix(".webp") + tmp = temp_sibling(dst) + try: + _WRITERS[fmt](frame, tmp, settings, track) + verify_thumbnail(tmp, fmt) + replace_unless_held(tmp, dst, guard) + finally: + _unlink_quietly(tmp) + return dst diff --git a/py/api/thumbnail_settings.py b/py/api/thumbnail_settings.py new file mode 100644 index 00000000..553bbe93 --- /dev/null +++ b/py/api/thumbnail_settings.py @@ -0,0 +1,186 @@ +"""Thumbnail settings and names (#317): the one place every preview writer reads. + +Settings (System › Appearance, ``POST /prompt_manager/settings``, config.json +``thumbnails``): ``thumbnail_format`` (``webp`` default, ``jxl``, ``jpeg``), +``thumbnail_quality`` (1–100, default 80) and ``thumbnail_edge`` (the longest +edge, 128–2048 px, default 512). ``current()`` is the snapshot a writer takes +once per preview; automatic previews, the Activity job, the history route and +the legacy gallery routes all go through it. + +Names: a media file ``/.`` previews as +``thumbnails//_thumb`` (``.webp``, +``.jxl``, ``.jpg``), images and videos alike. Two files with one stem in one +folder share that name (ruling R2). Older releases kept the source's own +extension (``.png``, ``.jpeg``, ``.gif``, ``.webp``, ``.jpg``; ``.jpg`` for a +video): those are read, served and converted, never written. A preview with a +format extension (``.webp``, ``.jxl``, ``.jpg``) belongs to every same-stem +source; a legacy-only one (``.png``, ``.jpeg``, ``.gif``) only to a source +with that extension (``owns``, ruling R4). Lookups take the first existing +preview the source owns, preferring the configured format. +""" + +import os +from dataclasses import dataclass +from pathlib import Path + +FORMATS = ("webp", "jxl", "jpeg") +FORMAT_EXTENSIONS = {"webp": ".webp", "jxl": ".jxl", "jpeg": ".jpg"} +FORMAT_LABELS = {"webp": "WebP", "jxl": "JPEG XL", "jpeg": "JPEG"} +DEFAULT_FORMAT = "webp" +DEFAULT_QUALITY = 80 +DEFAULT_EDGE = 512 +MIN_QUALITY, MAX_QUALITY = 1, 100 +MIN_EDGE, MAX_EDGE = 128, 2048 + +# Extensions a format writes, then the ones only older releases wrote. +WRITTEN_EXTENSIONS = (".webp", ".jxl", ".jpg") +LEGACY_EXTENSIONS = (".png", ".jpeg", ".gif") +THUMBNAIL_EXTENSIONS = WRITTEN_EXTENSIONS + LEGACY_EXTENSIONS +CONTENT_TYPES = { + ".webp": "image/webp", + ".jxl": "image/jxl", + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".png": "image/png", + ".gif": "image/gif", +} +THUMB_SUFFIX = "_thumb" + + +@dataclass(frozen=True) +class ThumbnailSettings: + """One preview's settings: the format, the quality and the longest edge.""" + + format: str + quality: int + edge: int + + @property + def extension(self): + return FORMAT_EXTENSIONS[self.format] + + @property + def size(self): + return (self.edge, self.edge) + + def describe(self): + """``WebP · quality 80 · 512 px`` (log lines and the job's scope).""" + return f"{FORMAT_LABELS[self.format]} · quality {self.quality} · {self.edge} px" + + +def current(): + """The configured settings now (read once per preview or per run).""" + from ..config import PromptManagerConfig + + fmt = PromptManagerConfig.THUMBNAIL_FORMAT + return ThumbnailSettings( + fmt if fmt in FORMATS else DEFAULT_FORMAT, + int(PromptManagerConfig.THUMBNAIL_QUALITY), + int(PromptManagerConfig.THUMBNAIL_EDGE), + ) + + +def _whole(value, low, high, message): + if type(value) is not int or not low <= value <= high: + raise ValueError(message) + return value + + +def validate_format(value): + if not isinstance(value, str) or value not in FORMATS: + raise ValueError('thumbnail_format must be "webp", "jxl" or "jpeg"') + return value + + +def validate_quality(value): + return _whole( + value, + MIN_QUALITY, + MAX_QUALITY, + f"thumbnail_quality must be a whole number from {MIN_QUALITY} to {MAX_QUALITY}", + ) + + +def validate_edge(value): + return _whole( + value, + MIN_EDGE, + MAX_EDGE, + "thumbnail_edge must be a whole number of pixels " + f"from {MIN_EDGE} to {MAX_EDGE}", + ) + + +# ── names ──────────────────────────────────────────────────────────── + + +def thumbnail_stem_rel(rel_path): + """``thumbnails/_thumb`` (POSIX).""" + return f"thumbnails/{Path(rel_path).with_suffix('').as_posix()}{THUMB_SUFFIX}" + + +def thumbnail_rel(rel_path, extension=None): + """The preview of *rel_path* in *extension* (the configured one when None).""" + return thumbnail_stem_rel(rel_path) + (extension or current().extension) + + +def owns(thumb_extension, source_extension): + """Whether a ``_thumb`` preview can belong to a source + with *source_extension* (ruling R4).""" + thumb_extension = thumb_extension.lower() + if thumb_extension in WRITTEN_EXTENSIONS: + return True + if thumb_extension in LEGACY_EXTENSIONS: + return thumb_extension == source_extension.lower() + return False + + +def _legacy_extension(source_extension): + from .inspection_service import VIDEO_EXTENSIONS + + source_extension = source_extension.lower() + return ".jpg" if source_extension in VIDEO_EXTENSIONS else source_extension + + +def candidate_extensions(source_extension, preferred=None): + """The preview extensions a source owns, in lookup order: the configured + one, the name an older release gave it, then the other written ones.""" + preferred = preferred or current().extension + ordered = [preferred, _legacy_extension(source_extension), *WRITTEN_EXTENSIONS] + seen = [] + for extension in ordered: + if extension not in seen and owns(extension, source_extension): + seen.append(extension) + return tuple(seen) + + +def _is_preview_file(path): + try: + return not path.is_symlink() and path.is_file() + except OSError: + return False + + +def _owned(root, rel_path, preferred): + rel_path = Path(rel_path) + stem = Path(root) / thumbnail_stem_rel(rel_path) + for extension in candidate_extensions(rel_path.suffix, preferred): + path = stem.with_name(stem.name + extension) + if _is_preview_file(path): + yield path + + +def owned_thumbnails(root, rel_path, preferred=None): + """Every existing preview of *rel_path* under *root*, in lookup order.""" + return list(_owned(root, rel_path, preferred)) + + +def existing_thumbnail(root, rel_path, preferred=None): + """The preview a listing shows for *rel_path*: the first owned one that + exists (the configured format first), or None.""" + return next(_owned(root, rel_path, preferred), None) + + +def content_type(path): + """The Content-Type of a preview file by its extension, or None.""" + return CONTENT_TYPES.get(os.path.splitext(str(path))[1].lower()) diff --git a/py/config.py b/py/config.py index 18f94f10..ba8a2943 100644 --- a/py/config.py +++ b/py/config.py @@ -200,7 +200,6 @@ class GalleryConfig: AUTO_CLEANUP_MISSING_FILES (bool): Automatically remove missing file records MAX_IMAGE_AGE_DAYS (int): Maximum age in days before cleaning up images IMAGES_PER_PAGE (int): Number of images to display per page in web UI - THUMBNAIL_SIZE (int): Size in pixels for generated thumbnails ENABLE_SEARCH (bool): Enable search functionality in web interface ENABLE_METADATA_VIEW (bool): Enable metadata viewing for images MAX_CONCURRENT_PROCESSING (int): Maximum concurrent image processing tasks @@ -225,7 +224,6 @@ class GalleryConfig: # Web interface settings IMAGES_PER_PAGE = 20 - THUMBNAIL_SIZE = 256 ENABLE_SEARCH = True ENABLE_METADATA_VIEW = True @@ -265,7 +263,6 @@ class GalleryConfig: }, "web_interface": { "images_per_page": cls.IMAGES_PER_PAGE, - "thumbnail_size": cls.THUMBNAIL_SIZE, "enable_search": cls.ENABLE_SEARCH, "enable_metadata_view": cls.ENABLE_METADATA_VIEW, }, @@ -328,8 +325,6 @@ class GalleryConfig: web_interface = new_config.get("web_interface", {}) if "images_per_page" in web_interface: cls.IMAGES_PER_PAGE = web_interface["images_per_page"] - if "thumbnail_size" in web_interface: - cls.THUMBNAIL_SIZE = web_interface["thumbnail_size"] if "enable_search" in web_interface: cls.ENABLE_SEARCH = web_interface["enable_search"] if "enable_metadata_view" in web_interface: @@ -663,6 +658,15 @@ class PromptManagerConfig: # database first (System › Backup & export › Recovery policy, #249 S6). SAFETY_BACKUP = True + # Thumbnails (#317): every preview writer reads these through + # py/api/thumbnail_settings.py (System › Appearance). The legacy + # ``gallery.web_interface.thumbnail_size`` was never read and is ignored. + THUMBNAIL_FORMATS = ("webp", "jxl", "jpeg") + THUMBNAIL_FORMAT = "webp" + THUMBNAIL_QUALITY = 80 + THUMBNAIL_EDGE = 512 + THUMBNAIL_EDGE_RANGE = (128, 2048) + @classmethod def max_worker_threads(cls) -> int: """Largest accepted worker-thread count: the detected core count.""" @@ -718,6 +722,11 @@ class PromptManagerConfig: "worker_threads": cls.WORKER_THREADS, }, "maintenance": {"safety_backup": cls.SAFETY_BACKUP}, + "thumbnails": { + "format": cls.THUMBNAIL_FORMAT, + "quality": cls.THUMBNAIL_QUALITY, + "edge": cls.THUMBNAIL_EDGE, + }, "gallery": GalleryConfig.get_config(), "integrations": IntegrationConfig.get_config(), "autotag": AutotagConfig.get_config(), @@ -859,6 +868,8 @@ class PromptManagerConfig: ): cls.SAFETY_BACKUP = maintenance["safety_backup"] + cls._update_thumbnails(new_config.get("thumbnails")) + # Update gallery config if "gallery" in new_config: GalleryConfig.update_config(new_config["gallery"]) @@ -870,6 +881,22 @@ class PromptManagerConfig: if "autotag" in new_config: AutotagConfig.update_config(new_config["autotag"]) + @classmethod + def _update_thumbnails(cls, section: Any) -> None: + """Apply a ``thumbnails`` section; a hand-edited bad value is ignored.""" + if not isinstance(section, dict): + return + fmt = section.get("format") + if fmt in cls.THUMBNAIL_FORMATS: + cls.THUMBNAIL_FORMAT = fmt + quality = section.get("quality") + if type(quality) is int and 1 <= quality <= 100: + cls.THUMBNAIL_QUALITY = quality + edge = section.get("edge") + low, high = cls.THUMBNAIL_EDGE_RANGE + if type(edge) is int and low <= edge <= high: + cls.THUMBNAIL_EDGE = edge + # Load configuration on import try: diff --git a/tests/test_api_admin.py b/tests/test_api_admin.py index 2aa28c9d..e1744bac 100644 --- a/tests/test_api_admin.py +++ b/tests/test_api_admin.py @@ -1328,6 +1328,103 @@ class TestSettingsMisc(AdminAPITestCase): self.assertEqual(resp.status, 500) +class TestThumbnailSettings(AdminAPITestCase): + """#317: thumbnail format, quality and size limit are server settings.""" + + URL = "/prompt_manager/settings" + + def _encoder(self, name): + from py.api import thumbnail_encode + + return patch.object(thumbnail_encode, "detect_jxl_encoder", return_value=name) + + async def test_get_reports_the_settings_and_the_encoders(self): + PromptManagerConfig.update_config( + {"thumbnails": {"format": "webp", "quality": 80, "edge": 512}} + ) + with self._encoder(None): + body = await (await self.client.request("GET", self.URL)).json() + settings = body["settings"] + self.assertEqual( + ( + settings["thumbnail_format"], + settings["thumbnail_quality"], + settings["thumbnail_edge"], + ), + ("webp", 80, 512), + ) + jxl = body["thumbnail_encoders"]["jxl"] + self.assertEqual((jxl["available"], jxl["encoder"]), (False, None)) + self.assertIn("pillow-jxl-plugin", jxl["reason"]) + with self._encoder("cjxl"): + body = await (await self.client.request("GET", self.URL)).json() + self.assertEqual( + body["thumbnail_encoders"]["jxl"], + {"available": True, "encoder": "cjxl", "reason": None}, + ) + + async def test_save_commits_the_thumbnails_section(self): + self._write_config({"custom": {"keep": 1}}) + resp = await self.client.request( + "POST", + self.URL, + json={ + "thumbnail_format": "jpeg", + "thumbnail_quality": 60, + "thumbnail_edge": 300, + }, + ) + self.assertEqual(resp.status, 200) + saved = self._read_config() + self.assertEqual( + saved["thumbnails"], {"format": "jpeg", "quality": 60, "edge": 300} + ) + self.assertEqual(saved["custom"], {"keep": 1}) + self.assertEqual( + ( + PromptManagerConfig.THUMBNAIL_FORMAT, + PromptManagerConfig.THUMBNAIL_QUALITY, + PromptManagerConfig.THUMBNAIL_EDGE, + ), + ("jpeg", 60, 300), + ) + + async def test_bad_values_are_refused_and_change_nothing(self): + before = PromptManagerConfig.get_config()["thumbnails"] + for payload in ( + {"thumbnail_format": "png"}, + {"thumbnail_quality": 0}, + {"thumbnail_quality": "80"}, + {"thumbnail_edge": 4096}, + {"thumbnail_edge": True}, + ): + resp = await self.client.request("POST", self.URL, json=payload) + self.assertEqual(resp.status, 400, payload) + self.assertEqual(PromptManagerConfig.get_config()["thumbnails"], before) + self.assertFalse(os.path.exists(self.config_path)) + + async def test_jpeg_xl_without_an_encoder_is_refused_with_a_code(self): + with self._encoder(None): + resp = await self.client.request( + "POST", self.URL, json={"thumbnail_format": "jxl", "result_timeout": 7} + ) + body = await resp.json() + self.assertEqual((resp.status, body["code"]), (400, "jxl-unavailable")) + self.assertIn("pillow-jxl-plugin", body["error"]) + self.assertNotEqual(PromptManagerConfig.THUMBNAIL_FORMAT, "jxl") + self.assertNotEqual(PromptManagerConfig.RESULT_TIMEOUT, 7) + self.assertFalse(os.path.exists(self.config_path)) + + async def test_jpeg_xl_with_an_encoder_is_saved(self): + with self._encoder("ffmpeg"): + resp = await self.client.request( + "POST", self.URL, json={"thumbnail_format": "jxl"} + ) + self.assertEqual(resp.status, 200) + self.assertEqual(self._read_config()["thumbnails"]["format"], "jxl") + self.assertEqual(PromptManagerConfig.THUMBNAIL_FORMAT, "jxl") + + class TestDiagnostics(AdminAPITestCase): def _install_monitor(self, monitor): diff --git a/tests/test_config.py b/tests/test_config.py index 0b2805a0..cf1da2fc 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -75,7 +75,8 @@ class TestGalleryConfig(unittest.TestCase): {"web_interface": {"images_per_page": 50, "thumbnail_size": 512}} ) self.assertEqual(GalleryConfig.IMAGES_PER_PAGE, 50) - self.assertEqual(GalleryConfig.THUMBNAIL_SIZE, 512) + # #317 R1: the never-read thumbnail_size is ignored (thumbnails.edge). + self.assertFalse(hasattr(GalleryConfig, "THUMBNAIL_SIZE")) def test_update_performance(self): GalleryConfig.update_config({"performance": {"max_concurrent_processing": 5}}) diff --git a/tests/test_thumbnail_encode.py b/tests/test_thumbnail_encode.py new file mode 100644 index 00000000..ed018c50 --- /dev/null +++ b/tests/test_thumbnail_encode.py @@ -0,0 +1,327 @@ +"""Thumbnail encoders (#317): WebP, JPEG and an optional JPEG XL encoder. + +``py/api/thumbnail_encode.py`` writes every preview: the first frame fitted +to the edge, encoded in the configured format into a temp sibling that is +closed, verified to decode and renamed over the preview. JPEG XL needs no new +dependency: Pillow (natively or through ``pillow_jxl``), ffmpeg's ``libjxl`` +or ``cjxl``, in that order. Encoder processes are tracked, bounded and +killed on shutdown. +""" + +import logging +import os +import shutil +import subprocess +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from unittest.mock import patch + +from unittest.mock import MagicMock + +sys.modules.setdefault("server", MagicMock()) +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from PIL import Image # noqa: E402 + +from py.api import thumbnail_encode as te # noqa: E402 +from py.api.thumbnail_settings import ThumbnailSettings # noqa: E402 + +JXL_CODESTREAM = b"\xff\x0a" +JXL_CONTAINER = b"\x00\x00\x00\x0cJXL \r\n\x87\n" + + +def _ffmpeg_has_libjxl(): + try: + return te._ffmpeg_has_libjxl() + except Exception: + return False + + +class TempCase(unittest.TestCase): + def setUp(self): + self.tmp = Path(os.path.realpath(tempfile.mkdtemp())) + self.addCleanup(shutil.rmtree, self.tmp, True) + te.reset_encoders() + self.addCleanup(te.reset_encoders) + + def _image(self, size=(800, 400), color=(200, 30, 30), mode="RGB"): + return Image.new(mode, size, color) + + def _leftovers(self): + return [p.name for p in self.tmp.rglob("*") if p.name.startswith(".")] + + +class TestDistance(unittest.TestCase): + def test_quality_maps_to_libjxl_distance(self): + self.assertAlmostEqual(te.jxl_distance(80), 1.9) + self.assertAlmostEqual(te.jxl_distance(85), 1.45) + self.assertAlmostEqual(te.jxl_distance(90), 1.0) + self.assertAlmostEqual(te.jxl_distance(30), 6.4) + self.assertAlmostEqual(te.jxl_distance(1), 23.8676666, places=4) + # A preview is never encoded lossless. + self.assertAlmostEqual(te.jxl_distance(100), 0.1) + + +class TestDetection(TempCase): + def test_pillow_native_first(self): + with patch.dict(Image.SAVE, {"JXL": object()}): + self.assertEqual(te.detect_jxl_encoder(refresh=True), "pillow") + + def test_pillow_plugin_is_imported_when_installed(self): + def fake_import(name): + if name != "pillow_jxl": + raise ImportError(name) + Image.SAVE["JXL"] = object() + return object() + + self.addCleanup(Image.SAVE.pop, "JXL", None) + with patch.object(te.importlib, "import_module", side_effect=fake_import): + self.assertEqual(te.detect_jxl_encoder(refresh=True), "pillow") + + def _no_pillow(self): + saved = Image.SAVE.pop("JXL", None) + if saved is not None: + self.addCleanup(Image.SAVE.__setitem__, "JXL", saved) + return patch.object(te.importlib, "import_module", side_effect=ImportError) + + def test_ffmpeg_with_libjxl_then_cjxl_then_none(self): + with self._no_pillow(): + with patch.object(te, "_ffmpeg_has_libjxl", return_value=True): + self.assertEqual(te.detect_jxl_encoder(refresh=True), "ffmpeg") + with ( + patch.object(te, "_ffmpeg_has_libjxl", return_value=False), + patch.object(te.shutil, "which", return_value="/usr/bin/cjxl"), + ): + self.assertEqual(te.detect_jxl_encoder(refresh=True), "cjxl") + with ( + patch.object(te, "_ffmpeg_has_libjxl", return_value=False), + patch.object(te.shutil, "which", return_value=None), + ): + self.assertIsNone(te.detect_jxl_encoder(refresh=True)) + status = te.encoder_status() + self.assertEqual( + status["jxl"], + {"available": False, "encoder": None, "reason": te.JXL_UNAVAILABLE}, + ) + self.assertIn("pillow-jxl-plugin", te.JXL_UNAVAILABLE) + self.assertTrue(status["webp"]["available"]) + self.assertTrue(status["jpeg"]["available"]) + + def test_detection_is_cached_until_refreshed(self): + with ( + self._no_pillow(), + patch.object(te, "_ffmpeg_has_libjxl", return_value=True) as probe, + ): + te.detect_jxl_encoder(refresh=True) + te.detect_jxl_encoder() + te.encoder_status() + self.assertEqual(probe.call_count, 1) + + def test_ffmpeg_probe_reads_the_encoder_list(self): + listing = " V....D libjxl libjxl JPEG XL (codec jpegxl)\n" + done = subprocess.CompletedProcess([], 0, listing, "") + with ( + patch.object(te, "run_encoder", return_value=done), + patch.object(te.shutil, "which", return_value="/usr/bin/ffmpeg"), + ): + self.assertTrue(te._ffmpeg_has_libjxl()) + other = subprocess.CompletedProcess([], 0, " V..... libwebp\n", "") + with ( + patch.object(te, "run_encoder", return_value=other), + patch.object(te.shutil, "which", return_value="/usr/bin/ffmpeg"), + ): + self.assertFalse(te._ffmpeg_has_libjxl()) + with patch.object(te.shutil, "which", return_value=None): + self.assertFalse(te._ffmpeg_has_libjxl()) + + +class TestFrame(TempCase): + def test_first_frame_of_an_animation_is_a_still(self): + path = self.tmp / "anim.gif" + frames = [Image.new("RGB", (64, 64), c) for c in ((255, 0, 0), (0, 0, 255))] + frames[0].save(path, save_all=True, append_images=frames[1:], duration=50) + with Image.open(path) as img: + img.seek(1) + frame = te.thumbnail_frame(img, 32) + self.assertEqual(frame.mode, "RGB") + self.assertEqual(frame.size, (32, 32)) + red, green, blue = frame.getpixel((16, 16)) + self.assertGreater(red, 200) + self.assertLess(blue, 50) + + def test_modes_without_rgb_are_converted_and_fitted(self): + for mode, color in (("RGBA", (1, 2, 3, 0)), ("I;16", 300), ("L", 9)): + frame = te.thumbnail_frame(self._image((1000, 250), color, mode), 512) + self.assertEqual((frame.mode, frame.size), ("RGB", (512, 128))) + + +class TestWrite(TempCase): + def _settings(self, fmt="webp", quality=80, edge=512): + return ThumbnailSettings(fmt, quality, edge) + + def test_webp_and_jpeg_are_written_in_their_format(self): + for fmt, name, pil in (("webp", "a.webp", "WEBP"), ("jpeg", "a.jpg", "JPEG")): + dst = self.tmp / name + written = te.write_thumbnail(self._image(), dst, self._settings(fmt)) + self.assertEqual(written, dst) + with Image.open(dst) as img: + self.assertEqual(img.format, pil) + self.assertEqual(self._leftovers(), []) + + def test_quality_changes_the_bytes(self): + noisy = Image.effect_noise((256, 256), 64).convert("RGB") + low = te.write_thumbnail( + noisy, self.tmp / "low.webp", self._settings(quality=10) + ) + high = te.write_thumbnail( + noisy, self.tmp / "high.webp", self._settings(quality=95) + ) + self.assertLess(low.stat().st_size, high.stat().st_size) + + def test_guard_false_keeps_the_old_preview(self): + dst = self.tmp / "a.webp" + dst.write_bytes(b"old") + with self.assertRaises(te.PreviewWriteHeld): + te.write_thumbnail( + self._image(), dst, self._settings(), guard=lambda: False + ) + self.assertEqual(dst.read_bytes(), b"old") + self.assertEqual(self._leftovers(), []) + + def test_a_file_that_does_not_decode_is_never_renamed_in(self): + dst = self.tmp / "a.webp" + dst.write_bytes(b"old") + + def broken(img, tmp, settings, track): + Path(tmp).write_bytes(b"not an image") + + with patch.dict(te._WRITERS, {"webp": broken}): + with self.assertRaises(te.ThumbnailVerifyError): + te.write_thumbnail(self._image(), dst, self._settings()) + self.assertEqual(dst.read_bytes(), b"old") + self.assertEqual(self._leftovers(), []) + + def test_jxl_without_an_encoder_falls_back_to_webp_and_logs_once(self): + with patch.object(te, "detect_jxl_encoder", return_value=None): + with self.assertLogs("prompt_manager.thumbnails", logging.WARNING) as logs: + first = te.write_thumbnail( + self._image(), self.tmp / "a.jxl", self._settings("jxl") + ) + second = te.write_thumbnail( + self._image(), self.tmp / "b.jxl", self._settings("jxl") + ) + self.assertEqual((first.name, second.name), ("a.webp", "b.webp")) + self.assertEqual(len([m for m in logs.output if "JPEG XL" in m]), 1) + with Image.open(first) as img: + self.assertEqual(img.format, "WEBP") + + def _write_jxl(self, encoder): + dst = self.tmp / f"{encoder}.jxl" + with patch.object(te, "detect_jxl_encoder", return_value=encoder): + written = te.write_thumbnail(self._image(), dst, self._settings("jxl")) + self.assertEqual(written, dst) + head = dst.read_bytes()[:12] + self.assertTrue( + head.startswith(JXL_CODESTREAM) or head.startswith(JXL_CONTAINER), head + ) + self.assertEqual(self._leftovers(), []) + + @unittest.skipUnless(_ffmpeg_has_libjxl(), "ffmpeg without libjxl") + def test_jxl_through_ffmpeg(self): + self._write_jxl("ffmpeg") + + @unittest.skipUnless(shutil.which("cjxl"), "no cjxl") + def test_jxl_through_cjxl(self): + self._write_jxl("cjxl") + + def test_jxl_through_pillow_passes_the_quality(self): + seen = {} + + def save_jxl(im, fp, filename): + seen.update(im.encoderinfo) + fp.write(JXL_CODESTREAM + b"\x00" * 16) + + Image.register_save("PMTESTJXL", save_jxl) + self.addCleanup(Image.SAVE.pop, "PMTESTJXL", None) + with ( + patch.object(te, "PILLOW_JXL_FORMAT", "PMTESTJXL"), + patch.object(te, "detect_jxl_encoder", return_value="pillow"), + ): + te.write_thumbnail(self._image(), self.tmp / "p.jxl", self._settings("jxl")) + self.assertEqual(seen["quality"], 80) + self.assertEqual(seen["num_threads"], 1) + + def test_a_failing_jxl_encoder_raises_and_keeps_nothing(self): + failed = subprocess.CompletedProcess([], 1, b"", b"boom") + with ( + patch.object(te, "detect_jxl_encoder", return_value="cjxl"), + patch.object(te, "run_encoder", return_value=failed), + ): + with self.assertRaises(te.EncoderFailed): + te.write_thumbnail( + self._image(), self.tmp / "c.jxl", self._settings("jxl") + ) + self.assertFalse((self.tmp / "c.jxl").exists()) + self.assertEqual(self._leftovers(), []) + + def test_verify_accepts_a_jxl_signature_when_pillow_cannot_decode(self): + path = self.tmp / "s.jxl" + path.write_bytes(JXL_CONTAINER + b"\x00" * 8) + with patch.object(te, "_pillow_decodes_jxl", return_value=False): + te.verify_thumbnail(path, "jxl") + path.write_bytes(b"\x89PNG") + with self.assertRaises(te.ThumbnailVerifyError): + te.verify_thumbnail(path, "jxl") + + +class TestEncoderProcesses(TempCase): + SLEEPER = [sys.executable, "-c", "import time; time.sleep(60)"] + + def test_a_process_past_its_timeout_is_killed(self): + with self.assertRaises(subprocess.TimeoutExpired): + te.run_encoder(self.SLEEPER, timeout=0.5) + + def test_shutdown_kills_running_encoders_and_refuses_new_ones(self): + started = threading.Event() + tracked = [] + + def track(process): + tracked.append(process) + started.set() + + outcome = {} + + def run(): + try: + outcome["result"] = te.run_encoder( + self.SLEEPER, timeout=60, track=track + ) + except Exception as exc: # noqa: BLE001 - recorded for the assertion + outcome["error"] = exc + + worker = threading.Thread(target=run) + worker.start() + self.assertTrue(started.wait(20)) + te.shutdown_encoders() + worker.join(20) + self.assertFalse(worker.is_alive()) + self.assertIsNotNone(tracked[0].poll()) + self.assertIsInstance(outcome.get("error"), te.EncoderClosed) + with self.assertRaises(te.EncoderClosed): + te.run_encoder(self.SLEEPER, timeout=1) + + def test_input_is_piped_and_output_returned(self): + echo = [ + sys.executable, + "-c", + "import sys; sys.stdout.buffer.write(sys.stdin.buffer.read()[::-1])", + ] + done = te.run_encoder(echo, input=b"abc", timeout=30) + self.assertEqual((done.returncode, done.stdout), (0, b"cba")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_thumbnail_settings.py b/tests/test_thumbnail_settings.py new file mode 100644 index 00000000..0b6c011f --- /dev/null +++ b/tests/test_thumbnail_settings.py @@ -0,0 +1,214 @@ +"""Thumbnail settings and naming (#317): the one module every generator reads. + +``py/api/thumbnail_settings.py`` owns ``thumbnail_format`` / ``thumbnail_quality`` +/ ``thumbnail_edge`` (``PromptManagerConfig``, config.json ``thumbnails``), the +preview's name ``thumbnails/_thumb.`` and the lookup +that accepts any preview a source owns, preferring the configured format. +""" + +import json +import os +import shutil +import sys +import tempfile +import unittest +from pathlib import Path + +from unittest.mock import MagicMock + +sys.modules.setdefault("server", MagicMock()) +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from py.api import thumbnail_settings as ts # noqa: E402 +from py.config import GalleryConfig, PromptManagerConfig # noqa: E402 + + +class ConfigCase(unittest.TestCase): + def setUp(self): + self._orig = PromptManagerConfig.get_config() + self.addCleanup(PromptManagerConfig.update_config, self._orig) + + +class TestDefaultsAndConfig(ConfigCase): + def test_defaults_are_webp_80_512(self): + PromptManagerConfig.update_config( + {"thumbnails": {"format": "webp", "quality": 80, "edge": 512}} + ) + settings = ts.current() + self.assertEqual( + (settings.format, settings.quality, settings.edge), ("webp", 80, 512) + ) + self.assertEqual(settings.extension, ".webp") + self.assertEqual(settings.size, (512, 512)) + self.assertEqual( + (ts.DEFAULT_FORMAT, ts.DEFAULT_QUALITY, ts.DEFAULT_EDGE), ("webp", 80, 512) + ) + + def test_config_round_trip_in_a_thumbnails_section(self): + PromptManagerConfig.update_config( + {"thumbnails": {"format": "jpeg", "quality": 55, "edge": 300}} + ) + self.assertEqual( + PromptManagerConfig.get_config()["thumbnails"], + {"format": "jpeg", "quality": 55, "edge": 300}, + ) + self.assertEqual(ts.current().extension, ".jpg") + + def test_hand_edited_bad_values_are_ignored(self): + PromptManagerConfig.update_config( + {"thumbnails": {"format": "webp", "quality": 80, "edge": 512}} + ) + for bad in ( + {"format": "png"}, + {"format": 3}, + {"quality": 0}, + {"quality": 101}, + {"quality": True}, + {"quality": "80"}, + {"edge": 64}, + {"edge": 4096}, + {"edge": 512.5}, + ): + PromptManagerConfig.update_config({"thumbnails": bad}) + self.assertEqual( + (ts.current().format, ts.current().quality, ts.current().edge), + ("webp", 80, 512), + ) + PromptManagerConfig.update_config({"thumbnails": "nonsense"}) + self.assertEqual(ts.current().format, "webp") + + def test_the_never_read_gallery_thumbnail_size_is_gone(self): + # R1: older save_to_file wrote its default 256 into config.json; it + # must never size a preview. + self.assertFalse(hasattr(GalleryConfig, "THUMBNAIL_SIZE")) + PromptManagerConfig.update_config( + {"gallery": {"web_interface": {"thumbnail_size": 256}}} + ) + self.assertNotIn("thumbnail_size", GalleryConfig.get_config()["web_interface"]) + self.assertNotEqual(ts.current().edge, 256) + + def test_load_from_file_reads_the_section(self): + tmp = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, tmp, True) + path = os.path.join(tmp, "config.json") + with open(path, "w") as handle: + json.dump({"thumbnails": {"format": "jxl", "quality": 90}}, handle) + PromptManagerConfig.load_from_file(path) + self.assertEqual((ts.current().format, ts.current().quality), ("jxl", 90)) + + +class TestValidation(unittest.TestCase): + def test_format(self): + for value in ("webp", "jxl", "jpeg"): + self.assertEqual(ts.validate_format(value), value) + for value in ("png", "JPEG", "", None, 1): + with self.assertRaises(ValueError): + ts.validate_format(value) + + def test_quality(self): + self.assertEqual(ts.validate_quality(1), 1) + self.assertEqual(ts.validate_quality(100), 100) + for value in (0, 101, True, 80.0, "80", None): + with self.assertRaises(ValueError): + ts.validate_quality(value) + + def test_edge(self): + self.assertEqual(ts.validate_edge(128), 128) + self.assertEqual(ts.validate_edge(2048), 2048) + for value in (127, 2049, False, 512.0, "512"): + with self.assertRaises(ValueError): + ts.validate_edge(value) + + +class TestNaming(ConfigCase): + def setUp(self): + super().setUp() + PromptManagerConfig.update_config( + {"thumbnails": {"format": "webp", "quality": 80, "edge": 512}} + ) + + def test_images_and_videos_take_the_configured_extension(self): + self.assertEqual( + ts.thumbnail_rel(Path("sub/a.PNG")), "thumbnails/sub/a_thumb.webp" + ) + self.assertEqual( + ts.thumbnail_rel(Path("clip.mp4")), "thumbnails/clip_thumb.webp" + ) + self.assertEqual( + ts.thumbnail_rel(Path("a.png"), ".jxl"), "thumbnails/a_thumb.jxl" + ) + + def test_ownership(self): + # R4: legacy-only extensions belong to a source of that extension. + self.assertTrue(ts.owns(".png", ".png")) + self.assertTrue(ts.owns(".png", ".PNG")) + self.assertFalse(ts.owns(".png", ".mp4")) + self.assertFalse(ts.owns(".png", ".jpg")) + self.assertTrue(ts.owns(".jpeg", ".jpeg")) + self.assertTrue(ts.owns(".gif", ".gif")) + for ext in (".webp", ".jxl", ".jpg"): + for source in (".png", ".mp4", ".gif", ".webp"): + self.assertTrue(ts.owns(ext, source)) + self.assertFalse(ts.owns(".txt", ".png")) + + def test_candidates_prefer_the_configured_format_then_the_legacy_name(self): + self.assertEqual( + ts.candidate_extensions(".png", ".webp"), (".webp", ".png", ".jxl", ".jpg") + ) + self.assertEqual( + ts.candidate_extensions(".mp4", ".jxl"), (".jxl", ".jpg", ".webp") + ) + self.assertEqual( + ts.candidate_extensions(".jpg", ".jpg"), (".jpg", ".webp", ".jxl") + ) + + def test_content_types(self): + self.assertEqual(ts.content_type(Path("a_thumb.webp")), "image/webp") + self.assertEqual(ts.content_type(Path("a_thumb.jxl")), "image/jxl") + self.assertEqual(ts.content_type(Path("a_thumb.jpg")), "image/jpeg") + self.assertEqual(ts.content_type(Path("a_thumb.PNG")), "image/png") + self.assertEqual(ts.content_type(Path("a_thumb.gif")), "image/gif") + self.assertIsNone(ts.content_type(Path("a.txt"))) + + +class TestLookup(ConfigCase): + def setUp(self): + super().setUp() + PromptManagerConfig.update_config( + {"thumbnails": {"format": "webp", "quality": 80, "edge": 512}} + ) + self.root = Path(os.path.realpath(tempfile.mkdtemp())) + self.addCleanup(shutil.rmtree, self.root, True) + + def _touch(self, rel): + path = self.root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b"x") + return path + + def test_missing_preview_is_none(self): + self.assertIsNone(ts.existing_thumbnail(self.root, Path("a.png"))) + + def test_legacy_preview_is_found_and_the_configured_one_preferred(self): + legacy = self._touch("thumbnails/sub/a_thumb.png") + self.assertEqual(ts.existing_thumbnail(self.root, Path("sub/a.png")), legacy) + fresh = self._touch("thumbnails/sub/a_thumb.webp") + self.assertEqual(ts.existing_thumbnail(self.root, Path("sub/a.png")), fresh) + self.assertEqual( + sorted(ts.owned_thumbnails(self.root, Path("sub/a.png"))), + sorted([legacy, fresh]), + ) + + def test_a_preview_owned_by_another_source_is_not_used(self): + self._touch("thumbnails/a_thumb.png") + self.assertIsNone(ts.existing_thumbnail(self.root, Path("a.mp4"))) + video = self._touch("thumbnails/a_thumb.jpg") + self.assertEqual(ts.existing_thumbnail(self.root, Path("a.mp4")), video) + + def test_a_folder_or_link_is_no_preview(self): + (self.root / "thumbnails" / "a_thumb.webp").mkdir(parents=True) + self.assertIsNone(ts.existing_thumbnail(self.root, Path("a.png"))) + + +if __name__ == "__main__": + unittest.main()