diff --git a/.env.example b/.env.example index 2751ce3..9a44a71 100644 --- a/.env.example +++ b/.env.example @@ -27,6 +27,11 @@ RESOURCE_REPRESENTATION_DURABLE_STORE=database API_RESPONSE_DURABLE_CACHE_STORE=database VISUAL_ASSET_DURABLE_STORE=database VISUAL_ASSET_CACHE_TTL_SECONDS=0 +OGM_THUMBNAIL_REFRESH_ENABLED=true +OGM_THUMBNAIL_REFRESH_BATCH_SIZE=500 +OGM_THUMBNAIL_REFRESH_CONCURRENCY=2 +PDF_THUMBNAIL_MAX_BYTES=33554432 +REMOTE_THUMBNAIL_MAX_BYTES=20971520 # Admin / webhook ADMIN_USERNAME=admin diff --git a/Dockerfile b/Dockerfile index bc0402b..a3eab80 100644 --- a/Dockerfile +++ b/Dockerfile @@ -12,6 +12,7 @@ RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \ libcairo2-dev \ gdal-bin \ libgdal-dev \ + poppler-utils \ curl \ ca-certificates \ cron \ diff --git a/backend/app/api/v1/endpoint_modules/resources/thumbnail.py b/backend/app/api/v1/endpoint_modules/resources/thumbnail.py index 00d0575..aedc056 100644 --- a/backend/app/api/v1/endpoint_modules/resources/thumbnail.py +++ b/backend/app/api/v1/endpoint_modules/resources/thumbnail.py @@ -11,6 +11,7 @@ from sqlalchemy.sql import select from app.api.v1.utils import _get_thumbnail_asset_url, sanitize_for_json +from app.services.access_policy import is_restricted_resource 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 @@ -21,13 +22,16 @@ from app.services.thumbnail_state_service import ( ThumbnailState, ThumbnailStatePayload, + infer_source_type, safe_record_thumbnail_state, ) from app.tasks.worker import ( _generate_cog_thumbnail_bytes, + _generate_pdf_thumbnail_bytes, _generate_pmtiles_thumbnail_bytes, _normalize_thumbnail_image, generate_cog_thumbnail, + generate_pdf_thumbnail, generate_pmtiles_thumbnail, ) from db.models import resources @@ -448,7 +452,7 @@ async def _get_resource_thumbnail_response( raise HTTPException(status_code=404, detail="Resource not found") # Check for restricted access rights - if resource_dict.get("dct_accessrights_s") == "Restricted": + if is_restricted_resource(resource_dict): await safe_record_thumbnail_state( ThumbnailStatePayload( resource_id=id, @@ -499,6 +503,8 @@ async def _get_resource_thumbnail_response( if image_service._is_cog_url(source_url) else "pmtiles" if image_service._is_pmtiles_url(source_url) + else "pdf" + if infer_source_type(source_url) == "pdf" else "manifest" if image_service._is_manifest_url(source_url) else "remote" @@ -536,6 +542,7 @@ async def _get_resource_thumbnail_response( 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 infer_source_type(source_url) != "pdf" and THUMBNAIL_REQUEST_PROBE_ENABLED ): fetch_url = image_service._standardize_iiif_url(source_url) @@ -608,6 +615,31 @@ async def _get_resource_thumbnail_response( state_detail="PMTiles thumbnail generation already queued", ) ) + elif infer_source_type(source_url) == "pdf": + if acquire_thumbnail_queue_slot(id, source_url): + task = generate_pdf_thumbnail.delay(source_url, id) + await safe_record_thumbnail_state( + ThumbnailStatePayload( + resource_id=id, + state=ThumbnailState.QUEUED, + source_type="pdf", + source_url=source_url, + source_hash=image_hash, + queue_task_id=task.id, + state_detail="Queued PDF first-page thumbnail generation", + ) + ) + else: + await safe_record_thumbnail_state( + ThumbnailStatePayload( + resource_id=id, + state=ThumbnailState.QUEUED, + source_type="pdf", + source_url=source_url, + source_hash=image_hash, + state_detail="PDF thumbnail generation already queued", + ) + ) elif image_service._is_manifest_url(source_url) or is_iiif_info_url(source_url): image_service._queue_thumbnail_processing(source_url, id) else: @@ -625,6 +657,8 @@ async def _get_resource_thumbnail_response( if image_service._is_cog_url(source_url) else "pmtiles" if image_service._is_pmtiles_url(source_url) + else "pdf" + if infer_source_type(source_url) == "pdf" else "manifest" if image_service._is_manifest_url(source_url) else "remote" @@ -701,7 +735,7 @@ async def get_resource_thumbnail_no_cache( resource_dict = sanitize_for_json(dict(row._mapping)) - if resource_dict.get("dct_accessrights_s") == "Restricted": + if is_restricted_resource(resource_dict): return _svg_placeholder(title="Thumbnail unavailable", subtitle="Restricted resource") distribution_context = await fetch_distribution_context(id) @@ -745,6 +779,21 @@ async def get_resource_thumbnail_no_cache( ) return await _svg_icon_for_resource(resource_dict, variant=variant) + # For PDF maps: render the first page synchronously for diagnostics. + if infer_source_type(source_url) == "pdf": + image_bytes = await asyncio.to_thread(_generate_pdf_thumbnail_bytes, source_url) + if image_bytes: + normalized_bytes, normalized_type = _normalize_thumbnail_image( + image_bytes, "image/png" + ) + if normalized_bytes and normalized_type: + return Response( + content=normalized_bytes, + media_type=normalized_type, + headers={"Cache-Control": "no-store"}, + ) + return await _svg_icon_for_resource(resource_dict, variant=variant) + # 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) diff --git a/backend/app/services/access_policy.py b/backend/app/services/access_policy.py new file mode 100644 index 0000000..6b7a7e5 --- /dev/null +++ b/backend/app/services/access_policy.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from typing import Any + +ACCESS_RIGHTS_KEYS = ( + "dct_accessRights_s", + "dct_accessrights_s", +) + + +def resource_access_rights(metadata: Mapping[str, Any] | None) -> str | None: + """Return a normalized access-rights value from canonical or legacy field names.""" + if not metadata: + return None + + for key in ACCESS_RIGHTS_KEYS: + value = metadata.get(key) + if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)): + value = next((item for item in value if item is not None), None) + if value is None: + continue + normalized = str(value).strip() + if normalized: + return normalized + return None + + +def is_restricted_resource(metadata: Mapping[str, Any] | None) -> bool: + """Return True when a resource is explicitly marked Restricted.""" + access_rights = resource_access_rights(metadata) + return bool(access_rights and access_rights.casefold() == "restricted") diff --git a/backend/app/services/image_service.py b/backend/app/services/image_service.py index 58fa8b8..e6a9e45 100644 --- a/backend/app/services/image_service.py +++ b/backend/app/services/image_service.py @@ -5,6 +5,7 @@ import os import re from typing import Any, Dict, List, Optional +from urllib.parse import urlsplit import aiohttp import redis @@ -12,6 +13,7 @@ from dotenv import load_dotenv from app.security_utils import url_hostname_matches +from app.services.access_policy import is_restricted_resource from app.services.distribution_repository import ( DistributionContext, build_distribution_context, @@ -43,6 +45,14 @@ REMOTE_THUMBNAIL_PREFIX = f"remote-thumb-normalized:{THUMBNAIL_CACHE_VERSION}:" COG_THUMBNAIL_PREFIX = "cog-thumb:" PMTILES_THUMBNAIL_PREFIX = "pmtiles-thumb:" +PDF_THUMBNAIL_PREFIX = "pdf-thumb:" + +DOWNLOAD_REFERENCE_URIS = ( + "http://schema.org/downloadUrl", + "https://schema.org/downloadUrl", +) +DIRECT_IMAGE_EXTENSIONS = (".jpg", ".jpeg", ".png", ".gif", ".webp") +DIRECT_IMAGE_LABELS = ("jpeg", "jpg", "png", "gif", "webp") # Shared Redis connection pool to avoid creating new connections for each ImageService instance _redis_connection_pool = None @@ -475,6 +485,70 @@ def resolve_thumbnail_source_url( return source_url return self._clean_external_thumbnail_url(thumbnail_asset_url) + def _download_source_candidates( + self, + references: Optional[Dict[str, Any]] = None, + ) -> List[tuple[str, str]]: + """Return normalized ``(url, label)`` candidates from download distributions.""" + candidates: List[tuple[str, str]] = [] + + if references is None: + for uri in DOWNLOAD_REFERENCE_URIS: + for record in self.by_uri.get(uri, []): + url = self._clean_external_thumbnail_url(record.url) + if url: + candidates.append((url, str(record.label or "").strip())) + legacy_references = self._parse_legacy_references() + if legacy_references: + candidates.extend(self._download_source_candidates(legacy_references)) + return candidates + + for uri in DOWNLOAD_REFERENCE_URIS: + raw_value = references.get(uri) + values = raw_value if isinstance(raw_value, list) else [raw_value] + for value in values: + label = "" + if isinstance(value, str): + raw_url = value + elif isinstance(value, dict): + raw_url = value.get("url") or value.get("@id") or value.get("id") + label = str(value.get("label") or value.get("title") or "").strip() + else: + continue + url = self._clean_external_thumbnail_url(raw_url) + if url: + candidates.append((url, label)) + return candidates + + @staticmethod + def _url_path_lower(url: str) -> str: + return urlsplit(url).path.lower() + + def _download_image_source_url( + self, + references: Optional[Dict[str, Any]] = None, + ) -> Optional[str]: + """Return an explicitly image-bearing download URL, if one is advertised.""" + for url, label in self._download_source_candidates(references): + path = self._url_path_lower(url) + normalized_label = label.casefold() + if path.endswith(DIRECT_IMAGE_EXTENSIONS) or any( + token == normalized_label or token in normalized_label.split() + for token in DIRECT_IMAGE_LABELS + ): + return url + return None + + def _download_pdf_source_url( + self, + references: Optional[Dict[str, Any]] = None, + ) -> Optional[str]: + """Return a PDF download URL suitable for first-page thumbnail rendering.""" + for url, label in self._download_source_candidates(references): + if self._url_path_lower(url).endswith(".pdf") or "pdf" in label.casefold().split(): + return url + return None + def thumbnail_image_hash_for_source_sync( self, source_url: str, @@ -490,6 +564,8 @@ 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_pdf_url(source_url): + return hashlib.sha256((PDF_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) @@ -649,7 +725,7 @@ def get_thumbnail_url(self, *, thumbnail_asset_url: Optional[str] = None) -> Opt """ try: # Check for restricted access rights - if self.metadata.get("dct_accessrights_s") == "Restricted": + if is_restricted_resource(self.metadata): self.logger.info("Skipping thumbnail for restricted item") return None @@ -686,7 +762,7 @@ def get_hot_thumbnail_url( over a blocking thumbnail generation path. """ try: - if self.metadata.get("dct_accessrights_s") == "Restricted": + if is_restricted_resource(self.metadata): return None doc_id = self.metadata.get("id") @@ -935,6 +1011,13 @@ def _get_thumbnail_source_url( if url := self._first_url(image_key, references=references): return url + # Some OGM records expose only downloadable derivatives. Use explicit + # image files first, then render the first page of a PDF map when needed. + if download_image_url := self._download_image_source_url(references=references): + return download_image_url + if download_pdf_url := self._download_pdf_source_url(references=references): + return download_pdf_url + # Return None when no thumbnail source is found # This allows the frontend to show a default icon based on resource class # (gbl_resourceClass_sm) @@ -959,6 +1042,12 @@ def _is_pmtiles_url(self, url: str) -> bool: url_lower = url.lower() return url_lower.endswith(".pmtiles") or ".pmtiles?" in url_lower + def _is_pdf_url(self, url: str) -> bool: + """Check whether a source URL points at a PDF download.""" + if not url: + return False + return self._url_path_lower(url).endswith(".pdf") + def _is_manifest_url(self, url: str) -> bool: """Check if URL looks like a IIIF manifest URL.""" return is_iiif_manifest_url(url) diff --git a/backend/app/services/ogm_harvest/harvest.py b/backend/app/services/ogm_harvest/harvest.py index 9d390b7..744f4ae 100644 --- a/backend/app/services/ogm_harvest/harvest.py +++ b/backend/app/services/ogm_harvest/harvest.py @@ -10,6 +10,7 @@ from app.services.ogm_harvest.importer import OGMResourceImporter from app.services.ogm_harvest.repo_sync import OGMRepoSync from app.services.ogm_harvest.repository import OGMHarvestRepository +from app.services.thumbnail_refresh_service import refresh_thumbnail_cache_for_changed_resources logger = logging.getLogger(__name__) @@ -77,6 +78,28 @@ async def harvest_repo( ogm_run_id=run_id, progress_meta={"head_sha": head_sha, "repo_action": sync_result.action}, ) + changed_thumbnail_ids = sorted(importer.changed_thumbnail_resource_ids) + stats["thumbnail_sources_changed"] = len(changed_thumbnail_ids) + if changed_thumbnail_ids: + await repo.update_harvest_run( + ogm_id=run_id, + ogm_stats_json={ + **(stats or {}), + "stage": "thumbnail_refresh", + "updated_at": datetime.utcnow().isoformat() + "Z", + }, + ) + try: + stats[ + "thumbnail_cache_refresh" + ] = await refresh_thumbnail_cache_for_changed_resources(changed_thumbnail_ids) + except Exception as exc: + logger.warning( + "OGM thumbnail cache refresh failed for repo=%s; continuing. err=%s", + repo_name, + exc, + ) + stats["thumbnail_cache_refresh"] = {"enabled": True, "error": str(exc)} await repo.update_harvest_run( ogm_id=run_id, ogm_stats_json={ diff --git a/backend/app/services/ogm_harvest/importer.py b/backend/app/services/ogm_harvest/importer.py index 4593a40..e966638 100644 --- a/backend/app/services/ogm_harvest/importer.py +++ b/backend/app/services/ogm_harvest/importer.py @@ -6,7 +6,7 @@ from datetime import date, datetime, timezone from typing import Any, Dict, Iterable, List, Optional, Set, Tuple -from sqlalchemy import text +from sqlalchemy import select, text from sqlalchemy.dialects.postgresql import insert as pg_insert from app.services.distribution_sync import ( @@ -25,6 +25,16 @@ logger = logging.getLogger(__name__) +THUMBNAIL_SOURCE_FIELDS = ( + "dct_references_s", + "b1g_image_ss", + "dct_accessRights_s", + "gbl_resourceClass_sm", + "gbl_wxsIdentifier_s", + "dcat_bbox", + "locn_geometry", +) + def derive_repo_alias(repo_name: str) -> Optional[str]: parts = [p for p in (repo_name or "").split(".") if p] @@ -101,6 +111,38 @@ class OGMResourceImporter: def __init__(self, repo: Optional[OGMHarvestRepository] = None): self.repo = repo or OGMHarvestRepository() self._resource_columns_cache: Optional[Set[str]] = None + self.changed_thumbnail_resource_ids: Set[str] = set() + + @staticmethod + def _thumbnail_source_signature(row: Dict[str, Any]) -> Tuple[Any, ...]: + def freeze(value: Any) -> Any: + if isinstance(value, list): + return tuple(freeze(item) for item in value) + if isinstance(value, dict): + return tuple(sorted((str(key), freeze(item)) for key, item in value.items())) + return value + + return tuple(freeze(row.get(field)) for field in THUMBNAIL_SOURCE_FIELDS) + + async def _changed_thumbnail_ids(self, rows: List[Dict[str, Any]]) -> Set[str]: + ids = [str(row.get("id")) for row in rows if row.get("id")] + if not ids: + return set() + columns = [resources.c.id] + columns.extend(resources.c[field] for field in THUMBNAIL_SOURCE_FIELDS) + existing_rows = await database.fetch_all(select(*columns).where(resources.c.id.in_(ids))) + existing_by_id = { + str(row["id"]): dict(getattr(row, "_mapping", row)) for row in existing_rows + } + changed: Set[str] = set() + for row in rows: + resource_id = str(row.get("id") or "") + existing = existing_by_id.get(resource_id) + if not existing or self._thumbnail_source_signature( + row + ) != self._thumbnail_source_signature(existing): + changed.add(resource_id) + return changed async def _resource_columns_in_db(self) -> Set[str]: """Return current resources table columns from the connected DB.""" @@ -377,6 +419,7 @@ async def upsert_stream( resources in batches, while also updating ogm_resource_state in batches. """ stats = {"processed": 0, "imported": 0, "skipped": 0, "errors": 0} + self.changed_thumbnail_resource_ids.clear() error_samples: List[Dict[str, Any]] = [] error_signature_counts: Dict[str, int] = {} @@ -451,6 +494,7 @@ async def _flush_rows(rows: List[Dict[str, Any]], seen: List[Dict[str, Any]]) -> if not rows: return 0 try: + changed_thumbnail_ids = await self._changed_thumbnail_ids(rows) stmt = pg_insert(resources).values(rows) update_map = { c.name: stmt.excluded[c.name] for c in upsert_columns if c.name != "id" @@ -473,6 +517,7 @@ async def _flush_rows(rows: List[Dict[str, Any]], seen: List[Dict[str, Any]]) -> str(rel_err), ) await self.repo.upsert_resources_seen_batch(repo_name, seen) + self.changed_thumbnail_resource_ids.update(changed_thumbnail_ids) return len(rows) except Exception as e: if len(rows) == 1: diff --git a/backend/app/services/remote_fetch.py b/backend/app/services/remote_fetch.py new file mode 100644 index 0000000..e7ab632 --- /dev/null +++ b/backend/app/services/remote_fetch.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +import ipaddress +import socket +from dataclasses import dataclass +from typing import Mapping +from urllib.parse import urldefrag, urljoin, urlsplit + +import requests + + +class UnsafeRemoteUrl(ValueError): + """Raised when a metadata-controlled URL could reach a non-public network.""" + + +class RemotePayloadTooLarge(ValueError): + """Raised when a remote response exceeds its configured byte budget.""" + + +@dataclass(frozen=True) +class RemoteBytes: + body: bytes + content_type: str + final_url: str + + +def validate_public_http_url(url: str, *, resolve_dns: bool = True) -> str: + """Validate an HTTP(S) URL and reject local, private, or reserved destinations.""" + cleaned, _fragment = urldefrag(str(url or "").strip()) + parsed = urlsplit(cleaned) + if parsed.scheme.lower() not in {"http", "https"} or not parsed.hostname: + raise UnsafeRemoteUrl("remote URL must use http or https and include a host") + if parsed.username or parsed.password: + raise UnsafeRemoteUrl("remote URL must not contain credentials") + + hostname = parsed.hostname.rstrip(".").casefold() + if hostname == "localhost" or hostname.endswith(".localhost"): + raise UnsafeRemoteUrl("localhost is not an allowed remote thumbnail host") + + try: + literal_address = ipaddress.ip_address(hostname) + except ValueError: + literal_address = None + if literal_address is not None and not literal_address.is_global: + raise UnsafeRemoteUrl("non-public IP addresses are not allowed") + + if resolve_dns and literal_address is None: + try: + addresses = { + ipaddress.ip_address(sockaddr[0]) + for _family, _type, _proto, _canonname, sockaddr in socket.getaddrinfo( + hostname, + parsed.port or (443 if parsed.scheme.lower() == "https" else 80), + type=socket.SOCK_STREAM, + ) + } + except (OSError, ValueError) as exc: + raise UnsafeRemoteUrl( + f"remote thumbnail host could not be resolved: {hostname}" + ) from exc + if not addresses or any(not address.is_global for address in addresses): + raise UnsafeRemoteUrl("remote thumbnail host resolves to a non-public address") + return cleaned + + +def fetch_public_http_bytes( + url: str, + *, + timeout: int, + max_bytes: int, + headers: Mapping[str, str] | None = None, + max_redirects: int = 5, +) -> RemoteBytes: + """Fetch bounded bytes while validating every redirect destination.""" + current_url = validate_public_http_url(url) + request_headers = dict(headers or {}) + + for redirect_count in range(max_redirects + 1): + response = requests.get( + current_url, + timeout=timeout, + headers=request_headers, + allow_redirects=False, + stream=True, + ) + if response.is_redirect or response.is_permanent_redirect: + if redirect_count >= max_redirects: + response.close() + raise requests.TooManyRedirects(f"too many redirects fetching {url}") + location = response.headers.get("Location") + if not location: + response.raise_for_status() + current_url = validate_public_http_url(urljoin(current_url, str(location))) + response.close() + continue + + try: + response.raise_for_status() + except requests.RequestException: + response.close() + raise + content_length = response.headers.get("Content-Length") + if content_length: + try: + declared_length = int(content_length) + except ValueError: + declared_length = 0 + if declared_length > max_bytes: + response.close() + raise RemotePayloadTooLarge(f"remote payload exceeds {max_bytes} bytes") + + chunks: list[bytes] = [] + byte_count = 0 + for chunk in response.iter_content(chunk_size=64 * 1024): + if not chunk: + continue + byte_count += len(chunk) + if byte_count > max_bytes: + response.close() + raise RemotePayloadTooLarge(f"remote payload exceeds {max_bytes} bytes") + chunks.append(chunk) + if not chunks: + body = bytes(response.content or b"") + if len(body) > max_bytes: + response.close() + raise RemotePayloadTooLarge(f"remote payload exceeds {max_bytes} bytes") + chunks.append(body) + response.close() + return RemoteBytes( + body=b"".join(chunks), + content_type=response.headers.get("Content-Type", ""), + final_url=current_url, + ) + + raise requests.TooManyRedirects(f"too many redirects fetching {url}") diff --git a/backend/app/services/thumbnail_refresh_service.py b/backend/app/services/thumbnail_refresh_service.py new file mode 100644 index 0000000..abdeeb1 --- /dev/null +++ b/backend/app/services/thumbnail_refresh_service.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import asyncio +import os +from collections import Counter +from typing import Any, Iterable + +from sqlalchemy import select + +from app.api.v1.utils import sanitize_for_json +from app.services.cache_service import CacheService +from app.services.distribution_repository import async_session_factory +from app.services.resource_representation_cache import delete_resource_representations +from db.models import resources + + +def _dedupe(values: Iterable[str]) -> list[str]: + return list(dict.fromkeys(str(value).strip() for value in values if str(value).strip())) + + +def _positive_env_int(name: str, default: int) -> int: + try: + return max(1, int(os.getenv(name, str(default)))) + except ValueError: + return default + + +def _enabled() -> bool: + return os.getenv("OGM_THUMBNAIL_REFRESH_ENABLED", "true").strip().lower() in { + "1", + "true", + "yes", + "on", + } + + +async def _fetch_resources(resource_ids: list[str]) -> list[dict[str, Any]]: + async with async_session_factory() as session: + result = await session.execute(select(resources).where(resources.c.id.in_(resource_ids))) + return [sanitize_for_json(dict(row._mapping)) for row in result.fetchall()] + + +async def _prime_resources( + resource_dicts: list[dict[str, Any]], + *, + concurrency: int, +) -> dict[str, int]: + from scripts.prime_thumbnail_cache import ( + FALLBACK_ICON_DETAIL, + _prime_thumbnail_with_fallback_for_resource, + ) + + counters: Counter[str] = Counter() + semaphore = asyncio.Semaphore(concurrency) + + async def run_one(resource_dict: dict[str, Any]) -> tuple[str, str, str]: + async with semaphore: + return await _prime_thumbnail_with_fallback_for_resource( + resource_dict, + force=True, + retry_failures=True, + retry_placeheld=True, + ) + + tasks = [asyncio.create_task(run_one(resource_dict)) for resource_dict in resource_dicts] + for task in asyncio.as_completed(tasks): + status, _resource_id, detail = await task + counters[status] += 1 + if FALLBACK_ICON_DETAIL in detail: + counters["fallback-icon"] += 1 + return {"attempted": len(resource_dicts), **dict(counters)} + + +async def refresh_thumbnail_cache_for_changed_resources( + resource_ids: Iterable[str], +) -> dict[str, Any]: + """Prime OGM-owned thumbnails and invalidate representations that embedded old URLs.""" + ids = _dedupe(resource_ids) + if not ids: + return {"enabled": True, "resources": 0, "thumbnails": {"attempted": 0}} + if not _enabled(): + return {"enabled": False, "resources": len(ids)} + + cache = CacheService() + batch_size = _positive_env_int("OGM_THUMBNAIL_REFRESH_BATCH_SIZE", 500) + concurrency = _positive_env_int("OGM_THUMBNAIL_REFRESH_CONCURRENCY", 2) + thumbnail_totals: Counter[str] = Counter() + redis_representations_deleted = 0 + durable_representations_deleted = True + api_responses_deleted = 0 + + for start in range(0, len(ids), batch_size): + resource_id_batch = ids[start : start + batch_size] + delete_stats = await delete_resource_representations( + resource_id_batch, + cache_service=cache, + ) + redis_representations_deleted += int(delete_stats.get("redis_deleted") or 0) + durable_representations_deleted = durable_representations_deleted and bool( + delete_stats.get("durable_deleted", True) + ) + api_responses_deleted += await cache.invalidate_tags( + [f"resource:{resource_id}" for resource_id in resource_id_batch] + ) + resource_dicts = await _fetch_resources(resource_id_batch) + thumbnail_totals.update( + await _prime_resources(resource_dicts, concurrency=concurrency) + ) + + return { + "enabled": True, + "resources": len(ids), + "representations_deleted": { + "durable_deleted": durable_representations_deleted, + "redis_deleted": redis_representations_deleted, + }, + "api_responses_deleted": api_responses_deleted, + "thumbnails": dict(thumbnail_totals), + } diff --git a/backend/app/services/thumbnail_state_service.py b/backend/app/services/thumbnail_state_service.py index 7eb8e22..8e334bc 100644 --- a/backend/app/services/thumbnail_state_service.py +++ b/backend/app/services/thumbnail_state_service.py @@ -47,6 +47,8 @@ def infer_source_type(source_url: str | None) -> str | None: or "display_raster" in lowered ): return "cog" + if urlparse(source_url).path.lower().endswith(".pdf"): + return "pdf" if is_iiif_manifest_url(source_url): return "manifest" return "remote" diff --git a/backend/app/tasks/worker.py b/backend/app/tasks/worker.py index a8f6a44..ed50c90 100644 --- a/backend/app/tasks/worker.py +++ b/backend/app/tasks/worker.py @@ -2,6 +2,10 @@ import io import logging import os +import shutil +import subprocess +import tempfile +from pathlib import Path from typing import Optional, Tuple import redis @@ -11,6 +15,12 @@ from PIL import Image, ImageOps from app.services.provider_throttle import provider_request_slot +from app.services.remote_fetch import ( + RemotePayloadTooLarge, + UnsafeRemoteUrl, + fetch_public_http_bytes, + validate_public_http_url, +) from app.services.thumbnail_queue_service import release_thumbnail_queue_slot from app.services.thumbnail_state_service import ( ThumbnailState, @@ -20,6 +30,7 @@ ) from app.services.visual_asset_cache import ( cache_visual_asset, + durable_visual_asset_enabled, store_durable_visual_asset, store_durable_visual_asset_link, ) @@ -56,6 +67,11 @@ THUMBNAIL_MAX_EDGE = int(os.getenv("THUMBNAIL_MAX_EDGE", "512")) THUMBNAIL_JPEG_QUALITY = int(os.getenv("THUMBNAIL_JPEG_QUALITY", "78")) REMOTE_THUMBNAIL_PREFIX = f"remote-thumb-normalized:{THUMBNAIL_CACHE_VERSION}:" +PDF_THUMBNAIL_PREFIX = "pdf-thumb:" +PDF_THUMBNAIL_MAX_BYTES = int(os.getenv("PDF_THUMBNAIL_MAX_BYTES", str(32 * 1024 * 1024))) +REMOTE_THUMBNAIL_MAX_BYTES = int( + os.getenv("REMOTE_THUMBNAIL_MAX_BYTES", str(20 * 1024 * 1024)) +) # Setup Celery broker_url = os.getenv( @@ -221,6 +237,43 @@ def _normalize_thumbnail_image( return None, None +def _persist_thumbnail_bytes( + image_hash: str, + image_bytes: bytes, + content_type: str, + *, + resource_id: str | None, + asset_kind: str = "thumbnail", +) -> bool: + """Persist OGM-owned bytes durably, with Redis as an optional hot cache.""" + durable_stored = store_durable_visual_asset( + image_hash, + asset_kind=asset_kind, + content_type=content_type, + body=image_bytes, + ) + if durable_stored and resource_id: + store_durable_visual_asset_link( + resource_id, + asset_hash=image_hash, + asset_kind="thumbnail", + source_signature=image_hash, + ) + + redis_stored = False + try: + redis_stored = bool( + cache_visual_asset(redis_client, f"image:{image_hash}", image_bytes) + ) + cache_visual_asset(redis_client, f"image_type:{image_hash}", content_type) + except Exception as exc: + logger.warning("Redis thumbnail hydration failed for %s: %s", image_hash[:12], exc) + + if durable_visual_asset_enabled(): + return durable_stored + return redis_stored + + @celery_app.task(bind=True, name="fetch_and_cache_image") def fetch_and_cache_image(self, url: str, doc_id: Optional[str] = None) -> bool: """ @@ -237,6 +290,7 @@ def fetch_and_cache_image(self, url: str, doc_id: Optional[str] = None) -> bool: source_type = infer_source_type(url) # Determine the actual image URL; handle IIIF manifests by resolving to a thumbnail resolved_url = _resolve_image_url(url) + resolved_url = validate_public_http_url(resolved_url, resolve_dns=False) logger.info(f"Resolved URL: {url} -> {resolved_url}") # Generate cache key based on the resolved image URL (not the original manifest URL) @@ -268,38 +322,22 @@ def fetch_and_cache_image(self, url: str, doc_id: Optional[str] = None) -> bool: fetch_timeout = int(os.getenv("THUMBNAIL_FETCH_TIMEOUT", "30")) headers = {"User-Agent": "BTAA-Geospatial-Data-API/1.0 (https://geo.btaa.org/)"} with provider_request_slot(resolved_url, action="thumbnail fetch") as lease: - response = requests.get(resolved_url, timeout=fetch_timeout, headers=headers) - - # Don't retry non-recoverable bot-block/authorization responses. - # - 401/403: auth - # - 418: common bot-block response (e.g., MSU) - if response.status_code in (401, 403, 418): - logger.warning( - f"Authorization error ({response.status_code}) for {resolved_url}. Not caching." + fetched = fetch_public_http_bytes( + resolved_url, + timeout=fetch_timeout, + max_bytes=REMOTE_THUMBNAIL_MAX_BYTES, + headers=headers, ) - _record_thumbnail_state( - doc_id, - state=ThumbnailState.FAILURE, - source_type=source_type, - source_url=url, - source_hash=source_hash, - queue_task_id=getattr(getattr(self, "request", None), "id", None), - state_detail=f"Non-retryable HTTP status {response.status_code}", - last_error=f"HTTP {response.status_code} from {resolved_url}", - ) - return False - - response.raise_for_status() # Validate that the response is actually an image - content_type = response.headers.get("Content-Type", "") - is_valid, detected_type = _validate_image_content(response.content, content_type) + content_type = fetched.content_type + is_valid, detected_type = _validate_image_content(fetched.body, content_type) if not is_valid: logger.error( f"❌ Invalid image content from {resolved_url}: " f"Content-Type={content_type}, detected_type={detected_type}, " - f"first_bytes={response.content[:100]!r}" + f"first_bytes={fetched.body[:100]!r}" ) # Don't cache invalid content - return False to indicate failure _record_thumbnail_state( @@ -315,7 +353,7 @@ def fetch_and_cache_image(self, url: str, doc_id: Optional[str] = None) -> bool: return False normalized_content, normalized_type = _normalize_thumbnail_image( - response.content, detected_type + fetched.body, detected_type ) if not normalized_content or not normalized_type: logger.error(f"❌ Thumbnail normalization failed for {resolved_url}") @@ -331,31 +369,17 @@ def fetch_and_cache_image(self, url: str, doc_id: Optional[str] = None) -> bool: ) return False - # Cache image if Redis is available; otherwise, skip caching without retry storms - if redis_available: - try: - # Store image content with detected type (prepend type as metadata) - # We'll use a simple format: store content as-is, content type in separate key - cache_visual_asset(redis_client, image_key, normalized_content) - # Store content type metadata separately (optional, for faster lookups) - type_key = f"image_type:{image_key.split(':')[1]}" - cache_visual_asset(redis_client, type_key, normalized_type) - store_durable_visual_asset( - source_hash, - asset_kind="thumbnail", - content_type=normalized_type, - body=normalized_content, - ) - if doc_id: - store_durable_visual_asset_link( - doc_id, - asset_hash=source_hash, - asset_kind="thumbnail", - source_signature=source_hash, - ) + # Durable OGM storage is authoritative; Redis is only a hot cache. + try: + if _persist_thumbnail_bytes( + source_hash, + normalized_content, + normalized_type, + resource_id=doc_id, + ): logger.info( f"✅ Successfully cached normalized thumbnail: {resolved_url} " - f"(type: {normalized_type}, original={len(response.content)} bytes, " + f"(type: {normalized_type}, original={len(fetched.body)} bytes, " f"normalized={len(normalized_content)} bytes)" ) # Note: No need to invalidate search cache - search results always include @@ -374,34 +398,34 @@ def fetch_and_cache_image(self, url: str, doc_id: Optional[str] = None) -> bool: state_detail=detail, ) return True - except Exception as redis_err: - logger.warning( - f"Failed to cache image due to Redis error for {resolved_url}: {redis_err}" - ) - _record_thumbnail_state( - doc_id, - state=ThumbnailState.FAILURE, - source_type=source_type, - source_url=url, - source_hash=source_hash, - queue_task_id=getattr(getattr(self, "request", None), "id", None), - state_detail="Redis cache write failed", - last_error=str(redis_err), - ) - return False - else: - logger.warning(f"Skipping cache store for {resolved_url}: Redis unavailable") - _record_thumbnail_state( - doc_id, - state=ThumbnailState.FAILURE, - source_type=source_type, - source_url=url, - source_hash=source_hash, - queue_task_id=getattr(getattr(self, "request", None), "id", None), - state_detail="Redis unavailable during cache store", - last_error="Redis unavailable during cache store", - ) - return False + except Exception as persist_error: + logger.warning("Failed to persist thumbnail for %s: %s", resolved_url, persist_error) + if not redis_available: + logger.info("Redis was unavailable; durable OGM thumbnail persistence was attempted") + _record_thumbnail_state( + doc_id, + state=ThumbnailState.FAILURE, + source_type=source_type, + source_url=url, + source_hash=source_hash, + queue_task_id=getattr(getattr(self, "request", None), "id", None), + state_detail="Durable thumbnail persistence failed", + last_error="Durable thumbnail persistence failed", + ) + return False + except (RemotePayloadTooLarge, UnsafeRemoteUrl) as source_error: + _record_thumbnail_state( + doc_id, + state=ThumbnailState.FAILURE, + source_type=infer_source_type(url), + source_url=url, + source_hash=None, + queue_task_id=getattr(getattr(self, "request", None), "id", None), + state_detail="Rejected unsafe or oversized thumbnail source", + last_error=str(source_error), + ) + logger.warning("Rejected thumbnail source %s: %s", url, source_error) + return False except requests.RequestException as http_err: # Don't retry non-recoverable bot-block/authorization responses. if isinstance(http_err, requests.HTTPError) and hasattr(http_err.response, "status_code"): @@ -545,6 +569,151 @@ def _validate_image_content( return False, None +def _pdf_thumbnail_image_hash(pdf_url: str) -> str: + """Compute the cache key hash for a first-page PDF thumbnail.""" + return hashlib.sha256((PDF_THUMBNAIL_PREFIX + pdf_url).encode()).hexdigest() + + +def _render_pdf_first_page(pdf_bytes: bytes) -> Optional[bytes]: + """Render the first PDF page to PNG using the bounded Poppler CLI.""" + if not pdf_bytes.startswith(b"%PDF-"): + return None + pdftoppm = shutil.which("pdftoppm") + if not pdftoppm: + logger.error("pdftoppm is unavailable; cannot render PDF thumbnails") + return None + + try: + with tempfile.TemporaryDirectory(prefix="ogm-pdf-thumbnail-") as temp_dir: + temp_path = Path(temp_dir) + pdf_path = temp_path / "source.pdf" + output_prefix = temp_path / "page" + pdf_path.write_bytes(pdf_bytes) + subprocess.run( + [ + pdftoppm, + "-f", + "1", + "-l", + "1", + "-singlefile", + "-scale-to", + "1024", + "-png", + str(pdf_path), + str(output_prefix), + ], + check=True, + capture_output=True, + timeout=60, + ) + output_path = output_prefix.with_suffix(".png") + if output_path.exists(): + return output_path.read_bytes() + except (OSError, subprocess.SubprocessError) as exc: + logger.warning("PDF first-page rendering failed: %s", exc) + return None + + +def _generate_pdf_thumbnail_bytes(pdf_url: str) -> Optional[bytes]: + """Fetch a public, bounded PDF and render its first page to PNG.""" + headers = {"User-Agent": "BTAA-Geospatial-Data-API/1.0 (https://geo.btaa.org/)"} + fetched = fetch_public_http_bytes( + pdf_url, + timeout=int(os.getenv("THUMBNAIL_FETCH_TIMEOUT", "30")), + max_bytes=PDF_THUMBNAIL_MAX_BYTES, + headers=headers, + ) + content_type = fetched.content_type.lower().split(";", 1)[0].strip() + if content_type and content_type not in {"application/pdf", "application/octet-stream"}: + logger.warning("PDF thumbnail source returned %s: %s", content_type, fetched.final_url) + return None + return _render_pdf_first_page(fetched.body) + + +@celery_app.task(bind=True, name="generate_pdf_thumbnail") +def generate_pdf_thumbnail(self, pdf_url: str, doc_id: Optional[str] = None) -> bool: + """Generate, normalize, and durably cache a first-page PDF thumbnail.""" + source_hash = _pdf_thumbnail_image_hash(pdf_url) + image_key = f"image:{source_hash}" + try: + try: + if redis_client.exists(image_key): + _record_thumbnail_state( + doc_id, + state=ThumbnailState.SUCCESS, + source_type="pdf", + source_url=pdf_url, + source_hash=source_hash, + queue_task_id=getattr(getattr(self, "request", None), "id", None), + state_detail="PDF thumbnail already cached", + ) + return True + except Exception as redis_error: + logger.warning("Redis unavailable during PDF thumbnail cache check: %s", redis_error) + + with provider_request_slot(pdf_url, action="PDF thumbnail generation") as lease: + image_bytes = _generate_pdf_thumbnail_bytes(pdf_url) + if not image_bytes: + raise ValueError("PDF first-page render returned no image") + + normalized_bytes, normalized_type = _normalize_thumbnail_image(image_bytes, "image/png") + if not normalized_bytes or not normalized_type: + raise ValueError("PDF first-page thumbnail normalization failed") + + if not _persist_thumbnail_bytes( + source_hash, + normalized_bytes, + normalized_type, + resource_id=doc_id, + asset_kind="thumbnail:pdf", + ): + raise RuntimeError("durable PDF thumbnail persistence failed") + detail = "Cached PDF first-page thumbnail successfully" + if lease.waited_seconds > 0: + detail = f"{detail}; provider pacing waited {lease.waited_seconds:.2f}s" + _record_thumbnail_state( + doc_id, + state=ThumbnailState.SUCCESS, + source_type="pdf", + source_url=pdf_url, + source_hash=source_hash, + queue_task_id=getattr(getattr(self, "request", None), "id", None), + state_detail=detail, + ) + return True + except (RemotePayloadTooLarge, UnsafeRemoteUrl) as source_error: + _record_thumbnail_state( + doc_id, + state=ThumbnailState.FAILURE, + source_type="pdf", + source_url=pdf_url, + source_hash=source_hash, + queue_task_id=getattr(getattr(self, "request", None), "id", None), + state_detail="Rejected unsafe or oversized PDF thumbnail source", + last_error=str(source_error), + ) + logger.warning("Rejected PDF thumbnail source %s: %s", pdf_url, source_error) + return False + except Exception as exc: + if _is_terminal_retry(self): + _record_thumbnail_state( + doc_id, + state=ThumbnailState.FAILURE, + source_type="pdf", + source_url=pdf_url, + source_hash=source_hash, + queue_task_id=getattr(getattr(self, "request", None), "id", None), + state_detail="Exhausted retries for PDF thumbnail generation", + last_error=str(exc), + ) + logger.error("PDF thumbnail generation failed for %s: %s", pdf_url, exc) + self.retry(exc=exc, countdown=60, max_retries=2) + return False + finally: + release_thumbnail_queue_slot(doc_id, pdf_url) + + COG_THUMBNAIL_PREFIX = "cog-thumb:" @@ -680,24 +849,16 @@ def generate_cog_thumbnail(self, cog_url: str, doc_id: Optional[str] = None) -> ) return False - # Cache in Redis + # Persist in OGM storage; Redis hydration is best effort. try: - cache_visual_asset(redis_client, image_key, normalized_bytes) - type_key = f"image_type:{image_hash}" - cache_visual_asset(redis_client, type_key, normalized_type) - store_durable_visual_asset( + if not _persist_thumbnail_bytes( image_hash, + normalized_bytes, + normalized_type, + resource_id=doc_id, asset_kind="thumbnail:cog", - content_type=normalized_type, - body=normalized_bytes, - ) - if doc_id: - store_durable_visual_asset_link( - doc_id, - asset_hash=image_hash, - asset_kind="thumbnail", - source_signature=image_hash, - ) + ): + raise RuntimeError("durable COG thumbnail persistence failed") logger.info( f"Successfully cached COG thumbnail for {cog_url} " f"(type: {normalized_type}, original={len(image_bytes)} bytes, " @@ -1080,22 +1241,14 @@ def generate_pmtiles_thumbnail(self, pmtiles_url: str, doc_id: Optional[str] = N return False try: - cache_visual_asset(redis_client, image_key, normalized_bytes) - type_key = f"image_type:{image_hash}" - cache_visual_asset(redis_client, type_key, normalized_type) - store_durable_visual_asset( + if not _persist_thumbnail_bytes( image_hash, + normalized_bytes, + normalized_type, + resource_id=doc_id, asset_kind="thumbnail:pmtiles", - content_type=normalized_type, - body=normalized_bytes, - ) - if doc_id: - store_durable_visual_asset_link( - doc_id, - asset_hash=image_hash, - asset_kind="thumbnail", - source_signature=image_hash, - ) + ): + raise RuntimeError("durable PMTiles thumbnail persistence failed") logger.info( f"Successfully cached PMTiles thumbnail for {pmtiles_url} " f"(type: {normalized_type}, original={len(image_bytes)} bytes, " diff --git a/backend/scripts/prime_generated_caches.py b/backend/scripts/prime_generated_caches.py index 5b4cdc2..0f02a06 100755 --- a/backend/scripts/prime_generated_caches.py +++ b/backend/scripts/prime_generated_caches.py @@ -31,7 +31,9 @@ logger = logging.getLogger(__name__) -STAGES = ("resources", "thumbnails", "static-maps") +# Thumbnail priming runs before representations so gallery payloads embed the +# final immutable OGM thumbnail/icon URLs instead of transient resolver URLs. +STAGES = ("thumbnails", "static-maps", "resources") def configure_logging(*, verbose: bool = False) -> None: @@ -67,21 +69,10 @@ async def _run(args: argparse.Namespace) -> int: from scripts.prime_thumbnail_cache import _run as prime_thumbnails # noqa: PLC0415 stages = _normalize_stages(args.stage) + resource_class = getattr(args, "resource_class", None) + provider = getattr(args, "provider", None) exit_code = 0 - if "resources" in stages: - logger.info("Priming generated resource representations...") - counters = await prime_resource_representation_cache( - resource_ids=args.resource_ids, - limit=args.limit, - batch_size=max(1, args.resource_batch_size), - concurrency=max(1, args.resource_concurrency), - force=args.force, - ) - _print_resource_summary(counters) - if counters["failed"] and args.strict_failures: - exit_code = max(exit_code, 1) - if "thumbnails" in stages: logger.info("Priming thumbnail generated visual assets...") thumbnail_code = await prime_thumbnails( @@ -96,6 +87,8 @@ async def _run(args: argparse.Namespace) -> int: strict_failures=args.strict_failures, hydrate_assets=args.hydrate_assets, allow_full_hydration=args.allow_full_hydration, + resource_class=resource_class, + provider=provider, ) ) exit_code = max(exit_code, thumbnail_code) @@ -112,10 +105,27 @@ async def _run(args: argparse.Namespace) -> int: hydrate_assets=args.hydrate_assets, allow_full_hydration=args.allow_full_hydration, strict_failures=args.strict_failures, + resource_class=resource_class, + provider=provider, ) ) exit_code = max(exit_code, static_map_code) + if "resources" in stages: + logger.info("Priming generated resource representations...") + counters = await prime_resource_representation_cache( + resource_ids=args.resource_ids, + limit=args.limit, + batch_size=max(1, args.resource_batch_size), + concurrency=max(1, args.resource_concurrency), + force=args.force, + resource_class=resource_class, + provider=provider, + ) + _print_resource_summary(counters) + if counters["failed"] and args.strict_failures: + exit_code = max(exit_code, 1) + print(f"Generated cache priming finished for stages: {', '.join(stages)}") return exit_code @@ -125,6 +135,14 @@ def _parse_args() -> argparse.Namespace: description="Prime generated resource, thumbnail, and static-map caches." ) parser.add_argument("resource_ids", nargs="*", help="Optional explicit resource IDs to prime") + parser.add_argument( + "--resource-class", + help="Limit thumbnail and representation stages to an exact resource class.", + ) + parser.add_argument( + "--provider", + help="Optional exact provider filter; omitted by default.", + ) parser.add_argument( "--stage", action="append", diff --git a/backend/scripts/prime_resource_representation_cache.py b/backend/scripts/prime_resource_representation_cache.py index 0a3cbaf..75a77ff 100644 --- a/backend/scripts/prime_resource_representation_cache.py +++ b/backend/scripts/prime_resource_representation_cache.py @@ -84,12 +84,46 @@ def configure_logging(*, verbose: bool = False) -> None: handler.setLevel(level) -async def _count_resources(resource_ids: list[str], limit: int | None) -> int: - if resource_ids: +def _apply_resource_filters(stmt: Any, *, resource_class: str | None, provider: str | None) -> Any: + if resource_class: + stmt = stmt.where(resources.c.gbl_resourceClass_sm.any(resource_class)) + if provider: + stmt = stmt.where(resources.c.schema_provider_s == provider) + return stmt + + +async def _count_resources( + resource_ids: list[str], + limit: int | None, + *, + resource_class: str | None = None, + provider: str | None = None, +) -> int: + if resource_ids and not resource_class and not provider: return len(resource_ids[:limit] if limit else resource_ids) + if resource_ids: + async with async_session_factory() as session: + stmt = ( + select(func.count()) + .select_from(resources) + .where(resources.c.id.in_(resource_ids)) + ) + stmt = _apply_resource_filters( + stmt, + resource_class=resource_class, + provider=provider, + ) + result = await session.execute(stmt) + total = int(result.scalar_one() or 0) + return min(total, limit) if limit else total async with async_session_factory() as session: stmt = select(func.count()).select_from(resources) + stmt = _apply_resource_filters( + stmt, + resource_class=resource_class, + provider=provider, + ) result = await session.execute(stmt) total = int(result.scalar_one() or 0) return min(total, limit) if limit else total @@ -110,27 +144,45 @@ async def _disconnect_legacy_database(opened: bool) -> None: async def _fetch_resources_by_ids( - resource_ids: list[str], limit: int | None + resource_ids: list[str], + limit: int | None, + *, + resource_class: str | None = None, + provider: str | None = None, ) -> list[dict[str, Any]]: if not resource_ids: return [] ids = resource_ids[:limit] if limit else resource_ids async with async_session_factory() as session: - stmt = select(resources).where(resources.c.id.in_(ids)).order_by(resources.c.id) + stmt = select(resources).where(resources.c.id.in_(ids)) + stmt = _apply_resource_filters( + stmt, + resource_class=resource_class, + provider=provider, + ).order_by(resources.c.id) result = await session.execute(stmt) return [sanitize_for_json(dict(row._mapping)) for row in result.fetchall()] async def _fetch_resource_batch( - last_id: str | None, batch_size: int, remaining: int | None + last_id: str | None, + batch_size: int, + remaining: int | None, + *, + resource_class: str | None = None, + provider: str | None = None, ) -> list[dict[str, Any]]: limit = min(batch_size, remaining) if remaining is not None else batch_size if limit <= 0: return [] async with async_session_factory() as session: - stmt = select(resources).order_by(resources.c.id).limit(limit) + stmt = _apply_resource_filters( + select(resources), + resource_class=resource_class, + provider=provider, + ).order_by(resources.c.id).limit(limit) if last_id is not None: stmt = stmt.where(resources.c.id > last_id) result = await session.execute(stmt) @@ -468,13 +520,20 @@ async def prime_resource_representation_cache( batch_size: int, concurrency: int, force: bool, + resource_class: str | None = None, + provider: str | None = None, ) -> Counter: if not ENDPOINT_CACHE: logger.warning("ENDPOINT_CACHE is false; cache writes will be skipped.") opened_legacy_database = await _connect_legacy_database() try: - total = await _count_resources(resource_ids, limit) + total = await _count_resources( + resource_ids, + limit, + resource_class=resource_class, + provider=provider, + ) counters: Counter = Counter() with tqdm( @@ -483,7 +542,12 @@ async def prime_resource_representation_cache( if resource_ids: selected_ids = resource_ids[:limit] if limit else resource_ids for resource_id_batch in _chunks(selected_ids, batch_size): - batch = await _fetch_resources_by_ids(resource_id_batch, None) + batch = await _fetch_resources_by_ids( + resource_id_batch, + None, + resource_class=resource_class, + provider=provider, + ) counters.update( await _prime_batch( batch, @@ -501,7 +565,13 @@ async def prime_resource_representation_cache( last_id = None remaining = limit while True: - batch = await _fetch_resource_batch(last_id, batch_size, remaining) + batch = await _fetch_resource_batch( + last_id, + batch_size, + remaining, + resource_class=resource_class, + provider=provider, + ) if not batch: break @@ -527,6 +597,8 @@ async def prime_resource_representation_cache( def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Prime resource representation cache entries.") parser.add_argument("resource_ids", nargs="*", help="Optional explicit resource IDs to prime") + parser.add_argument("--resource-class", help="Exact gbl_resourceClass_sm value to include") + parser.add_argument("--provider", help="Optional exact schema_provider_s value to include") parser.add_argument("--limit", type=int, help="Maximum number of resources to prime") parser.add_argument("--batch-size", type=int, default=500, help="Database batch size") parser.add_argument("--concurrency", type=int, default=16, help="Concurrent resource builders") @@ -545,6 +617,8 @@ def main() -> int: batch_size=max(1, args.batch_size), concurrency=max(1, args.concurrency), force=args.force, + resource_class=getattr(args, "resource_class", None), + provider=getattr(args, "provider", None), ) ) print( diff --git a/backend/scripts/prime_static_map_cache.py b/backend/scripts/prime_static_map_cache.py index ce4aacb..2660a7a 100644 --- a/backend/scripts/prime_static_map_cache.py +++ b/backend/scripts/prime_static_map_cache.py @@ -46,7 +46,25 @@ logger = logging.getLogger(__name__) -async def _count_resources(resource_ids: list[str]) -> int: +def _apply_resource_filters( + stmt: Any, + *, + resource_class: str | None, + provider: str | None, +) -> Any: + if resource_class: + stmt = stmt.where(resources.c.gbl_resourceClass_sm.any(resource_class)) + if provider: + stmt = stmt.where(resources.c.schema_provider_s == provider) + return stmt + + +async def _count_resources( + resource_ids: list[str], + *, + resource_class: str | None = None, + provider: str | None = None, +) -> int: async with async_session_factory() as session: if resource_ids: stmt = ( @@ -54,31 +72,52 @@ async def _count_resources(resource_ids: list[str]) -> int: ) else: stmt = select(func.count()).select_from(resources) + stmt = _apply_resource_filters( + stmt, + resource_class=resource_class, + provider=provider, + ) result = await session.execute(stmt) return int(result.scalar_one() or 0) -async def _fetch_resources_by_ids(resource_ids: list[str]) -> list[dict[str, Any]]: +async def _fetch_resources_by_ids( + resource_ids: list[str], + *, + resource_class: str | None = None, + provider: str | None = None, +) -> list[dict[str, Any]]: if not resource_ids: return [] async with async_session_factory() as session: - stmt = ( - select(resources.c.id, resources.c.locn_geometry, resources.c.dcat_bbox) - .where(resources.c.id.in_(resource_ids)) - .order_by(resources.c.id) - ) + stmt = select( + resources.c.id, + resources.c.locn_geometry, + resources.c.dcat_bbox, + ).where(resources.c.id.in_(resource_ids)) + stmt = _apply_resource_filters( + stmt, + resource_class=resource_class, + provider=provider, + ).order_by(resources.c.id) result = await session.execute(stmt) return [dict(row._mapping) for row in result.fetchall()] -async def _fetch_resource_batch(last_id: str | None, batch_size: int) -> list[dict[str, Any]]: +async def _fetch_resource_batch( + last_id: str | None, + batch_size: int, + *, + resource_class: str | None = None, + provider: str | None = None, +) -> list[dict[str, Any]]: async with async_session_factory() as session: - stmt = ( - select(resources.c.id, resources.c.locn_geometry, resources.c.dcat_bbox) - .order_by(resources.c.id) - .limit(batch_size) - ) + stmt = _apply_resource_filters( + select(resources.c.id, resources.c.locn_geometry, resources.c.dcat_bbox), + resource_class=resource_class, + provider=provider, + ).order_by(resources.c.id).limit(batch_size) if last_id is not None: stmt = stmt.where(resources.c.id > last_id) result = await session.execute(stmt) @@ -225,7 +264,13 @@ async def _run(resource_dict: dict[str, Any]) -> tuple[str, str, str]: async def _run(args: argparse.Namespace) -> int: resource_ids = args.resource_ids - total = len(resource_ids) if resource_ids else await _count_resources(resource_ids) + resource_class = getattr(args, "resource_class", None) + provider = getattr(args, "provider", None) + total = await _count_resources( + resource_ids, + resource_class=resource_class, + provider=provider, + ) if args.limit is not None: total = min(total, args.limit) @@ -266,7 +311,11 @@ async def _run(args: argparse.Namespace) -> int: try: if resource_ids: - remaining = await _fetch_resources_by_ids(resource_ids) + remaining = await _fetch_resources_by_ids( + resource_ids, + resource_class=resource_class, + provider=provider, + ) if args.limit is not None: remaining = remaining[: args.limit] for start in range(0, len(remaining), args.batch_size): @@ -285,7 +334,12 @@ async def _run(args: argparse.Namespace) -> int: processed = 0 while processed < total: batch_size = min(args.batch_size, total - processed) - batch = await _fetch_resource_batch(last_id, batch_size) + batch = await _fetch_resource_batch( + last_id, + batch_size, + resource_class=resource_class, + provider=provider, + ) if not batch: break await _process_batch( @@ -323,6 +377,8 @@ async def _run(args: argparse.Namespace) -> int: def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Prime static-map and basemap cache entries.") parser.add_argument("resource_ids", nargs="*", help="Optional explicit resource IDs to prime") + parser.add_argument("--resource-class", help="Exact gbl_resourceClass_sm value to include") + parser.add_argument("--provider", help="Optional exact schema_provider_s value to include") parser.add_argument("--limit", type=int, default=None, help="Limit number of resources") parser.add_argument( "--batch-size", type=int, default=100, help="Database batch size for resource fetches" diff --git a/backend/scripts/prime_thumbnail_cache.py b/backend/scripts/prime_thumbnail_cache.py index 7f04fdc..624312c 100644 --- a/backend/scripts/prime_thumbnail_cache.py +++ b/backend/scripts/prime_thumbnail_cache.py @@ -13,6 +13,7 @@ Examples: python scripts/prime_thumbnail_cache.py + python scripts/prime_thumbnail_cache.py --resource-class Maps python scripts/prime_thumbnail_cache.py --limit 250 --concurrency 4 python scripts/prime_thumbnail_cache.py --force b1g_PJxxfKgpqpUT b1g_abc123 """ @@ -44,6 +45,7 @@ os.environ.setdefault("VISUAL_ASSET_REDIS_LOADING_RETRY_SECONDS", "5") from app.api.v1.utils import _get_thumbnail_asset_url, sanitize_for_json # noqa: E402 +from app.services.access_policy import is_restricted_resource # noqa: E402 from app.services.distribution_repository import ( # noqa: E402 async_session_factory, fetch_distribution_context, @@ -55,6 +57,10 @@ record_provider_failure, record_provider_success, ) +from app.services.remote_fetch import ( # noqa: E402 + fetch_public_http_bytes, + validate_public_http_url, +) from app.services.thumbnail_state_service import ( # noqa: E402 ThumbnailState, ThumbnailStatePayload, @@ -63,11 +69,13 @@ ) from app.services.visual_asset_cache import ( # noqa: E402 cache_visual_asset, + durable_visual_asset_enabled, store_durable_visual_asset, store_durable_visual_asset_link, ) from app.tasks.worker import ( # noqa: E402 _generate_cog_thumbnail_bytes, + _generate_pdf_thumbnail_bytes, _generate_pmtiles_thumbnail_bytes, _resolve_image_url, _validate_image_content, @@ -79,6 +87,10 @@ logger = logging.getLogger(__name__) USER_AGENT = "BTAA-Geospatial-Data-API/1.0 (https://geo.btaa.org/)" +FALLBACK_ICON_DETAIL = "OGM resource-class fallback icon materialized" +REMOTE_THUMBNAIL_MAX_BYTES = int( + os.getenv("REMOTE_THUMBNAIL_MAX_BYTES", str(20 * 1024 * 1024)) +) def _thumbnail_fetch_timeout() -> int: @@ -100,27 +112,27 @@ def _store_image_bytes( *, resource_id: str | None = None, ) -> bool: - """Store image bytes and MIME metadata in Redis.""" - try: - cache_visual_asset(redis_client, f"image:{image_hash}", image_bytes) - cache_visual_asset(redis_client, f"image_type:{image_hash}", content_type) - store_durable_visual_asset( - image_hash, + """Store OGM-owned image bytes durably and optionally hydrate Redis.""" + durable_stored = store_durable_visual_asset( + image_hash, + asset_kind="thumbnail", + content_type=content_type, + body=image_bytes, + ) + if durable_stored and resource_id: + store_durable_visual_asset_link( + resource_id, + asset_hash=image_hash, asset_kind="thumbnail", - content_type=content_type, - body=image_bytes, + source_signature=image_hash, ) - if resource_id: - store_durable_visual_asset_link( - resource_id, - asset_hash=image_hash, - asset_kind="thumbnail", - source_signature=image_hash, - ) - return True + redis_stored = False + try: + redis_stored = bool(cache_visual_asset(redis_client, f"image:{image_hash}", image_bytes)) + cache_visual_asset(redis_client, f"image_type:{image_hash}", content_type) except Exception as exc: - logger.warning("Failed to cache thumbnail %s: %s", image_hash[:12], exc) - return False + logger.warning("Failed to hydrate Redis thumbnail %s: %s", image_hash[:12], exc) + return durable_stored if durable_visual_asset_enabled() else redis_stored def _set_pmtiles_skip_marker(image_hash: str) -> bool: @@ -149,6 +161,22 @@ def _prime_cog_thumbnail( return _store_image_bytes(image_hash, image_bytes, "image/png", resource_id=resource_id) +def _prime_pdf_thumbnail( + source_url: str, + image_hash: str, + *, + resource_id: str | None = None, +) -> bool: + with provider_request_slot(source_url, action="thumbnail prime (PDF)"): + image_bytes = _generate_pdf_thumbnail_bytes(source_url) + if not image_bytes or len(image_bytes) < 100: + return False + is_valid, _ = _validate_image_content(image_bytes, "image/png") + if not is_valid: + return False + return _store_image_bytes(image_hash, image_bytes, "image/png", resource_id=resource_id) + + def _prime_pmtiles_thumbnail( source_url: str, image_hash: str, @@ -187,7 +215,7 @@ def _prime_remote_thumbnail( *, resource_id: str | None = None, ) -> tuple[str, str]: - resolved_url = _resolve_image_url(source_url) + resolved_url = validate_public_http_url(_resolve_image_url(source_url), resolve_dns=False) cooldown_remaining = provider_origin_cooldown_remaining(resolved_url) if cooldown_remaining > 0: return ( @@ -199,8 +227,11 @@ def _prime_remote_thumbnail( started = time.monotonic() try: with provider_request_slot(resolved_url, action="thumbnail prime (remote)"): - response = requests.get( - resolved_url, timeout=_thumbnail_fetch_timeout(), headers=headers + fetched = fetch_public_http_bytes( + resolved_url, + timeout=_thumbnail_fetch_timeout(), + max_bytes=REMOTE_THUMBNAIL_MAX_BYTES, + headers=headers, ) except requests.Timeout: elapsed = time.monotonic() - started @@ -232,50 +263,19 @@ def _prime_remote_thumbnail( ) return ("failed", f"{exc} ({resolved_url})") - if response.status_code in (401, 403, 418): - logger.warning( - "Authorization/bot-block status %s for %s", response.status_code, resolved_url - ) - record_provider_failure( - resolved_url, - elapsed_seconds=time.monotonic() - started, - failure_type=f"http_{response.status_code}", - status_code=response.status_code, - ) - return ("failed", f"HTTP {response.status_code} ({resolved_url})") - - try: - response.raise_for_status() - except requests.RequestException as exc: - elapsed = time.monotonic() - started - cooldown_seconds = record_provider_failure( - resolved_url, - elapsed_seconds=elapsed, - failure_type="request_error", - status_code=response.status_code, - ) - if cooldown_seconds > 0: - return ( - "deprioritized", - f"provider HTTP failure; cooling down for {cooldown_seconds:.0f}s ({resolved_url})", - ) - return ("failed", f"{exc} ({resolved_url})") - - is_valid, detected_type = _validate_image_content( - response.content, response.headers.get("Content-Type") - ) + is_valid, detected_type = _validate_image_content(fetched.body, fetched.content_type) if not is_valid: record_provider_failure( resolved_url, elapsed_seconds=time.monotonic() - started, failure_type="invalid_content", - status_code=response.status_code, + status_code=None, ) return ("failed", f"invalid image content ({resolved_url})") if _store_image_bytes( image_hash, - response.content, + fetched.body, detected_type or "image/jpeg", resource_id=resource_id, ): @@ -285,7 +285,25 @@ def _prime_remote_thumbnail( return ("failed", f"failed to cache thumbnail ({resolved_url})") -async def _count_resources(resource_ids: list[str]) -> int: +def _apply_resource_filters( + stmt: Any, + *, + resource_class: str | None, + provider: str | None, +) -> Any: + if resource_class: + stmt = stmt.where(resources.c.gbl_resourceClass_sm.any(resource_class)) + if provider: + stmt = stmt.where(resources.c.schema_provider_s == provider) + return stmt + + +async def _count_resources( + resource_ids: list[str], + *, + resource_class: str | None = None, + provider: str | None = None, +) -> int: async with async_session_factory() as session: if resource_ids: stmt = ( @@ -293,23 +311,52 @@ async def _count_resources(resource_ids: list[str]) -> int: ) else: stmt = select(func.count()).select_from(resources) + stmt = _apply_resource_filters( + stmt, + resource_class=resource_class, + provider=provider, + ) result = await session.execute(stmt) return int(result.scalar_one() or 0) -async def _fetch_resources_by_ids(resource_ids: list[str]) -> list[dict[str, Any]]: +async def _fetch_resources_by_ids( + resource_ids: list[str], + *, + resource_class: str | None = None, + provider: str | None = None, +) -> list[dict[str, Any]]: if not resource_ids: return [] async with async_session_factory() as session: - stmt = select(resources).where(resources.c.id.in_(resource_ids)).order_by(resources.c.id) + stmt = select(resources).where(resources.c.id.in_(resource_ids)) + stmt = _apply_resource_filters( + stmt, + resource_class=resource_class, + provider=provider, + ).order_by(resources.c.id) result = await session.execute(stmt) return [sanitize_for_json(dict(row._mapping)) for row in result.fetchall()] -async def _fetch_resource_batch(last_id: str | None, batch_size: int) -> list[dict[str, Any]]: +async def _fetch_resource_batch( + last_id: str | None, + batch_size: int, + *, + resource_class: str | None = None, + provider: str | None = None, +) -> list[dict[str, Any]]: async with async_session_factory() as session: - stmt = select(resources).order_by(resources.c.id).limit(batch_size) + stmt = ( + _apply_resource_filters( + select(resources), + resource_class=resource_class, + provider=provider, + ) + .order_by(resources.c.id) + .limit(batch_size) + ) if last_id is not None: stmt = stmt.where(resources.c.id > last_id) result = await session.execute(stmt) @@ -331,6 +378,44 @@ async def _fetch_thumbnail_states(resource_ids: list[str]) -> dict[str, dict[str } +async def _prime_resource_class_icon( + resource_dict: dict[str, Any], + *, + force: bool, +) -> str | None: + """Materialize an OGM-owned immutable icon for a record with no preview source.""" + from app.api.v1.endpoint_modules.resources.thumbnail import ( + _resource_class_icon_signature, + _svg_icon_bytes_for_resource, + ) + from app.services.static_map_service import StaticMapService + + resource_id = str(resource_dict.get("id") or "") + if not resource_id: + return None + service = StaticMapService() + signature = _resource_class_icon_signature(resource_dict, variant="icon-basemap") + if not force: + cached_hash = await asyncio.to_thread( + service.materialize_cached_variant_sync, + resource_id, + variant="resource-class-icon", + source_signature=signature, + hydrate_asset=False, + ) + if cached_hash: + return cached_hash + + svg_bytes = await _svg_icon_bytes_for_resource(resource_dict, variant="icon-basemap") + return await service.materialize_asset( + resource_id, + variant="resource-class-icon", + map_bytes=svg_bytes, + source_signature=signature, + hydrate_asset=False, + ) + + def _should_resume_skip( existing_state: dict[str, Any] | None, *, @@ -372,7 +457,7 @@ async def _prime_thumbnail_for_resource( if should_skip: return ("skipped-resume", resource_id, skip_reason) - if resource_dict.get("dct_accessrights_s") == "Restricted": + if is_restricted_resource(resource_dict): return ("skipped-restricted", resource_id, "restricted") distribution_context = await fetch_distribution_context(resource_id) @@ -382,15 +467,22 @@ async def _prime_thumbnail_for_resource( ) if not source_url: + icon_hash = await _prime_resource_class_icon(resource_dict, force=force) await safe_record_thumbnail_state( ThumbnailStatePayload( resource_id=resource_id, state=ThumbnailState.PLACEHELD, source_type=None, source_url=None, - state_detail="No thumbnail source available during prime run", + state_detail=( + "No preview source; OGM resource-class icon materialized" + if icon_hash + else "No thumbnail source available during prime run" + ), ) ) + if icon_hash: + return ("generated-icon", resource_id, "resource-class icon") return ("skipped-no-source", resource_id, "no thumbnail source") try: @@ -514,6 +606,38 @@ async def _prime_thumbnail_for_resource( ) return ("failed", resource_id, "pmtiles") + if infer_source_type(source_url) == "pdf": + ok = await asyncio.to_thread( + _prime_pdf_thumbnail, + source_url, + image_hash, + resource_id=resource_id, + ) + await safe_record_thumbnail_state( + ThumbnailStatePayload( + resource_id=resource_id, + state=ThumbnailState.SUCCESS if ok else ThumbnailState.FAILURE, + source_type="pdf", + source_url=source_url, + source_hash=image_hash, + state_detail=( + "PDF first-page thumbnail primed" + if ok + else "PDF first-page thumbnail prime failed" + ), + last_error=None if ok else "PDF first-page thumbnail prime failed", + ) + ) + return ( + ("generated", resource_id, "pdf") + if ok + else ( + "failed", + resource_id, + "pdf", + ) + ) + remote_status, remote_detail = await asyncio.to_thread( _prime_remote_thumbnail, image_hash, @@ -552,6 +676,32 @@ async def _prime_thumbnail_for_resource( return ("failed", resource_id, str(exc)) +async def _prime_thumbnail_with_fallback_for_resource( + resource_dict: dict[str, Any], + *, + force: bool, + retry_failures: bool = False, + retry_placeheld: bool = False, + existing_state: dict[str, Any] | None = None, +) -> tuple[str, str, str]: + """Prime a preview and ensure a durable local icon for every eligible miss.""" + result = await _prime_thumbnail_for_resource( + resource_dict, + force=force, + retry_failures=retry_failures, + retry_placeheld=retry_placeheld, + existing_state=existing_state, + ) + status, resource_id, detail = result + if status in {"generated", "cached", "generated-icon", "skipped-restricted"}: + return result + + icon_hash = await _prime_resource_class_icon(resource_dict, force=force) + if not icon_hash: + return result + return (status, resource_id, f"{detail}; {FALLBACK_ICON_DETAIL}") + + async def _process_batch( batch: list[dict[str, Any]], *, @@ -568,7 +718,7 @@ async def _process_batch( async def _run(resource_dict: dict[str, Any]) -> tuple[str, str, str]: async with semaphore: - return await _prime_thumbnail_for_resource( + return await _prime_thumbnail_with_fallback_for_resource( resource_dict, force=force, retry_failures=retry_failures, @@ -581,11 +731,14 @@ async def _run(resource_dict: dict[str, Any]) -> tuple[str, str, str]: for future in asyncio.as_completed(tasks): status, resource_id, detail = await future counters[status] += 1 + if FALLBACK_ICON_DETAIL in detail: + counters["fallback-icon"] += 1 if status == "failed": failures.append(f"{resource_id}: {detail}") progress.update(1) progress.set_postfix( generated=counters["generated"] + counters["generated-skip"], + icons=counters["generated-icon"] + counters["fallback-icon"], cached=counters["cached"] + counters["cached-skip"], skipped=( counters["skipped-no-source"] @@ -599,7 +752,13 @@ async def _run(resource_dict: dict[str, Any]) -> tuple[str, str, str]: async def _run(args: argparse.Namespace) -> int: resource_ids = args.resource_ids - total = len(resource_ids) if resource_ids else await _count_resources(resource_ids) + resource_class = getattr(args, "resource_class", None) + provider = getattr(args, "provider", None) + total = await _count_resources( + resource_ids, + resource_class=resource_class, + provider=provider, + ) if args.limit is not None: total = min(total, args.limit) @@ -619,7 +778,11 @@ async def _run(args: argparse.Namespace) -> int: try: if resource_ids: - remaining = await _fetch_resources_by_ids(resource_ids) + remaining = await _fetch_resources_by_ids( + resource_ids, + resource_class=resource_class, + provider=provider, + ) if args.limit is not None: remaining = remaining[: args.limit] for start in range(0, len(remaining), args.batch_size): @@ -639,7 +802,12 @@ async def _run(args: argparse.Namespace) -> int: processed = 0 while processed < total: batch_size = min(args.batch_size, total - processed) - batch = await _fetch_resource_batch(last_id, batch_size) + batch = await _fetch_resource_batch( + last_id, + batch_size, + resource_class=resource_class, + provider=provider, + ) if not batch: break await _process_batch( @@ -659,11 +827,14 @@ async def _run(args: argparse.Namespace) -> int: logger.info( "Thumbnail priming complete: generated=%s generated_skip=%s cached=%s cached_skip=%s " - "skipped_no_source=%s skipped_restricted=%s skipped_resume=%s deprioritized=%s failed=%s", + "generated_icon=%s fallback_icon=%s skipped_no_source=%s skipped_restricted=%s " + "skipped_resume=%s deprioritized=%s failed=%s", counters["generated"], counters["generated-skip"], counters["cached"], counters["cached-skip"], + counters["generated-icon"], + counters["fallback-icon"], counters["skipped-no-source"], counters["skipped-restricted"], counters["skipped-resume"], @@ -684,6 +855,16 @@ async def _run(args: argparse.Namespace) -> int: def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Prime thumbnail cache entries.") parser.add_argument("resource_ids", nargs="*", help="Optional explicit resource IDs to prime") + parser.add_argument( + "--resource-class", + default=None, + help="Limit priming to an exact gbl_resourceClass_sm value (for example, Maps).", + ) + parser.add_argument( + "--provider", + default=None, + help="Optionally limit priming to an exact schema_provider_s value.", + ) parser.add_argument("--limit", type=int, default=None, help="Limit number of resources") parser.add_argument( "--batch-size", type=int, default=100, help="Database batch size for resource fetches" diff --git a/backend/scripts/report_thumbnail_completeness.py b/backend/scripts/report_thumbnail_completeness.py index 5d49ce2..109a783 100644 --- a/backend/scripts/report_thumbnail_completeness.py +++ b/backend/scripts/report_thumbnail_completeness.py @@ -26,8 +26,10 @@ "cog": 5, "pmtiles": 6, "schema_image": 7, - "bridge_asset": 8, - "no_source": 9, + "download_image": 8, + "download_pdf": 9, + "bridge_asset": 10, + "no_source": 11, } OUTCOME_COLUMNS = ( @@ -47,6 +49,8 @@ def _sync_database_url() -> str: def _scope_condition(scope: str) -> str: + if scope == "maps": + return "'Maps' = ANY(COALESCE(r.\"gbl_resourceClass_sm\", ARRAY[]::varchar[]))" if scope == "urban": return "'b1g_urbanBaseLayers' = ANY(COALESCE(r.\"pcdm_memberOf_sm\", ARRAY[]::varchar[]))" if scope == "iiif": @@ -54,8 +58,17 @@ def _scope_condition(scope: str) -> str: return "TRUE" -def _classified_cte(scope: str) -> str: - return f""" +def _source_condition(source_bucket: str | None) -> str: + if source_bucket is None: + return "TRUE" + if source_bucket not in SOURCE_ORDER: + raise ValueError(f"Unknown thumbnail source bucket: {source_bucket}") + return f"source_bucket = '{source_bucket}'" + + +def _classified_cte(scope: str, *, provider: str | None = None) -> str: + provider_condition = "AND r.schema_provider_s = :provider" if provider else "" + return rf""" WITH dist AS ( SELECT rd.resource_id, @@ -70,7 +83,7 @@ def _classified_cte(scope: str) -> str: ) AS has_pmtiles, BOOL_OR( COALESCE(dt.distribution_uri, '') = 'https://github.com/cogeotiff/cog-spec' - OR COALESCE(rd.url, '') ~* '\\.tiff?(\\?|$)' + OR COALESCE(rd.url, '') ~* '\.tiff?(\?|$)' OR COALESCE(rd.url, '') ILIKE '%geotiff%' OR COALESCE(rd.url, '') ILIKE '%display_raster%' ) AS has_cog, @@ -79,7 +92,28 @@ def _classified_cte(scope: str) -> str: OR COALESCE(dt.distribution_uri, '') ILIKE '%/ogc/wms%' OR COALESCE(dt.distribution_uri, '') ILIKE '%/ogc/tms%' OR COALESCE(rd.url, '') ILIKE '%/arcgis/rest/services/%' - ) AS has_service + ) AS has_service, + BOOL_OR( + dt.distribution_uri IN ( + 'http://schema.org/downloadUrl', + 'https://schema.org/downloadUrl' + ) + AND ( + COALESCE(rd.url, '') ~* '\.(jpe?g|png|gif|webp)(\?|$)' + OR COALESCE(rd.label, '') ~* + '(^|[[:space:]])(jpe?g|png|gif|webp)([[:space:]]|$)' + ) + ) AS has_download_image, + BOOL_OR( + dt.distribution_uri IN ( + 'http://schema.org/downloadUrl', + 'https://schema.org/downloadUrl' + ) + AND ( + COALESCE(rd.url, '') ~* '\.pdf(\?|$)' + OR COALESCE(rd.label, '') ~* '(^|[[:space:]])pdf([[:space:]]|$)' + ) + ) AS has_download_pdf FROM resource_distributions rd LEFT JOIN distribution_types dt ON dt.id = rd.distribution_type_id GROUP BY rd.resource_id @@ -93,6 +127,19 @@ def _classified_cte(scope: str) -> str: AND NULLIF(BTRIM(file_url), '') IS NOT NULL GROUP BY resource_id ), +icon AS ( + SELECT + links.resource_id, + BOOL_OR( + visuals.byte_size > 0 + AND visuals.content_type LIKE 'image/%' + ) AS has_durable_icon + FROM generated_visual_asset_links links + JOIN generated_visual_assets visuals + ON visuals.asset_hash = links.asset_hash + WHERE links.asset_kind = 'resource-class-icon' + GROUP BY links.resource_id +), base AS ( SELECT r.id, @@ -104,7 +151,10 @@ def _classified_cte(scope: str) -> str: COALESCE(d.has_pmtiles, false) AS has_pmtiles, COALESCE(d.has_cog, false) AS has_cog, COALESCE(d.has_service, false) AS has_service, + COALESCE(d.has_download_image, false) AS has_download_image, + COALESCE(d.has_download_pdf, false) AS has_download_pdf, COALESCE(asset.has_bridge_asset, false) AS has_bridge_asset, + COALESCE(icon.has_durable_icon, false) AS has_durable_icon, s.state, s.source_type, s.source_url, @@ -117,11 +167,13 @@ def _classified_cte(scope: str) -> str: FROM resources r LEFT JOIN dist d ON d.resource_id = r.id LEFT JOIN asset ON asset.resource_id = r.id + LEFT JOIN icon ON icon.resource_id = r.id LEFT JOIN resource_thumbnail_state s ON s.resource_id = r.id LEFT JOIN generated_visual_assets gva ON gva.asset_hash = s.source_hash - AND gva.asset_kind = 'thumbnail' + AND gva.asset_kind LIKE 'thumbnail%' WHERE {_scope_condition(scope)} + {provider_condition} ), sourced AS ( SELECT @@ -137,11 +189,13 @@ def _classified_cte(scope: str) -> str: OR refs ILIKE '%/arcgis/rest/services/%' THEN 'service' WHEN has_cog OR refs ILIKE '%cogeotiff%' - OR refs ~* '\\.tiff?(\\?|")' THEN 'cog' + OR refs ~* '\.tiff?(\?|")' THEN 'cog' WHEN has_pmtiles OR refs ILIKE '%PMTiles%' OR refs ILIKE '%.pmtiles%' THEN 'pmtiles' WHEN refs ILIKE '%schema.org/image%' THEN 'schema_image' + WHEN has_download_image THEN 'download_image' + WHEN has_download_pdf THEN 'download_pdf' WHEN has_bridge_asset THEN 'bridge_asset' ELSE 'no_source' END AS source_bucket @@ -151,7 +205,7 @@ def _classified_cte(scope: str) -> str: SELECT *, CASE - WHEN LOWER(access_rights) = 'restricted' THEN 'restricted' + WHEN LOWER(BTRIM(access_rights)) = 'restricted' THEN 'restricted' WHEN state = 'success' AND has_durable_thumbnail THEN 'success' WHEN state = 'success' THEN 'stale_success' WHEN state = 'placeheld' THEN 'placeheld' @@ -165,19 +219,30 @@ def _classified_cte(scope: str) -> str: """ -def _summary_sql(scope: str) -> str: +def _summary_sql( + scope: str, + *, + provider: str | None = None, + source_bucket: str | None = None, +) -> str: outcome_counts = ",\n ".join( f"COUNT(*) FILTER (WHERE outcome = '{column}')::bigint AS {column}" for column in OUTCOME_COLUMNS ) return ( - _classified_cte(scope) + _classified_cte(scope, provider=provider) + f""" SELECT source_bucket, COUNT(*)::bigint AS total, + COUNT(*) FILTER (WHERE outcome <> 'restricted')::bigint AS eligible, + COUNT(*) FILTER ( + WHERE outcome = 'success' + OR (outcome <> 'restricted' AND has_durable_icon) + )::bigint AS gallery_ready, {outcome_counts} FROM classified +WHERE {_source_condition(source_bucket)} GROUP BY source_bucket ORDER BY CASE source_bucket @@ -188,18 +253,25 @@ def _summary_sql(scope: str) -> str: WHEN 'cog' THEN 5 WHEN 'pmtiles' THEN 6 WHEN 'schema_image' THEN 7 - WHEN 'bridge_asset' THEN 8 - ELSE 9 + WHEN 'download_image' THEN 8 + WHEN 'download_pdf' THEN 9 + WHEN 'bridge_asset' THEN 10 + ELSE 11 END, source_bucket; """ ) -def _missing_sql(scope: str) -> str: +def _missing_sql( + scope: str, + *, + provider: str | None = None, + source_bucket: str | None = None, +) -> str: return ( - _classified_cte(scope) - + """ + _classified_cte(scope, provider=provider) + + f""" SELECT id, dct_title_s, @@ -210,6 +282,7 @@ def _missing_sql(scope: str) -> str: source_hash FROM classified WHERE outcome <> 'success' + AND {_source_condition(source_bucket)} ORDER BY CASE source_bucket WHEN 'iiif' THEN 1 @@ -219,8 +292,10 @@ def _missing_sql(scope: str) -> str: WHEN 'cog' THEN 5 WHEN 'pmtiles' THEN 6 WHEN 'schema_image' THEN 7 - WHEN 'bridge_asset' THEN 8 - ELSE 9 + WHEN 'download_image' THEN 8 + WHEN 'download_pdf' THEN 9 + WHEN 'bridge_asset' THEN 10 + ELSE 11 END, id LIMIT :limit; @@ -233,21 +308,30 @@ def _rows_to_dicts(rows: list[Any]) -> list[dict[str, Any]]: def _total_row(rows: list[dict[str, Any]]) -> dict[str, Any]: - total = {"source_bucket": "TOTAL", "total": 0} + total = {"source_bucket": "TOTAL", "total": 0, "eligible": 0, "gallery_ready": 0} for column in OUTCOME_COLUMNS: total[column] = 0 for row in rows: total["total"] += int(row["total"] or 0) + total["eligible"] += int(row["eligible"] or 0) + total["gallery_ready"] += int(row["gallery_ready"] or 0) for column in OUTCOME_COLUMNS: total[column] += int(row[column] or 0) return total def _success_pct(row: dict[str, Any]) -> float: - total = int(row["total"] or 0) - if total <= 0: + eligible = int(row["eligible"] or 0) + if eligible <= 0: return 100.0 - return round((int(row["success"] or 0) / total) * 100, 2) + return round((int(row["success"] or 0) / eligible) * 100, 2) + + +def _gallery_ready_pct(row: dict[str, Any]) -> float: + eligible = int(row["eligible"] or 0) + if eligible <= 0: + return 100.0 + return round((int(row["gallery_ready"] or 0) / eligible) * 100, 2) def _format_table(rows: list[dict[str, Any]], sample_rows: list[dict[str, Any]]) -> str: @@ -255,6 +339,9 @@ def _format_table(rows: list[dict[str, Any]], sample_rows: list[dict[str, Any]]) headings = [ "source", "total", + "eligible", + "ready", + "ready%", "ok", "ok%", "held", @@ -272,6 +359,9 @@ def _format_table(rows: list[dict[str, Any]], sample_rows: list[dict[str, Any]]) rendered = { "source": str(row["source_bucket"]), "total": str(row["total"]), + "eligible": str(row["eligible"]), + "ready": str(row["gallery_ready"]), + "ready%": f"{_gallery_ready_pct(row):.2f}", "ok": str(row["success"]), "ok%": f"{_success_pct(row):.2f}", "held": str(row["placeheld"]), @@ -301,12 +391,21 @@ def _format_table(rows: list[dict[str, Any]], sample_rows: list[dict[str, Any]]) def _write_csv(rows: list[dict[str, Any]]) -> None: - fieldnames = ["source_bucket", "total", *OUTCOME_COLUMNS, "success_pct"] + fieldnames = [ + "source_bucket", + "total", + "eligible", + "gallery_ready", + "gallery_ready_pct", + *OUTCOME_COLUMNS, + "success_pct", + ] writer = csv.DictWriter(sys.stdout, fieldnames=fieldnames) writer.writeheader() for row in [*rows, _total_row(rows)]: out = dict(row) out["success_pct"] = _success_pct(row) + out["gallery_ready_pct"] = _gallery_ready_pct(row) writer.writerow(out) @@ -314,9 +413,20 @@ def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Report thumbnail completeness.") parser.add_argument( "--scope", - choices=("all", "urban", "iiif"), + choices=("all", "maps", "urban", "iiif"), default=os.getenv("THUMBNAIL_REPORT_SCOPE", "all"), ) + parser.add_argument( + "--provider", + default=None, + help="Limit the report to an exact schema_provider_s value.", + ) + parser.add_argument( + "--source-bucket", + choices=tuple(SOURCE_ORDER), + default=os.getenv("THUMBNAIL_REPORT_SOURCE_BUCKET") or None, + help="Limit the report to one classified thumbnail source bucket.", + ) parser.add_argument( "--format", choices=("table", "json", "csv"), @@ -342,14 +452,33 @@ def _parse_args() -> argparse.Namespace: def main() -> int: args = _parse_args() + query_params = {"provider": args.provider} if args.provider else {} engine = create_app_sync_engine(_sync_database_url()) with engine.begin() as conn: - rows = _rows_to_dicts(conn.execute(text(_summary_sql(args.scope))).fetchall()) + rows = _rows_to_dicts( + conn.execute( + text( + _summary_sql( + args.scope, + provider=args.provider, + source_bucket=args.source_bucket, + ) + ), + query_params, + ).fetchall() + ) sample_rows: list[dict[str, Any]] = [] if args.show_missing > 0: sample_rows = _rows_to_dicts( conn.execute( - text(_missing_sql(args.scope)), {"limit": args.show_missing} + text( + _missing_sql( + args.scope, + provider=args.provider, + source_bucket=args.source_bucket, + ) + ), + {**query_params, "limit": args.show_missing}, ).fetchall() ) @@ -357,8 +486,21 @@ def main() -> int: payload = { "generated_at": datetime.now(timezone.utc).isoformat(), "scope": args.scope, - "summary": {**total, "success_pct": _success_pct(total)}, - "by_source": [{**row, "success_pct": _success_pct(row)} for row in rows], + "provider": args.provider, + "source_bucket": args.source_bucket, + "summary": { + **total, + "success_pct": _success_pct(total), + "gallery_ready_pct": _gallery_ready_pct(total), + }, + "by_source": [ + { + **row, + "success_pct": _success_pct(row), + "gallery_ready_pct": _gallery_ready_pct(row), + } + for row in rows + ], "missing_sample": sample_rows, } @@ -367,7 +509,12 @@ def main() -> int: elif args.format == "csv": _write_csv(rows) else: - print(f"thumbnail completeness | scope={args.scope}") + filters = [f"scope={args.scope}"] + if args.provider: + filters.append(f"provider={args.provider}") + if args.source_bucket: + filters.append(f"source={args.source_bucket}") + print(f"thumbnail completeness | {' | '.join(filters)}") print(_format_table(rows, sample_rows)) if args.fail_under is not None and payload["summary"]["success_pct"] < args.fail_under: diff --git a/backend/tests/scripts/test_prime_generated_caches.py b/backend/tests/scripts/test_prime_generated_caches.py new file mode 100644 index 0000000..b4284e2 --- /dev/null +++ b/backend/tests/scripts/test_prime_generated_caches.py @@ -0,0 +1,72 @@ +from argparse import Namespace +from collections import Counter +from unittest.mock import patch + +import pytest + +from scripts import prime_generated_caches + + +def _args() -> Namespace: + return Namespace( + resource_ids=[], + limit=None, + resource_batch_size=500, + resource_concurrency=16, + visual_batch_size=100, + thumbnail_concurrency=4, + static_map_concurrency=2, + force=False, + retry_thumbnail_failures=False, + retry_thumbnail_placeheld=False, + strict_failures=False, + hydrate_assets=False, + allow_full_hydration=False, + resource_class="Maps", + provider=None, + stage=["all"], + ) + + +@pytest.mark.asyncio +async def test_all_stages_prime_visuals_before_resource_representations(): + calls: list[str] = [] + + async def thumbnails(args): + calls.append("thumbnails") + assert args.resource_class == "Maps" + assert args.provider is None + return 0 + + async def static_maps(args): + calls.append("static-maps") + assert args.resource_class == "Maps" + assert args.provider is None + return 0 + + async def resources(**kwargs): + calls.append("resources") + assert kwargs["resource_class"] == "Maps" + assert kwargs["provider"] is None + return Counter() + + with ( + patch("scripts.prime_thumbnail_cache._run", side_effect=thumbnails), + patch("scripts.prime_static_map_cache._run", side_effect=static_maps), + patch( + "scripts.prime_resource_representation_cache.prime_resource_representation_cache", + side_effect=resources, + ), + ): + result = await prime_generated_caches._run(_args()) + + assert result == 0 + assert calls == ["thumbnails", "static-maps", "resources"] + + +def test_default_stage_order_places_resources_last(): + assert prime_generated_caches._normalize_stages(None) == [ + "thumbnails", + "static-maps", + "resources", + ] diff --git a/backend/tests/scripts/test_prime_resource_representation_cache.py b/backend/tests/scripts/test_prime_resource_representation_cache.py index 3214043..37976bc 100644 --- a/backend/tests/scripts/test_prime_resource_representation_cache.py +++ b/backend/tests/scripts/test_prime_resource_representation_cache.py @@ -7,6 +7,18 @@ import scripts.prime_resource_representation_cache as prime_resource_cache +def test_resource_filters_compile_exact_resource_class_and_provider(): + statement = prime_resource_cache._apply_resource_filters( + prime_resource_cache.select(prime_resource_cache.resources), + resource_class="Maps", + provider="OpenGeoMetadata", + ) + compiled = str(statement.compile(compile_kwargs={"literal_binds": True})) + + assert "'Maps' = ANY" in compiled + assert "schema_provider_s = 'OpenGeoMetadata'" in compiled + + class DummyProgress: def __init__(self, *args, **kwargs): self.total = kwargs.get("total") @@ -267,7 +279,12 @@ async def test_prime_resource_cache_counts_missing_explicit_resource_ids(): assert counters == Counter({"primed": 1, "missing": 1}) mock_connect.assert_awaited_once() mock_disconnect.assert_awaited_once_with(True) - mock_fetch.assert_awaited_once_with(["resource-1", "resource-missing"], None) + mock_fetch.assert_awaited_once_with( + ["resource-1", "resource-missing"], + None, + resource_class=None, + provider=None, + ) mock_prime_batch.assert_awaited_once() diff --git a/backend/tests/scripts/test_prime_static_map_cache.py b/backend/tests/scripts/test_prime_static_map_cache.py index 0b4ab6d..64c7381 100644 --- a/backend/tests/scripts/test_prime_static_map_cache.py +++ b/backend/tests/scripts/test_prime_static_map_cache.py @@ -5,6 +5,18 @@ import scripts.prime_static_map_cache as prime_static_map_cache +def test_resource_filters_compile_exact_resource_class_and_provider(): + statement = prime_static_map_cache._apply_resource_filters( + prime_static_map_cache.select(prime_static_map_cache.resources), + resource_class="Maps", + provider="OpenGeoMetadata", + ) + compiled = str(statement.compile(compile_kwargs={"literal_binds": True})) + + assert "'Maps' = ANY" in compiled + assert "schema_provider_s = 'OpenGeoMetadata'" in compiled + + def test_prime_static_maps_reuses_signature_aware_durable_cache(): service = MagicMock() service.geometry_signature.return_value = "sig-123" diff --git a/backend/tests/scripts/test_prime_thumbnail_cache.py b/backend/tests/scripts/test_prime_thumbnail_cache.py index ce84a5e..9d17c8d 100644 --- a/backend/tests/scripts/test_prime_thumbnail_cache.py +++ b/backend/tests/scripts/test_prime_thumbnail_cache.py @@ -5,6 +5,18 @@ import scripts.prime_thumbnail_cache as prime_thumbnail_cache +def test_resource_filters_compile_exact_resource_class_and_provider(): + statement = prime_thumbnail_cache._apply_resource_filters( + prime_thumbnail_cache.select(prime_thumbnail_cache.resources), + resource_class="Maps", + provider="OpenGeoMetadata", + ) + compiled = str(statement.compile(compile_kwargs={"literal_binds": True})) + + assert "'Maps' = ANY" in compiled + assert "schema_provider_s = 'OpenGeoMetadata'" in compiled + + @pytest.mark.asyncio async def test_prime_thumbnail_no_source_records_placeheld(): resource = {"id": "resource-no-source", "dct_accessrights_s": "Public"} @@ -17,6 +29,11 @@ async def test_prime_thumbnail_no_source_records_placeheld(): patch.object( prime_thumbnail_cache, "_get_thumbnail_asset_url", AsyncMock(return_value=None) ), + patch.object( + prime_thumbnail_cache, + "_prime_resource_class_icon", + AsyncMock(return_value="icon-hash"), + ), patch.object(prime_thumbnail_cache, "ImageService") as mock_service_cls, ): service = MagicMock() @@ -26,10 +43,55 @@ async def test_prime_thumbnail_no_source_records_placeheld(): result = await prime_thumbnail_cache._prime_thumbnail_for_resource(resource, force=False) - assert result == ("skipped-no-source", "resource-no-source", "no thumbnail source") + assert result == ("generated-icon", "resource-no-source", "resource-class icon") payload = mock_state.await_args.args[0] assert payload.state == "placeheld" assert payload.resource_id == "resource-no-source" + assert "OGM resource-class icon" in payload.state_detail + + +@pytest.mark.asyncio +async def test_prime_thumbnail_skips_canonical_restricted_resource_before_source_lookup(): + resource = {"id": "resource-restricted", "dct_accessRights_s": "Restricted"} + + with ( + patch.object(prime_thumbnail_cache, "fetch_distribution_context", AsyncMock()) as fetch, + patch.object(prime_thumbnail_cache, "ImageService") as mock_service_cls, + ): + result = await prime_thumbnail_cache._prime_thumbnail_for_resource(resource, force=False) + + assert result == ("skipped-restricted", "resource-restricted", "restricted") + fetch.assert_not_awaited() + mock_service_cls.assert_not_called() + + +@pytest.mark.asyncio +async def test_prime_thumbnail_failure_materializes_local_fallback_icon(): + resource = {"id": "resource-failed", "gbl_resourceClass_sm": ["Maps"]} + + with ( + patch.object( + prime_thumbnail_cache, + "_prime_thumbnail_for_resource", + AsyncMock(return_value=("failed", "resource-failed", "upstream failed")), + ), + patch.object( + prime_thumbnail_cache, + "_prime_resource_class_icon", + AsyncMock(return_value="icon-hash"), + ) as prime_icon, + ): + result = await prime_thumbnail_cache._prime_thumbnail_with_fallback_for_resource( + resource, + force=False, + ) + + assert result == ( + "failed", + "resource-failed", + f"upstream failed; {prime_thumbnail_cache.FALLBACK_ICON_DETAIL}", + ) + prime_icon.assert_awaited_once_with(resource, force=False) @pytest.mark.asyncio diff --git a/backend/tests/scripts/test_report_thumbnail_completeness.py b/backend/tests/scripts/test_report_thumbnail_completeness.py index 64e7dad..58f2cdf 100644 --- a/backend/tests/scripts/test_report_thumbnail_completeness.py +++ b/backend/tests/scripts/test_report_thumbnail_completeness.py @@ -6,6 +6,8 @@ def test_total_row_and_success_percentage_classify_outcomes(): { "source_bucket": "iiif", "total": 4, + "eligible": 4, + "gallery_ready": 3, "success": 2, "placeheld": 1, "failed": 0, @@ -18,6 +20,8 @@ def test_total_row_and_success_percentage_classify_outcomes(): { "source_bucket": "bridge_asset", "total": 2, + "eligible": 2, + "gallery_ready": 1, "success": 1, "placeheld": 0, "failed": 1, @@ -32,19 +36,31 @@ def test_total_row_and_success_percentage_classify_outcomes(): total = report._total_row(rows) assert total["total"] == 6 + assert total["eligible"] == 6 + assert total["gallery_ready"] == 4 assert total["success"] == 3 assert total["placeheld"] == 1 assert total["failed"] == 1 assert total["stale_success"] == 1 assert report._success_pct(total) == 50.0 + assert report._gallery_ready_pct(total) == 66.67 def test_scope_conditions_are_specific(): + assert "gbl_resourceClass_sm" in report._scope_condition("maps") + assert "'Maps'" in report._scope_condition("maps") assert "b1g_urbanBaseLayers" in report._scope_condition("urban") assert "iiif" in report._scope_condition("iiif").lower() assert report._scope_condition("all") == "TRUE" +def test_percentages_exclude_restricted_records_from_denominator(): + row = {"total": 10, "eligible": 8, "success": 6, "gallery_ready": 8} + + assert report._success_pct(row) == 75.0 + assert report._gallery_ready_pct(row) == 100.0 + + def test_summary_sql_tracks_source_buckets_and_real_durable_bytes(): sql = report._summary_sql("all") @@ -56,6 +72,8 @@ def test_summary_sql_tracks_source_buckets_and_real_durable_bytes(): "cog", "pmtiles", "schema_image", + "download_image", + "download_pdf", "bridge_asset", "no_source", ): @@ -63,15 +81,45 @@ def test_summary_sql_tracks_source_buckets_and_real_durable_bytes(): assert "gva.byte_size > 0" in sql assert "gva.content_type LIKE 'image/%'" in sql + assert "gva.asset_kind LIKE 'thumbnail%'" in sql + assert "links.asset_kind = 'resource-class-icon'" in sql + assert "gallery_ready" in sql assert "stale_success" in sql assert "not_attempted" in sql +def test_report_sql_supports_provider_and_source_filters(): + sql = report._summary_sql( + "maps", + provider="OpenGeoMetadata", + source_bucket="iiif", + ) + + assert "r.schema_provider_s = :provider" in sql + assert "source_bucket = 'iiif'" in sql + + +def test_provider_filter_is_never_applied_implicitly(monkeypatch): + monkeypatch.setenv("THUMBNAIL_REPORT_PROVIDER", "UNR") + monkeypatch.setattr(report.sys, "argv", ["report_thumbnail_completeness.py"]) + + assert report._parse_args().provider is None + + +def test_source_filter_rejects_unknown_bucket(): + import pytest + + with pytest.raises(ValueError, match="Unknown thumbnail source bucket"): + report._summary_sql("maps", source_bucket="not-real") + + def test_table_output_includes_missing_sample(): rows = [ { "source_bucket": "iiif", "total": 1, + "eligible": 1, + "gallery_ready": 0, "success": 0, "placeheld": 0, "failed": 0, diff --git a/backend/tests/services/test_access_policy.py b/backend/tests/services/test_access_policy.py new file mode 100644 index 0000000..13a5b06 --- /dev/null +++ b/backend/tests/services/test_access_policy.py @@ -0,0 +1,30 @@ +from app.services.access_policy import is_restricted_resource, resource_access_rights +from app.services.thumbnail_state_service import infer_source_type + + +def test_resource_access_rights_prefers_canonical_field(): + metadata = { + "dct_accessRights_s": "Restricted", + "dct_accessrights_s": "Public", + } + + assert resource_access_rights(metadata) == "Restricted" + assert is_restricted_resource(metadata) is True + + +def test_restricted_resource_accepts_legacy_field_and_normalizes_case(): + assert is_restricted_resource({"dct_accessrights_s": " restricted "}) is True + + +def test_resource_access_rights_accepts_single_value_arrays(): + assert resource_access_rights({"dct_accessRights_s": ["Public"]}) == "Public" + + +def test_non_restricted_and_missing_values_are_not_restricted(): + assert is_restricted_resource({"dct_accessRights_s": "Public"}) is False + assert is_restricted_resource({}) is False + assert is_restricted_resource(None) is False + + +def test_thumbnail_state_infers_pdf_sources(): + assert infer_source_type("https://example.org/download/map.PDF?version=2") == "pdf" diff --git a/backend/tests/services/test_image_service.py b/backend/tests/services/test_image_service.py index 546dce9..f4c1938 100644 --- a/backend/tests/services/test_image_service.py +++ b/backend/tests/services/test_image_service.py @@ -616,6 +616,58 @@ def test_get_thumbnail_source_url_pmtiles_overrides_schema_image(self): except Exception as e: assert _is_redis_connection_error(e) + def test_get_thumbnail_source_url_uses_labeled_download_image_as_fallback(self): + metadata = {"id": "test-doc"} + references = { + "http://schema.org/downloadUrl": [ + {"url": "https://example.com/data.zip", "label": "Shapefile"}, + {"url": "https://example.com/preview", "label": "JPEG"}, + ] + } + + service = ImageService(metadata) + + assert service._get_thumbnail_source_url(references) == "https://example.com/preview" + + def test_get_thumbnail_source_url_uses_pdf_download_after_image_options(self): + metadata = {"id": "test-doc"} + references = { + "http://schema.org/downloadUrl": [ + {"url": "https://example.com/data.zip", "label": "Shapefile"}, + {"url": "https://example.com/map.pdf", "label": "PDF"}, + ] + } + + service = ImageService(metadata) + source_url = service._get_thumbnail_source_url(references) + + assert source_url == "https://example.com/map.pdf" + assert service._is_pdf_url(source_url) + assert ( + service.thumbnail_image_hash_for_source_sync(source_url) + == hashlib.sha256(f"pdf-thumb:{source_url}".encode()).hexdigest() + ) + + def test_public_thumbnail_url_never_exposes_the_external_source(self): + source_url = "https://images.example.edu/map.jpg" + service = ImageService( + { + "id": "test-doc", + "dct_references_s": json.dumps( + {"http://schema.org/thumbnailUrl": source_url} + ), + } + ) + + with ( + patch.object(service, "_api_v1_base_url", return_value="https://ogm.example/api/v1"), + patch.object(service, "current_thumbnail_hash_for_source_sync", return_value=None), + ): + thumbnail_url = service.get_thumbnail_url() + + assert thumbnail_url == "https://ogm.example/api/v1/resources/test-doc/thumbnail" + assert source_url not in thumbnail_url + class TestImageServiceIsCogUrl: """Test _is_cog_url helper.""" diff --git a/backend/tests/services/test_ogm_harvest_service.py b/backend/tests/services/test_ogm_harvest_service.py index 04f816a..1020aba 100644 --- a/backend/tests/services/test_ogm_harvest_service.py +++ b/backend/tests/services/test_ogm_harvest_service.py @@ -78,6 +78,7 @@ async def test_import_and_missing_tracking_and_tags(self, tmp_path): run_started_at=run1_started, ) assert stats1["imported"] == 2 + assert importer.changed_thumbnail_resource_ids == {"test-ogm-a", "test-ogm-b"} # Verify tags injected row_a = await database.fetch_one( @@ -112,6 +113,7 @@ async def test_import_and_missing_tracking_and_tags(self, tmp_path): run_started_at=run2_started, ) assert stats2["imported"] == 1 + assert importer.changed_thumbnail_resource_ids == set() missing_rows2 = await database.fetch_all( select(ogm_resource_state) diff --git a/backend/tests/services/test_ogm_importer_thumbnail_changes.py b/backend/tests/services/test_ogm_importer_thumbnail_changes.py new file mode 100644 index 0000000..29fad38 --- /dev/null +++ b/backend/tests/services/test_ogm_importer_thumbnail_changes.py @@ -0,0 +1,48 @@ +from unittest.mock import AsyncMock, patch + +import pytest + +from app.services.ogm_harvest.importer import OGMResourceImporter + + +@pytest.mark.asyncio +async def test_changed_thumbnail_ids_detects_new_and_changed_sources_only(): + existing = { + "id": "unchanged", + "dct_references_s": '{"http://schema.org/thumbnailUrl":"https://example/a.jpg"}', + "b1g_image_ss": None, + "dct_accessRights_s": "Public", + "gbl_resourceClass_sm": ["Maps"], + "gbl_wxsIdentifier_s": None, + "dcat_bbox": None, + "locn_geometry": None, + } + changed = {**existing, "id": "changed"} + incoming = [ + dict(existing), + {**changed, "dct_references_s": '{"http://schema.org/thumbnailUrl":"https://example/b.jpg"}'}, + {**existing, "id": "new"}, + ] + + with patch( + "app.services.ogm_harvest.importer.database.fetch_all", + AsyncMock(return_value=[existing, changed]), + ): + result = await OGMResourceImporter()._changed_thumbnail_ids(incoming) + + assert result == {"changed", "new"} + + +def test_thumbnail_source_signature_is_stable_for_nested_metadata(): + left = { + "gbl_resourceClass_sm": ["Maps", "Datasets"], + "dct_references_s": {"image": {"url": "https://example/image.jpg"}}, + } + right = { + "dct_references_s": {"image": {"url": "https://example/image.jpg"}}, + "gbl_resourceClass_sm": ["Maps", "Datasets"], + } + + assert OGMResourceImporter._thumbnail_source_signature( + left + ) == OGMResourceImporter._thumbnail_source_signature(right) diff --git a/backend/tests/services/test_remote_fetch.py b/backend/tests/services/test_remote_fetch.py new file mode 100644 index 0000000..c570598 --- /dev/null +++ b/backend/tests/services/test_remote_fetch.py @@ -0,0 +1,66 @@ +from unittest.mock import MagicMock, patch + +import pytest + +from app.services.remote_fetch import ( + RemotePayloadTooLarge, + UnsafeRemoteUrl, + fetch_public_http_bytes, + validate_public_http_url, +) + + +@pytest.mark.parametrize( + "url", + ( + "http://127.0.0.1/secret", + "http://169.254.169.254/latest/meta-data", + "http://localhost/internal", + "file:///etc/passwd", + "https://user:password@example.com/image.jpg", + ), +) +def test_validate_public_http_url_rejects_unsafe_destinations(url): + with pytest.raises(UnsafeRemoteUrl): + validate_public_http_url(url, resolve_dns=False) + + +def test_fetch_public_http_bytes_validates_redirect_and_enforces_size(): + redirect = MagicMock( + is_redirect=True, + is_permanent_redirect=False, + headers={"Location": "https://cdn.example.org/map.pdf"}, + ) + response = MagicMock( + is_redirect=False, + is_permanent_redirect=False, + headers={"Content-Type": "application/pdf"}, + ) + response.iter_content.return_value = [b"%PDF", b"-payload"] + + with ( + patch( + "app.services.remote_fetch.socket.getaddrinfo", + return_value=[(2, 1, 6, "", ("93.184.216.34", 443))], + ), + patch("app.services.remote_fetch.requests.get", side_effect=[redirect, response]), + ): + result = fetch_public_http_bytes( + "https://example.org/map.pdf", + timeout=5, + max_bytes=32, + ) + + assert result.body == b"%PDF-payload" + assert result.final_url == "https://cdn.example.org/map.pdf" + + response.iter_content.return_value = [b"x" * 33] + with ( + patch( + "app.services.remote_fetch.socket.getaddrinfo", + return_value=[(2, 1, 6, "", ("93.184.216.34", 443))], + ), + patch("app.services.remote_fetch.requests.get", return_value=response), + pytest.raises(RemotePayloadTooLarge), + ): + fetch_public_http_bytes("https://example.org/map.pdf", timeout=5, max_bytes=32) diff --git a/backend/tests/services/test_thumbnail_refresh_service.py b/backend/tests/services/test_thumbnail_refresh_service.py new file mode 100644 index 0000000..05f3835 --- /dev/null +++ b/backend/tests/services/test_thumbnail_refresh_service.py @@ -0,0 +1,45 @@ +from unittest.mock import AsyncMock, patch + +import pytest + +import app.services.thumbnail_refresh_service as refresh + + +class _Cache: + def __init__(self): + self.invalidated = [] + + async def invalidate_tags(self, tags): + self.invalidated.append(tags) + return len(tags) + + +@pytest.mark.asyncio +async def test_refresh_is_thumbnail_only_and_deduplicates_resources(monkeypatch): + cache = _Cache() + resource_dicts = [{"id": "map-1"}, {"id": "map-2"}] + monkeypatch.setenv("OGM_THUMBNAIL_REFRESH_ENABLED", "true") + monkeypatch.setenv("OGM_THUMBNAIL_REFRESH_BATCH_SIZE", "2") + + with ( + patch.object(refresh, "CacheService", return_value=cache), + patch.object( + refresh, + "delete_resource_representations", + new=AsyncMock(return_value={"durable_deleted": True, "redis_deleted": 2}), + ), + patch.object(refresh, "_fetch_resources", new=AsyncMock(return_value=resource_dicts)), + patch.object( + refresh, + "_prime_resources", + new=AsyncMock(return_value={"attempted": 2, "generated": 2}), + ) as prime, + ): + stats = await refresh.refresh_thumbnail_cache_for_changed_resources( + ["map-1", "map-2", "map-1"] + ) + + assert stats["resources"] == 2 + assert stats["thumbnails"] == {"attempted": 2, "generated": 2} + assert cache.invalidated == [["resource:map-1", "resource:map-2"]] + prime.assert_awaited_once_with(resource_dicts, concurrency=2) diff --git a/backend/tests/tasks/test_worker_cog_thumbnail.py b/backend/tests/tasks/test_worker_cog_thumbnail.py index 2417f9c..f2d241b 100644 --- a/backend/tests/tasks/test_worker_cog_thumbnail.py +++ b/backend/tests/tasks/test_worker_cog_thumbnail.py @@ -168,6 +168,8 @@ def test_returns_true_and_caches_on_success(self, patch_worker_side_effects): with ( patch("app.tasks.worker.redis_client") as mock_redis, + patch("app.tasks.worker.store_durable_visual_asset", return_value=True), + patch("app.tasks.worker.store_durable_visual_asset_link", return_value=True), patch( "app.tasks.worker._generate_cog_thumbnail_bytes", return_value=png_bytes, diff --git a/backend/tests/tasks/test_worker_fetch_and_cache_image.py b/backend/tests/tasks/test_worker_fetch_and_cache_image.py index 48a7eb0..81f39a3 100644 --- a/backend/tests/tasks/test_worker_fetch_and_cache_image.py +++ b/backend/tests/tasks/test_worker_fetch_and_cache_image.py @@ -62,6 +62,8 @@ def test_fetch_and_cache_image_records_success_and_uses_provider_throttle(): source_url = "https://example.com/thumb.png" response = MagicMock() response.status_code = 200 + response.is_redirect = False + response.is_permanent_redirect = False response.content = _valid_png_bytes() response.headers = {"Content-Type": "image/png"} response.raise_for_status = MagicMock() @@ -69,6 +71,12 @@ def test_fetch_and_cache_image_records_success_and_uses_provider_throttle(): with ( patch("app.tasks.worker._resolve_image_url", return_value=source_url), patch("app.tasks.worker.redis_client") as mock_redis, + patch("app.tasks.worker.store_durable_visual_asset", return_value=True), + patch("app.tasks.worker.store_durable_visual_asset_link", return_value=True), + patch( + "app.services.remote_fetch.socket.getaddrinfo", + return_value=[(2, 1, 6, "", ("93.184.216.34", 443))], + ), patch("app.tasks.worker.requests.get", return_value=response), patch( "app.tasks.worker.provider_request_slot", @@ -93,6 +101,8 @@ def test_fetch_and_cache_image_records_failure_for_invalid_content(): source_url = "https://example.com/thumb.png" response = MagicMock() response.status_code = 200 + response.is_redirect = False + response.is_permanent_redirect = False response.content = b"not an image" response.headers = {"Content-Type": "text/html"} response.raise_for_status = MagicMock() @@ -100,6 +110,12 @@ def test_fetch_and_cache_image_records_failure_for_invalid_content(): with ( patch("app.tasks.worker._resolve_image_url", return_value=source_url), patch("app.tasks.worker.redis_client") as mock_redis, + patch("app.tasks.worker.store_durable_visual_asset", return_value=True), + patch("app.tasks.worker.store_durable_visual_asset_link", return_value=True), + patch( + "app.services.remote_fetch.socket.getaddrinfo", + return_value=[(2, 1, 6, "", ("93.184.216.34", 443))], + ), patch("app.tasks.worker.requests.get", return_value=response), patch( "app.tasks.worker.provider_request_slot", @@ -123,6 +139,8 @@ def test_fetch_and_cache_image_resizes_large_remote_image_before_caching(): source_url = "https://example.com/huge-thumb.jpg" response = MagicMock() response.status_code = 200 + response.is_redirect = False + response.is_permanent_redirect = False response.content = _large_jpeg_bytes() response.headers = {"Content-Type": "image/jpeg"} response.raise_for_status = MagicMock() @@ -130,6 +148,12 @@ def test_fetch_and_cache_image_resizes_large_remote_image_before_caching(): with ( patch("app.tasks.worker._resolve_image_url", return_value=source_url), patch("app.tasks.worker.redis_client") as mock_redis, + patch("app.tasks.worker.store_durable_visual_asset", return_value=True), + patch("app.tasks.worker.store_durable_visual_asset_link", return_value=True), + patch( + "app.services.remote_fetch.socket.getaddrinfo", + return_value=[(2, 1, 6, "", ("93.184.216.34", 443))], + ), patch("app.tasks.worker.requests.get", return_value=response), patch( "app.tasks.worker.provider_request_slot", @@ -157,6 +181,24 @@ def test_worker_does_not_treat_dataset_manifest_as_iiif_manifest(): ) +def test_worker_rejects_private_thumbnail_source_without_fetching(): + source_url = "http://127.0.0.1/private.png" + + with ( + patch("app.tasks.worker._resolve_image_url", return_value=source_url), + patch("app.tasks.worker.requests.get") as request_get, + patch("app.tasks.worker.safe_record_thumbnail_state_sync") as mock_state, + patch("app.tasks.worker.release_thumbnail_queue_slot"), + ): + result = fetch_and_cache_image(source_url, "resource-private") + + assert result is False + request_get.assert_not_called() + payload = mock_state.call_args.args[0] + assert payload.state == "failure" + assert "unsafe or oversized" in payload.state_detail + + def test_worker_resolves_iiif_info_before_fetching_image(): info_url = "https://example.com/iiif/item/info.json" image_url = "https://example.com/iiif/item/full/924,/0/default.jpg" @@ -174,6 +216,8 @@ def test_unr_level_zero_info_worker_fetches_and_caches_real_rendition(): """Exercise the complete raw info.json -> rendition -> image-cache worker path.""" response = MagicMock() response.status_code = 200 + response.is_redirect = False + response.is_permanent_redirect = False response.content = _valid_png_bytes() response.headers = {"Content-Type": "image/png"} response.raise_for_status = MagicMock() @@ -184,6 +228,12 @@ def test_unr_level_zero_info_worker_fetches_and_caches_real_rendition(): return_value=UNR_LEVEL_ZERO_INFO, ), patch("app.tasks.worker.redis_client") as mock_redis, + patch("app.tasks.worker.store_durable_visual_asset", return_value=True), + patch("app.tasks.worker.store_durable_visual_asset_link", return_value=True), + patch( + "app.services.remote_fetch.socket.getaddrinfo", + return_value=[(2, 1, 6, "", ("93.184.216.34", 443))], + ), patch("app.tasks.worker.requests.get", return_value=response) as mock_get, patch( "app.tasks.worker.provider_request_slot", @@ -200,6 +250,8 @@ def test_unr_level_zero_info_worker_fetches_and_caches_real_rendition(): UNR_IMAGE_URL, timeout=30, headers={"User-Agent": "BTAA-Geospatial-Data-API/1.0 (https://geo.btaa.org/)"}, + allow_redirects=False, + stream=True, ) image_key, cached_bytes = _cached_image_write(mock_redis) resolved_hash = _remote_thumbnail_image_hash(UNR_IMAGE_URL) diff --git a/backend/tests/tasks/test_worker_pdf_thumbnail.py b/backend/tests/tasks/test_worker_pdf_thumbnail.py new file mode 100644 index 0000000..926708f --- /dev/null +++ b/backend/tests/tasks/test_worker_pdf_thumbnail.py @@ -0,0 +1,72 @@ +import io +from pathlib import Path +from unittest.mock import MagicMock, patch + +from PIL import Image + +from app.services.remote_fetch import RemoteBytes +from app.tasks import worker + + +def _png_bytes() -> bytes: + output = io.BytesIO() + Image.new("RGB", (20, 20), "white").save(output, format="PNG") + return output.getvalue() + + +def test_render_pdf_first_page_rejects_non_pdf_bytes(): + assert worker._render_pdf_first_page(b"not a pdf") is None + + +def test_render_pdf_first_page_reads_poppler_output(): + expected = _png_bytes() + + def fake_run(args, **_kwargs): + Path(f"{args[-1]}.png").write_bytes(expected) + return MagicMock(returncode=0) + + with ( + patch("app.tasks.worker.shutil.which", return_value="/usr/bin/pdftoppm"), + patch("app.tasks.worker.subprocess.run", side_effect=fake_run), + ): + result = worker._render_pdf_first_page(b"%PDF-1.7\nmock") + + assert result == expected + + +def test_generate_pdf_thumbnail_bytes_fetches_bounded_pdf_then_renders(): + fetched = RemoteBytes( + body=b"%PDF-1.7\nmock", + content_type="application/pdf", + final_url="https://example.org/map.pdf", + ) + expected = _png_bytes() + with ( + patch("app.tasks.worker.fetch_public_http_bytes", return_value=fetched) as fetch, + patch("app.tasks.worker._render_pdf_first_page", return_value=expected) as render, + ): + result = worker._generate_pdf_thumbnail_bytes("https://example.org/map.pdf") + + assert result == expected + assert fetch.call_args.kwargs["max_bytes"] == worker.PDF_THUMBNAIL_MAX_BYTES + render.assert_called_once_with(fetched.body) + + +def test_persist_thumbnail_succeeds_durably_when_redis_is_unavailable(): + with ( + patch("app.tasks.worker.store_durable_visual_asset", return_value=True) as store, + patch("app.tasks.worker.store_durable_visual_asset_link", return_value=True) as link, + patch("app.tasks.worker.cache_visual_asset", side_effect=RuntimeError("redis down")), + patch("app.tasks.worker.durable_visual_asset_enabled", return_value=True), + ): + result = worker._persist_thumbnail_bytes( + "a" * 64, + _png_bytes(), + "image/png", + resource_id="resource-1", + asset_kind="thumbnail:pdf", + ) + + assert result is True + store.assert_called_once() + link.assert_called_once() diff --git a/backend/tests/tasks/test_worker_pmtiles_thumbnail.py b/backend/tests/tasks/test_worker_pmtiles_thumbnail.py index 76861e8..0b3a9b7 100644 --- a/backend/tests/tasks/test_worker_pmtiles_thumbnail.py +++ b/backend/tests/tasks/test_worker_pmtiles_thumbnail.py @@ -385,6 +385,8 @@ def test_returns_true_and_caches_on_success(self, patch_worker_side_effects): with ( patch("app.tasks.worker.redis_client") as mock_redis, + patch("app.tasks.worker.store_durable_visual_asset", return_value=True), + patch("app.tasks.worker.store_durable_visual_asset_link", return_value=True), patch( "app.tasks.worker._generate_pmtiles_thumbnail_bytes", return_value=png_bytes, diff --git a/config/deploy.yml b/config/deploy.yml index cc29e9d..e3fcd02 100644 --- a/config/deploy.yml +++ b/config/deploy.yml @@ -67,6 +67,11 @@ env: CRON_LOCAL_TIMEZONE: America/Chicago OGM_TRIGGER: nightly OGM_NIGHTLY_CRON_ENABLED: "false" + OGM_THUMBNAIL_REFRESH_ENABLED: "true" + OGM_THUMBNAIL_REFRESH_BATCH_SIZE: "500" + OGM_THUMBNAIL_REFRESH_CONCURRENCY: "2" + PDF_THUMBNAIL_MAX_BYTES: "33554432" + REMOTE_THUMBNAIL_MAX_BYTES: "20971520" RATE_LIMIT_ENABLED: "false" secret: diff --git a/docs/cache_priming.md b/docs/cache_priming.md index f6aa24e..2ec2115 100644 --- a/docs/cache_priming.md +++ b/docs/cache_priming.md @@ -10,11 +10,17 @@ cd backend python scripts/prime_generated_caches.py ``` -It runs three stages: +It runs three stages, in this order: -- `resources`: durable JSON:API resource representations for resource detail and search result payloads - `thumbnails`: durable thumbnail visual assets, thumbnail state, and resource-to-asset links - `static-maps`: durable static-map and basemap visual assets plus aliases +- `resources`: durable JSON:API resource representations for resource detail and search result payloads + +The ordering is intentional: representations are generated after visual assets, +so gallery payloads contain immutable OGM API asset URLs. External IIIF, image, +service, COG, PMTiles, download-image, and PDF URLs are source inputs only. The +API never asks a browser to load those source URLs as thumbnails. When a preview +cannot be materialized, the primer generates and stores an OGM resource-class icon. By default, full-corpus runs persist durable database-backed rows and avoid loading every image body into Redis. Runtime requests can rehydrate Redis from @@ -34,6 +40,15 @@ Prime only resource representations: make cache-prime ARGS="--stage resources" ``` +Prime the exact Maps resource-class cohort, with no provider filter: + +```bash +make cache-prime ARGS="--stage thumbnails --stage resources --resource-class Maps" +``` + +Do not add `--provider` for the cross-provider Maps coverage run. The option is +available only for diagnosing an individual provider. + Hydrate Redis image bodies for a bounded hotset: ```bash @@ -86,3 +101,58 @@ Use these options for retries: python scripts/prime_generated_caches.py --retry-thumbnail-failures python scripts/prime_generated_caches.py --retry-thumbnail-placeheld ``` + +## Maps Coverage Report + +Measure the same cohort as the gallery's +`include_filters[gbl_resourceClass_sm][]=Maps` query: + +```bash +cd backend +python scripts/report_thumbnail_completeness.py --scope maps --format table +python scripts/report_thumbnail_completeness.py --scope maps --format json +``` + +The report deliberately has no default provider filter. It separates two metrics: + +- `success_pct`: eligible records with a real, durable preview derived and stored by OGM. +- `gallery_ready_pct`: eligible records with either a durable preview or a durable + OGM-generated resource-class icon. Restricted records are excluded from this + denominator and are never fetched. + +Use `--show-missing 50` to sample real-preview gaps and `--source-bucket` to work +one source family at a time. A durable asset means image bytes exist in OGM's +visual-asset store; a source URL alone does not count. + +## Production Backfill Sequence + +After deploying the image containing Poppler (`pdftoppm`), capture a baseline, +prime locally owned visual assets, rebuild gallery representations, and verify: + +```bash +kamal app exec "cd /app/backend && python scripts/report_thumbnail_completeness.py --scope maps --format json" +kamal app exec "cd /app/backend && ./scripts/start_cache_prime_background.sh --stage thumbnails --stage resources --resource-class Maps --retry-thumbnail-failures --retry-thumbnail-placeheld" +kamal app exec "tail -f /app/backend/logs/prime_generated_caches.log" +kamal app exec "cd /app/backend && python scripts/report_thumbnail_completeness.py --scope maps --format json" +``` + +Start with a bounded canary by adding `--limit 500`, inspect failures, then run +the full cohort. Do not use `--hydrate-assets` for the unbounded run unless Redis +is sized for all image bodies; durable database assets remain gallery-ready and +can be rehydrated on demand. + +## Ongoing Refresh + +OGM harvests compare fields that can change thumbnail selection. Only changed or +new records are re-primed, and their generated API representations are invalidated. +Controls: + +- `OGM_THUMBNAIL_REFRESH_ENABLED` (default `true`) +- `OGM_THUMBNAIL_REFRESH_BATCH_SIZE` (default `500`) +- `OGM_THUMBNAIL_REFRESH_CONCURRENCY` (default `2`) +- `PDF_THUMBNAIL_MAX_BYTES` (default `33554432`) +- `REMOTE_THUMBNAIL_MAX_BYTES` (default `20971520`) + +Set `OGM_THUMBNAIL_REFRESH_ENABLED=false` to stop post-harvest refresh without +disabling harvest. Existing immutable OGM assets continue to serve during a +rollback. diff --git a/docs/harvester.md b/docs/harvester.md index e6f4a1d..92f0f86 100644 --- a/docs/harvester.md +++ b/docs/harvester.md @@ -71,3 +71,12 @@ For the full application workflow, these backend scripts usually matter more tha - `scripts/run_index.py` Those scripts populate the database tables used by the API and enqueue the nightly ingest pipeline, while `ogm_harvester.py` remains useful for direct inspection and lower-level debugging. + +After a successful repository upsert, the application detects new records and +changes to thumbnail-bearing fields (references, image fields, access rights, +resource class, service identifier, and geometry). It re-primes only those +records into OGM-owned visual storage and invalidates generated resource +representations that may contain an older asset URL. This refresh can be disabled +with `OGM_THUMBNAIL_REFRESH_ENABLED=false`; a refresh failure is recorded in the +harvest statistics and does not turn an otherwise successful metadata harvest +into a failure.