Fix IIIF Level 0 thumbnail resolution

backend/app/api/v1/endpoint_modules/resources/thumbnail.py

@@ -13,6 +13,7 @@
 from app.api.v1.utils import _get_thumbnail_asset_url, sanitize_for_json
 from app.services.cache_service import alias_redirect_cache_control_header
 from app.services.distribution_repository import fetch_distribution_context
+from app.services.iiif_url import is_iiif_info_url
 from app.services.image_service import ImageService
 from app.services.static_map_service import StaticMapService
 from app.services.thumbnail_alias_service import is_thumbnail_hash, thumbnail_alias_service
@@ -532,6 +533,7 @@ async def _get_resource_thumbnail_response(
     # URLs are processed server-side, so skip probe.
     if (
         not image_service._is_manifest_url(source_url)
+        and not is_iiif_info_url(source_url)
         and not image_service._is_cog_url(source_url)
         and not image_service._is_pmtiles_url(source_url)
         and THUMBNAIL_REQUEST_PROBE_ENABLED
@@ -606,7 +608,7 @@ async def _get_resource_thumbnail_response(
                         state_detail="PMTiles thumbnail generation already queued",
                     )
                 )
-        elif image_service._is_manifest_url(source_url):
+        elif image_service._is_manifest_url(source_url) or is_iiif_info_url(source_url):
             image_service._queue_thumbnail_processing(source_url, id)
         else:
             standardized_url = image_service._standardize_iiif_url(source_url)
@@ -743,10 +745,19 @@ async def get_resource_thumbnail_no_cache(
                 )
             return await _svg_icon_for_resource(resource_dict, variant=variant)
 
-        # Resolve manifests to actual image URLs when needed
-        if image_service._is_manifest_url(source_url):
+        # Resolve IIIF metadata to an actual image URL when needed.
+        if is_iiif_info_url(source_url):
+            resolved = await asyncio.to_thread(image_service.get_iiif_image_thumbnail, source_url)
+            if not resolved:
+                return _svg_placeholder(
+                    title="Thumbnail unavailable", subtitle="Error resolving IIIF"
+                )
+            fetch_url = resolved
+        elif image_service._is_manifest_url(source_url):
             # This may fetch the manifest once to resolve thumbnail URL
-            resolved = image_service.get_iiif_manifest_thumbnail(source_url)
+            resolved = await asyncio.to_thread(
+                image_service.get_iiif_manifest_thumbnail, source_url
+            )
             if not resolved:
                 return _svg_placeholder(
                     title="Thumbnail unavailable", subtitle="Error resolving IIIF"

backend/app/services/iiif_url.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from urllib.parse import urlparse
+
+
+def _path_basename(url: str | None) -> str:
+    if not url:
+        return ""
+    try:
+        path = urlparse(url).path.rstrip("/").lower()
+    except (TypeError, ValueError):
+        return ""
+    return path.rsplit("/", 1)[-1]
+
+
+def is_iiif_manifest_url(url: str | None) -> bool:
+    """Return True for an actual IIIF Presentation manifest path.
+
+    Matching the final path component avoids treating OGM package metadata such
+    as ``dataset_manifest.json`` as a IIIF Presentation manifest.
+    """
+    basename = _path_basename(url)
+    if basename == "manifest" or (basename.startswith("manifest") and basename.endswith(".json")):
+        return True
+    return isinstance(url, str) and "/cgi/i/image/api/" in url.lower()
+
+
+def is_iiif_info_url(url: str | None) -> bool:
+    """Return True for a IIIF Image API info document path."""
+    return _path_basename(url) == "info.json"

backend/app/services/image_service.py

@@ -16,6 +16,7 @@
     DistributionContext,
     build_distribution_context,
 )
+from app.services.iiif_url import is_iiif_info_url, is_iiif_manifest_url
 from app.services.thumbnail_alias_service import thumbnail_alias_service
 from app.services.thumbnail_queue_service import acquire_thumbnail_queue_slot
 from app.services.thumbnail_state_service import (
@@ -313,6 +314,84 @@ def get_iiif_manifest_thumbnail(self, manifest_url: str) -> Optional[str]:
 
         return self._extract_thumbnail_from_manifest_json(manifest_json, manifest_url)
 
+    def _iiif_thumbnail_target_edge(self) -> int:
+        """Return the requested thumbnail edge encoded in IIIF_THUMBNAIL_BOX."""
+        dimensions = [int(value) for value in re.findall(r"\d+", IIIF_THUMBNAIL_BOX)]
+        return max(dimensions, default=800)
+
+    def _is_level_zero_iiif_info(self, info_json: Dict[str, Any]) -> bool:
+        """Return True when an Image API info document advertises Level 0."""
+        profile = info_json.get("profile")
+        candidates: List[Any]
+        if isinstance(profile, list):
+            candidates = profile
+        else:
+            candidates = [profile]
+
+        for candidate in candidates:
+            if isinstance(candidate, str) and "level0" in candidate.lower():
+                return True
+            if isinstance(candidate, dict):
+                identifier = candidate.get("@id") or candidate.get("id")
+                if isinstance(identifier, str) and "level0" in identifier.lower():
+                    return True
+        return False
+
+    def _extract_thumbnail_from_iiif_info_json(
+        self,
+        info_json: Dict[str, Any],
+        info_url: str,
+    ) -> Optional[str]:
+        """Resolve a thumbnail request supported by a IIIF Image API service."""
+        service_id = info_json.get("@id") or info_json.get("id")
+        if not isinstance(service_id, str) or not service_id.strip():
+            service_id = info_url[: -len("/info.json")] if is_iiif_info_url(info_url) else None
+        if not service_id:
+            return None
+        service_id = service_id.rstrip("/")
+
+        if not self._is_level_zero_iiif_info(info_json):
+            return f"{service_id}{IIIF_THUMBNAIL_PATH}"
+
+        advertised_sizes = []
+        for size in info_json.get("sizes") or []:
+            if not isinstance(size, dict):
+                continue
+            width = size.get("width")
+            height = size.get("height")
+            if (
+                isinstance(width, int)
+                and not isinstance(width, bool)
+                and width > 0
+                and isinstance(height, int)
+                and not isinstance(height, bool)
+                and height > 0
+            ):
+                advertised_sizes.append((width, height))
+
+        if advertised_sizes:
+            target_edge = self._iiif_thumbnail_target_edge()
+            width, _height = min(
+                advertised_sizes,
+                key=lambda size: (abs(max(size) - target_edge), max(size)),
+            )
+            # Level 0 static services commonly materialize advertised sizes using
+            # the aspect-preserving width form (for example ``924,``), not an
+            # arbitrary bounded-box request such as ``!800,800``.
+            return f"{service_id}/full/{width},/0/default.jpg"
+
+        # Level 0 guarantees only a small request set. If no sizes are advertised,
+        # request the full image rather than inventing an unsupported resize.
+        return f"{service_id}/full/full/0/default.jpg"
+
+    def get_iiif_image_thumbnail(self, info_url: str) -> Optional[str]:
+        """Fetch Image API info.json and select a supported thumbnail rendition."""
+        info_json = self._get_manifest(info_url)
+        if not info_json:
+            self.logger.warning(f"Could not fetch IIIF image info {info_url}")
+            return None
+        return self._extract_thumbnail_from_iiif_info_json(info_json, info_url)
+
     def _standardize_iiif_url(self, url: str) -> str:
         """
         Standardize IIIF image URLs to ensure consistent size.
@@ -411,6 +490,28 @@ def thumbnail_image_hash_for_source_sync(
                 return hashlib.sha256((COG_THUMBNAIL_PREFIX + source_url).encode()).hexdigest()
             if self._is_pmtiles_url(source_url):
                 return hashlib.sha256((PMTILES_THUMBNAIL_PREFIX + source_url).encode()).hexdigest()
+            if self._is_iiif_info_url(source_url):
+                info_cache_key = f"manifest:{source_url}"
+                cached_info_data = self.cache.get(info_cache_key)
+                if cached_info_data:
+                    info_json = json.loads(cached_info_data)
+                    resolved_url = self._extract_thumbnail_from_iiif_info_json(
+                        info_json, source_url
+                    )
+                    if resolved_url:
+                        return hashlib.sha256(
+                            (REMOTE_THUMBNAIL_PREFIX + resolved_url).encode()
+                        ).hexdigest()
+
+                if resolve_manifest:
+                    from app.tasks.worker import _resolve_image_url
+
+                    resolved_url = _resolve_image_url(source_url)
+                    if resolved_url and resolved_url != source_url:
+                        return hashlib.sha256(
+                            (REMOTE_THUMBNAIL_PREFIX + resolved_url).encode()
+                        ).hexdigest()
+                return None
             if self._is_manifest_url(source_url):
                 manifest_cache_key = f"manifest:{source_url}"
                 cached_manifest_data = self.cache.get(manifest_cache_key)
@@ -720,7 +821,12 @@ def _get_thumbnail_source_url(
                         f"https://cdm16022.contentdm.oclc.org/iiif/2/{collection_item}"
                     )
 
-            # For non-ContentDM IIIF URLs, use standard format
+            # Preserve Image API info documents for the worker. Level 0 services
+            # must be read before choosing one of their advertised static sizes.
+            if self._is_iiif_info_url(iiif_url):
+                return iiif_url
+
+            # For non-ContentDM IIIF URLs, use standard format.
             return self._standardize_iiif_url(iiif_url)
 
         # Check for IIIF Manifest - only extract URL, don't fetch manifest
@@ -729,15 +835,11 @@ def _get_thumbnail_source_url(
             "http://iiif.io/api/presentation#manifest", references=references
         ) or self._first_url("https://iiif.io/api/presentation#manifest", references=references)
 
-        # If not found, scan values for common manifest endings
+        # If not found, scan values for actual manifest path components. This
+        # intentionally excludes package metadata such as dataset_manifest.json.
         if not manifest_url:
             for value in self._all_reference_urls(references=references):
-                if (
-                    value.endswith(
-                        ("/iiif3/manifest", "/iiif/manifest", "/manifest", "manifest.json")
-                    )
-                    or "/manifest" in value
-                ):
+                if is_iiif_manifest_url(value):
                     manifest_url = value
                     break
 
@@ -762,10 +864,8 @@ def _get_thumbnail_source_url(
                     )
                     return image_url
 
-            # For other manifests, queue background resolution and return manifest URL
-            # The Celery worker will resolve the manifest and extract the image URL
-            self.logger.info(f"🚀 Queueing manifest resolution for {manifest_url}")
-            self._queue_manifest_processing(manifest_url)
+            # For other manifests, return the source without side effects. The
+            # thumbnail endpoint owns queueing so each request creates at most one job.
             return manifest_url
 
         # Use curated b1g_image_ss only after exhausting IIIF-based options.
@@ -861,20 +961,11 @@ def _is_pmtiles_url(self, url: str) -> bool:
 
     def _is_manifest_url(self, url: str) -> bool:
         """Check if URL looks like a IIIF manifest URL."""
-        if not url:
-            return False
-        url_lower = url.lower()
-        # Check for common IIIF manifest patterns
-        return (
-            url.endswith(("/iiif3/manifest", "/iiif/manifest", "/manifest", "manifest.json"))
-            or "/manifest" in url
-            or (
-                ".json" in url
-                and ("iiif" in url_lower or "/object/" in url or "/collection/" in url)
-            )
-            or ("/api/" in url and ("iiif" in url_lower or "image" in url_lower))
-            or ("/cgi/i/image/api/" in url_lower)  # U of Michigan pattern
-        )
+        return is_iiif_manifest_url(url)
+
+    def _is_iiif_info_url(self, url: str) -> bool:
+        """Check if URL points to an Image API info document."""
+        return is_iiif_info_url(url)
 
     def _first_url(self, uri: str, references: Optional[Dict[str, Any]] = None) -> Optional[str]:
         # Prefer distribution context if available and no explicit references provided
@@ -971,21 +1062,6 @@ def _queue_thumbnail_processing(self, thumbnail_url: str, doc_id: str) -> None:
             self.logger.error(f"Failed to queue thumbnail processing for {doc_id}: {e}")
             # Don't raise - this is a background operation that shouldn't fail the main request
 
-    def _queue_manifest_processing(self, manifest_url: str) -> None:
-        """
-        Queue manifest processing in the background without blocking.
-        This method is fire-and-forget.
-        """
-        try:
-            from app.tasks.worker import fetch_and_cache_image
-
-            task = fetch_and_cache_image.delay(manifest_url)
-            self.logger.info(f"Manifest resolution queued: {task.id}")
-
-        except Exception as e:
-            self.logger.error(f"Failed to queue manifest processing for {manifest_url}: {e}")
-            # Don't raise - this is a background operation that shouldn't fail the main request
-
     async def get_cached_image(self, image_hash: str) -> Optional[bytes]:
         """Retrieve a cached image by its hash."""
         image_key = f"image:{image_hash}"

backend/app/services/thumbnail_state_service.py

@@ -12,6 +12,7 @@
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from app.services.distribution_repository import async_session_factory
+from app.services.iiif_url import is_iiif_manifest_url
 from app.services.thumbnail_alias_service import thumbnail_alias_service
 from db.models import resource_thumbnail_state
 from db.session import sync_engine as _sync_engine
@@ -46,11 +47,7 @@ def infer_source_type(source_url: str | None) -> str | None:
         or "display_raster" in lowered
     ):
         return "cog"
-    if (
-        source_url.endswith(("/iiif3/manifest", "/iiif/manifest", "/manifest", "manifest.json"))
-        or "/manifest" in source_url
-        or (".json" in source_url and ("iiif" in lowered or "/object/" in lowered))
-    ):
+    if is_iiif_manifest_url(source_url):
         return "manifest"
     return "remote"
 

backend/app/tasks/worker.py

@@ -456,16 +456,9 @@ def fetch_and_cache_image(self, url: str, doc_id: Optional[str] = None) -> bool:
 
 def _looks_like_manifest_url(url: str) -> bool:
     """Heuristic to detect IIIF manifest URLs by path patterns."""
-    if not url:
-        return False
-    lowered = url.lower()
-    return (
-        url.endswith(("/iiif3/manifest", "/iiif/manifest", "/manifest", "manifest.json"))
-        or "/manifest" in url
-        or (".json" in url and ("iiif" in lowered or "/object/" in url or "/collection/" in url))
-        or ("/api/" in url and ("iiif" in lowered or "image" in lowered))
-        or "/cgi/i/image/api/" in lowered  # U of Michigan pattern
-    )
+    from app.services.iiif_url import is_iiif_manifest_url
+
+    return is_iiif_manifest_url(url)
 
 
 def _validate_image_content(
@@ -1158,8 +1151,18 @@ def generate_pmtiles_thumbnail(self, pmtiles_url: str, doc_id: Optional[str] = N
 
 
 def _resolve_image_url(url: str) -> str:
-    """Resolve the URL to an actual image URL if given a manifest; otherwise return the original."""
+    """Resolve IIIF metadata URLs to supported image renditions."""
     try:
+        from app.services.iiif_url import is_iiif_info_url
+
+        if is_iiif_info_url(url):
+            from app.services.image_service import ImageService
+
+            service = ImageService({})
+            image_url = service.get_iiif_image_thumbnail(url)
+            if image_url:
+                return image_url
+
         # Only run manifest resolution when it looks like a manifest URL
         if _looks_like_manifest_url(url):
             from app.services.image_service import ImageService