music-assistant-server

1"""Utilities for image manipulation and retrieval."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7import hashlib
8import itertools
9import logging
10import os
11import random
12import re
13import tempfile
14import time
15import urllib.parse
16from base64 import b64decode
17from collections import OrderedDict
18from collections.abc import Iterable
19from io import BytesIO
20from pathlib import Path
21from typing import TYPE_CHECKING, cast
22
23import aiofiles
24import aiofiles.os
25from aiohttp.client_exceptions import ClientError
26from music_assistant_models.enums import ProviderIconVariant
27from music_assistant_models.errors import MediaNotFoundError, MusicAssistantError
28from PIL import Image, UnidentifiedImageError
29
30from music_assistant.constants import APPLICATION_NAME
31from music_assistant.helpers.security import is_safe_path
32from music_assistant.helpers.tags import get_embedded_image
33from music_assistant.helpers.util import join_task
34from music_assistant.models.metadata_provider import MetadataProvider
35from music_assistant.models.music_provider import MusicProvider
36from music_assistant.models.player_provider import PlayerProvider
37from music_assistant.models.plugin import PluginProvider
38
39if TYPE_CHECKING:
40    from music_assistant_models.media_items import MediaItemImage
41    from PIL.Image import Image as ImageClass
42
43    from music_assistant.mass import MusicAssistant
44
45
46LOGGER = logging.getLogger(__name__)
47
48# Thumbnail cache: on-disk (persistent) + small in-memory FIFO (hot path)
49_THUMB_CACHE_DIR = "thumbnails"
50_THUMB_MEMORY_CACHE_MAX = 50
51_ALLOWED_THUMB_FORMATS: frozenset[str] = frozenset({"PNG", "JPEG"})
52
53# Bump on encoding-rule changes so stale entries (e.g. old black-bg JPEGs for
54# transparent logos) aren't served from a colliding filename after upgrade.
55_THUMB_CACHE_VERSION = 2
56
57# By construction the filename is `<sha256>_<int>_v<int>[_flat].(jpg|png)`; the
58# regex is an explicit sanitizer that also lets CodeQL prove the value is safe
59# to join into a filesystem path. The `_flat` marker separates the flattened
60# and transparency-preserving cache variants.
61_THUMB_FILENAME_RE = re.compile(r"^[0-9a-f]{64}_\d+_v\d+(?:_flat)?\.(?:jpg|png)$")
62
63_thumb_memory_cache: OrderedDict[str, bytes] = OrderedDict()
64
65# Source-image cache: the raw origin bytes (remote download, local file read or
66# ffmpeg-extracted embedded art) that every derived artifact (thumb sizes/formats,
67# color palette, collage tiles) is generated from. Without it, first display of a
68# single item fetches the same source several times within seconds. The memory
69# tier is byte-budgeted (originals can be multi-MB); sources also get an on-disk
70# `<hash>_src` entry in the thumbnail cache dir so multi-variant generation
71# after a restart doesn't re-fetch either.
72_SOURCE_CACHE_SUFFIX = "_src"
73# remote urls can serve new content behind a stable url, so keep the TTL modest;
74# local files don't rely on it as invalidate_cached_image busts them on change
75_SOURCE_CACHE_TTL = 3600
76_SOURCE_MEMORY_MAX_BYTES = 32 * 1024 * 1024
77# a single huge original would evict everything else, so cap the entry size
78_SOURCE_MEMORY_ENTRY_MAX_BYTES = 8 * 1024 * 1024
79
80_MAX_IMAGEPROXY_RECURSION_DEPTH = 5
81
82# Leading magic bytes used to sniff raster image formats from their content.
83_PNG_MAGIC = b"\x89PNG\r\n\x1a\n"
84_JPEG_MAGIC = b"\xff\xd8\xff"
85
86
87def is_svg_data(data: bytes) -> bool:
88    """Return True when the given bytes appear to be an SVG image."""
89    if not data:
90        return False
91    # the root <svg> may be preceded by an xml declaration, doctype or comment
92    sample = data[:1024].lstrip()
93    if not sample[:64].lower().startswith((b"<?xml", b"<svg", b"<!--", b"<!doctype")):
94        return False
95    return b"<svg" in sample.lower()
96
97
98def detect_image_content_format(data: bytes) -> str | None:
99    """
100    Return the sniffed image format (`png`, `jpg` or `svg`), or None if unknown.
101
102    :param data: Raw image bytes to inspect.
103    """
104    if not data:
105        return None
106    if data.startswith(_PNG_MAGIC):
107        return "png"
108    if data.startswith(_JPEG_MAGIC):
109        return "jpg"
110    if is_svg_data(data):
111        return "svg"
112    return None
113
114
115def create_thumb_hash(provider: str, path_or_url: str) -> str:
116    """Create a safe filesystem hash from provider and image path."""
117    raw = f"{provider}/{path_or_url}"
118    return hashlib.sha256(raw.encode(), usedforsecurity=False).hexdigest()
119
120
121def _thumb_cache_filename(
122    thumb_hash: str,
123    size: int | None,
124    image_format: str,
125    flatten_transparency: bool = False,
126) -> str:
127    """Build the cache filename for a thumbnail."""
128    ext = image_format.lower()
129    if ext == "jpeg":
130        ext = "jpg"
131    suffix = "_flat" if flatten_transparency else ""
132    return f"{thumb_hash}_{size or 0}_v{_THUMB_CACHE_VERSION}{suffix}.{ext}"
133
134
135def _get_from_memory_cache(key: str) -> bytes | None:
136    """Retrieve thumbnail from in-memory FIFO cache."""
137    if key in _thumb_memory_cache:
138        _thumb_memory_cache.move_to_end(key)
139        return _thumb_memory_cache[key]
140    return None
141
142
143def _put_in_memory_cache(key: str, data: bytes) -> None:
144    """Store thumbnail in in-memory FIFO cache."""
145    _thumb_memory_cache[key] = data
146    _thumb_memory_cache.move_to_end(key)
147    while len(_thumb_memory_cache) > _THUMB_MEMORY_CACHE_MAX:
148        _thumb_memory_cache.popitem(last=False)
149
150
151class _SourceMemoryCache:
152    """In-memory byte-budgeted LRU tier of the source-image cache."""
153
154    def __init__(self) -> None:
155        self.entries: OrderedDict[str, tuple[bytes, float]] = OrderedDict()
156        self.total_bytes = 0
157
158    def get(self, key: str) -> bytes | None:
159        """Return cached source bytes for key, or None on miss/expiry."""
160        entry = self.entries.get(key)
161        if entry is None:
162            return None
163        data, stored_at = entry
164        if time.monotonic() - stored_at > _SOURCE_CACHE_TTL:
165            self.pop(key)
166            return None
167        self.entries.move_to_end(key)
168        return data
169
170    def put(self, key: str, data: bytes) -> None:
171        """Store source bytes for key, evicting oldest entries over the byte budget."""
172        if len(data) > _SOURCE_MEMORY_ENTRY_MAX_BYTES:
173            return
174        self.pop(key)
175        self.entries[key] = (data, time.monotonic())
176        self.total_bytes += len(data)
177        while self.total_bytes > _SOURCE_MEMORY_MAX_BYTES and self.entries:
178            _, (evicted, _stored_at) = self.entries.popitem(last=False)
179            self.total_bytes -= len(evicted)
180
181    def pop(self, key: str) -> None:
182        """Remove the entry for key (if present)."""
183        if entry := self.entries.pop(key, None):
184            self.total_bytes -= len(entry[0])
185
186    def clear(self) -> None:
187        """Remove all entries."""
188        self.entries.clear()
189        self.total_bytes = 0
190
191
192_source_memory_cache = _SourceMemoryCache()
193
194# Negative cache for sources that recently failed to fetch. Without it, a
195# persistently failing origin (e.g. an artwork URL that keeps returning 404)
196# is re-fetched — with a fresh error logged each time — for every thumbnail,
197# palette or metadata request that references it, and each consumer waits for
198# the full network round-trip just to fail again.
199_FAILED_SOURCE_TTL = 300
200_FAILED_SOURCE_MAX_ENTRIES = 256
201_failed_sources: OrderedDict[str, tuple[float, str]] = OrderedDict()
202
203
204def _get_failed_source(cache_key: str) -> str | None:
205    """Return the failure message for a recently failed source, or None."""
206    entry = _failed_sources.get(cache_key)
207    if entry is None:
208        return None
209    expires_at, message = entry
210    if time.monotonic() >= expires_at:
211        _failed_sources.pop(cache_key, None)
212        return None
213    return message
214
215
216def _store_failed_source(cache_key: str, message: str) -> None:
217    """Remember a failed source fetch so it is not retried for a short while."""
218    _failed_sources[cache_key] = (time.monotonic() + _FAILED_SOURCE_TTL, message)
219    _failed_sources.move_to_end(cache_key)
220    while len(_failed_sources) > _FAILED_SOURCE_MAX_ENTRIES:
221        _failed_sources.popitem(last=False)
222
223
224def _has_alpha(img: ImageClass) -> bool:
225    """Return True if the image actually uses transparency."""
226    if img.mode == "P":
227        return "transparency" in img.info
228    if img.mode in ("RGBA", "LA", "PA"):
229        # an alpha channel may still be fully opaque; only treat it as
230        # transparent when some pixel is not fully opaque
231        # single-band getextrema() returns a (min, max) numeric tuple
232        return cast("tuple[float, float]", img.getchannel("A").getextrema())[0] < 255
233    return False
234
235
236_IMAGEPROXY_V2_PREFIX = "/imageproxy/"
237
238
239def _extract_imageproxy_id(url: str) -> str | None:
240    """
241    Return the 64-hex image_id from a /imageproxy/<id> URL, or None.
242
243    The path must match the canonical shape `/imageproxy/<id>` (optionally
244    with a single trailing slash) — extra segments are rejected so this
245    helper agrees with what `MetaDataController.handle_imageproxy` accepts.
246    """
247    # bail out early on anything that obviously can't be a v2 imageproxy URL
248    # (non-strings such as MagicMock from tests; urlparse would TypeError)
249    if not isinstance(url, str) or _IMAGEPROXY_V2_PREFIX not in url:
250        return None
251    parsed = urllib.parse.urlparse(url)
252    if not parsed.path.startswith(_IMAGEPROXY_V2_PREFIX):
253        return None
254    remainder = parsed.path[len(_IMAGEPROXY_V2_PREFIX) :].rstrip("/").lower()
255    if len(remainder) != 64 or any(c not in "0123456789abcdef" for c in remainder):
256        return None
257    return remainder
258
259
260def player_image_url(mass: MusicAssistant, url: str | None) -> str | None:
261    """
262    Rewrite an imageproxy URL for consumption by a (physical) player.
263
264    :param mass: The MusicAssistant instance.
265    :param url: Image URL as produced for frontend/API consumers.
266    """
267    if not url:
268        return url
269    webserver_base = mass.webserver.base_url
270    if webserver_base and url.startswith(f"{webserver_base}/imageproxy"):
271        # players may not be able to reach the webserver, so serve from the streams
272        # server, and force jpeg (= flatten transparency) as players such as legacy
273        # AirPlay receivers cannot be assumed to handle PNG alpha
274        url = mass.streams.base_url + url[len(webserver_base) :]
275        parsed = urllib.parse.urlparse(url)
276        query = urllib.parse.parse_qs(parsed.query)
277        query["fmt"] = ["jpeg"]
278        return parsed._replace(query=urllib.parse.urlencode(query, doseq=True)).geturl()
279    return url
280
281
282async def get_image_data(
283    mass: MusicAssistant, path_or_url: str, provider: str, *, _depth: int = 0
284) -> bytes:
285    """
286    Retrieve image data from a path or URL.
287
288    Source bytes are cached (in memory and on disk) so that deriving multiple
289    artifacts from one image (thumb sizes/formats, palette, collage tiles)
290    only fetches the origin once. Concurrent requests for the same source
291    share a single fetch.
292
293    :param mass: The MusicAssistant instance.
294    :param path_or_url: The image path, URL, or base64 data URI.
295    :param provider: The provider ID that can resolve the image.
296    :param _depth: Internal recursion depth counter (do not set manually).
297    """
298    if _depth >= _MAX_IMAGEPROXY_RECURSION_DEPTH:
299        msg = f"Maximum recursion depth exceeded when fetching image: {path_or_url}"
300        raise FileNotFoundError(msg)
301    # base64 data URIs carry their content inline; just decode them
302    if path_or_url.startswith("data:image"):
303        return b64decode(path_or_url.rsplit(",", maxsplit=1)[-1])
304    # imageproxy URLs pointing at our own server are resolved to their underlying
305    # (provider, path) before anything is cached: an alias-keyed cache entry
306    # would keep serving after the underlying image was invalidated
307    if path_or_url.startswith("http") and (
308        resolved := await _resolve_own_imageproxy_url(mass, path_or_url)
309    ):
310        extracted_provider, extracted_path = resolved
311        return await get_image_data(mass, extracted_path, extracted_provider, _depth=_depth + 1)
312    cache_key = create_thumb_hash(provider, path_or_url)
313    if (cached := _source_memory_cache.get(cache_key)) is not None:
314        return cached
315    # fail fast on sources that just failed instead of hammering the origin
316    if (failure := _get_failed_source(cache_key)) is not None:
317        raise FileNotFoundError(failure)
318    # fetch de-duplicated across concurrent requests for the same source
319    task: asyncio.Task[bytes] = mass.create_task(
320        _fetch_and_cache_source_image,
321        mass,
322        path_or_url,
323        provider,
324        cache_key,
325        _depth,
326        task_id=f"imgsrc.{cache_key}",
327        abort_existing=False,
328    )
329    return await join_task(task)
330
331
332async def _resolve_own_imageproxy_url(mass: MusicAssistant, url: str) -> tuple[str, str] | None:
333    """
334    Resolve an imageproxy URL pointing at our own server to its (provider, path).
335
336    Returns None when the URL does not point at this server at all; raises
337    FileNotFoundError for own-server URLs that carry an invalid or unknown id.
338
339    :param mass: The MusicAssistant instance.
340    :param url: The (http/https) URL to inspect.
341    """
342    parsed_url = urllib.parse.urlparse(url)
343    url_origin = f"{parsed_url.scheme}://{parsed_url.netloc}"
344    server_origins = {
345        f"{p.scheme}://{p.netloc}"
346        for b in (mass.webserver.base_url, mass.streams.base_url)
347        if (p := urllib.parse.urlparse(b)).netloc
348    }
349    if url_origin not in server_origins:
350        return None
351    # opaque-id form: /imageproxy/<image_id>?size=...&fmt=...
352    if image_id := _extract_imageproxy_id(url):
353        resolved = await mass.metadata.resolve_image_id(image_id)
354        if resolved is None:
355            msg = f"Unknown image id in URL: {url}"
356            raise FileNotFoundError(msg)
357        return resolved
358    msg = f"Invalid imageproxy URL: {url}"
359    raise FileNotFoundError(msg)
360
361
362def _source_cache_filepath(mass: MusicAssistant, cache_key: str) -> str:
363    """Return the on-disk path for a cached source image, validating containment."""
364    thumb_dir = os.path.realpath(os.path.join(mass.cache_path, _THUMB_CACHE_DIR))
365    filepath = os.path.realpath(os.path.join(thumb_dir, f"{cache_key}{_SOURCE_CACHE_SUFFIX}"))
366    if not filepath.startswith(thumb_dir + os.sep):
367        msg = f"Cache path escapes thumbnail directory: {filepath}"
368        raise OSError(msg)
369    return filepath
370
371
372async def _fetch_and_cache_source_image(
373    mass: MusicAssistant, path_or_url: str, provider: str, cache_key: str, depth: int
374) -> bytes:
375    """
376    Fetch source image bytes, store them in the source cache tiers and return them.
377
378    :param mass: The MusicAssistant instance.
379    :param path_or_url: The image path or URL.
380    :param provider: The provider ID that can resolve the image.
381    :param cache_key: Source cache key (create_thumb_hash of provider + path).
382    :param depth: Recursion depth of the originating get_image_data call.
383    """
384    filepath = _source_cache_filepath(mass, cache_key)
385
386    def _read_disk_entry() -> bytes | None:
387        # remote urls only count as fresh within the TTL (a CDN can serve new
388        # content behind a stable url); local files rely on invalidation instead
389        try:
390            if not os.path.isfile(filepath):
391                return None
392            if (
393                path_or_url.startswith("http")
394                and time.time() - Path(filepath).stat().st_mtime > _SOURCE_CACHE_TTL
395            ):
396                return None
397            with open(filepath, "rb") as _file:
398                return _file.read()
399        except OSError:
400            return None
401
402    if disk_data := await asyncio.to_thread(_read_disk_entry):
403        _source_memory_cache.put(cache_key, disk_data)
404        return disk_data
405
406    try:
407        img_data, disk_cacheable = await _fetch_source_image(mass, path_or_url, provider, depth)
408    except (FileNotFoundError, MediaNotFoundError) as err:
409        # remember the failure briefly and log it once, concisely: every
410        # thumbnail/palette/metadata request for this source would otherwise
411        # retry the origin and log the same error over and over
412        # a provider signals a missing source with MediaNotFoundError, which is not an
413        # OSError and would otherwise bypass this negative cache entirely
414        _store_failed_source(cache_key, str(err))
415        LOGGER.warning("%s (not retrying for %s seconds)", err, _FAILED_SOURCE_TTL)
416        raise
417    _failed_sources.pop(cache_key, None)
418    _source_memory_cache.put(cache_key, img_data)
419    if disk_cacheable:
420        # persist to disk cache (best-effort, don't fail on I/O errors)
421        try:
422            await asyncio.to_thread(os.makedirs, os.path.dirname(filepath), exist_ok=True)
423            async with aiofiles.open(filepath, "wb") as _file:
424                await _file.write(img_data)
425        except OSError:
426            pass
427    return img_data
428
429
430async def _fetch_source_image(
431    mass: MusicAssistant, path_or_url: str, provider: str, depth: int
432) -> tuple[bytes, bool]:
433    """
434    Fetch image bytes from their origin.
435
436    Returns the raw bytes plus whether they may be persisted on disk under this
437    (provider, path) cache key. That is True for every origin except results
438    resolved under a different key (imageproxy URLs pointing at our own server):
439    an alias-keyed copy would keep being served after the underlying image is
440    invalidated.
441
442    :param mass: The MusicAssistant instance.
443    :param path_or_url: The image path or URL.
444    :param provider: The provider ID that can resolve the image.
445    :param depth: Recursion depth of the originating get_image_data call.
446    """
447    if prov := mass.get_provider(provider):
448        assert isinstance(prov, MusicProvider | MetadataProvider | PlayerProvider | PluginProvider)
449        if resolved_image := await prov.resolve_image(path_or_url):
450            if isinstance(resolved_image, bytes):
451                return resolved_image, True
452            if isinstance(resolved_image, str):
453                path_or_url = resolved_image
454    # handle HTTP location
455    if path_or_url.startswith("http"):
456        # handle imageproxy URLs pointing to our own server
457        if resolved := await _resolve_own_imageproxy_url(mass, path_or_url):
458            extracted_provider, extracted_path = resolved
459            # route through the public entrypoint so the result is cached under
460            # the resolved key; don't persist it under this alias key as well
461            return (
462                await get_image_data(mass, extracted_path, extracted_provider, _depth=depth + 1),
463                False,
464            )
465        try:
466            return await _fetch_remote_image(mass, path_or_url), True
467        except ClientError as err:
468            msg = f"Failed to fetch image from {path_or_url}: {err}"
469            raise FileNotFoundError(msg) from err
470    # handle base64 embedded images
471    if path_or_url.startswith("data:image"):
472        return b64decode(path_or_url.split(",")[-1]), True
473    # handle FILE location (of type image)
474    if path_or_url.endswith(("jpg", "JPG", "png", "PNG", "jpeg", "svg", "SVG")) and is_safe_path(
475        path_or_url
476    ):
477        if await asyncio.to_thread(os.path.isfile, path_or_url):
478            async with aiofiles.open(path_or_url, "rb") as _file:
479                return cast("bytes", await _file.read()), True
480    # use ffmpeg for embedded images
481    if is_safe_path(path_or_url) and (img_data := await get_embedded_image(path_or_url)):
482        return img_data, True
483    msg = f"Image not found: {path_or_url}"
484    raise FileNotFoundError(msg)
485
486
487async def _fetch_remote_image(mass: MusicAssistant, url: str) -> bytes:
488    """
489    Fetch raw image bytes over HTTP.
490
491    :param mass: The MusicAssistant instance.
492    :param url: The (http/https) image URL to fetch.
493    """
494    # Bot-protected CDNs (e.g. Akamai) reject our normal self-identifying User-Agent,
495    # and even regular browser User-Agents, while still serving well-known fetch tools.
496    # We keep identifying as Music Assistant but carry a Wget compatibility token, which
497    # such CDNs allowlist, so artwork is served on the first (and only) request.
498    user_agent = f"{APPLICATION_NAME}/{mass.version} (Wget/1.24.5; +https://music-assistant.io)"
499    async with mass.http_session_no_ssl.get(
500        url, raise_for_status=True, headers={"User-Agent": user_agent}
501    ) as resp:
502        return await resp.read()
503
504
505async def get_image_thumb(
506    mass: MusicAssistant,
507    path_or_url: str,
508    size: int | None,
509    provider: str,
510    image_format: str = "PNG",
511    flatten_transparency: bool = False,
512) -> bytes:
513    """
514    Get (optimized) thumbnail from image url.
515
516    Uses a two-tier cache (in-memory FIFO + on-disk) keyed by a hash of
517    provider + path so that repeated requests never trigger ffmpeg or
518    PIL processing again.  Concurrent requests for the same thumbnail
519    are de-duplicated via create_task.
520
521    :param mass: The MusicAssistant instance.
522    :param path_or_url: Path or URL to the source image.
523    :param size: Target thumbnail size (square), or None for original.
524    :param provider: Provider identifier for the image source.
525    :param image_format: Output format (PNG or JPEG/JPG).
526    :param flatten_transparency: When True, alpha is composited onto white and
527        kept as JPEG; when False, transparent sources are emitted as PNG.
528    """
529    thumb_data, _cache_filepath = await _get_image_thumb(
530        mass,
531        path_or_url,
532        size,
533        provider,
534        image_format,
535        flatten_transparency,
536    )
537    return thumb_data
538
539
540async def get_image_thumb_path(
541    mass: MusicAssistant,
542    path_or_url: str,
543    size: int | None,
544    provider: str,
545    image_format: str = "PNG",
546    flatten_transparency: bool = False,
547) -> str:
548    """
549    Get the absolute on-disk cache path for a thumbnail.
550
551    Unlike :func:`get_image_thumb`, cache persistence is required and any
552    filesystem error is raised to the caller.
553
554    :param mass: The MusicAssistant instance.
555    :param path_or_url: Path or URL to the source image.
556    :param size: Target thumbnail size (square), or None for original.
557    :param provider: Provider identifier for the image source.
558    :param image_format: Output format (PNG or JPEG/JPG).
559    :param flatten_transparency: When True, alpha is composited onto white and
560        kept as JPEG; when False, transparent sources are emitted as PNG.
561    """
562    thumb_data, cache_filepath = await _get_image_thumb(
563        mass,
564        path_or_url,
565        size,
566        provider,
567        image_format,
568        flatten_transparency,
569    )
570    await _ensure_thumb_on_disk(mass, cache_filepath, thumb_data)
571    return cache_filepath
572
573
574async def _get_image_thumb(
575    mass: MusicAssistant,
576    path_or_url: str,
577    size: int | None,
578    provider: str,
579    image_format: str,
580    flatten_transparency: bool,
581) -> tuple[bytes, str]:
582    """
583    Resolve thumbnail bytes and their validated cache path.
584
585    :param mass: The MusicAssistant instance.
586    :param path_or_url: Path or URL to the source image.
587    :param size: Target thumbnail size (square), or None for original.
588    :param provider: Provider identifier for the image source.
589    :param image_format: Output format (PNG or JPEG/JPG).
590    :param flatten_transparency: Whether to flatten alpha onto white for JPEG output.
591    """
592    image_format = image_format.upper()
593    if image_format == "JPG":
594        image_format = "JPEG"
595    if image_format not in _ALLOWED_THUMB_FORMATS:
596        msg = f"Unsupported thumbnail format: {image_format}"
597        raise ValueError(msg)
598
599    thumb_hash = create_thumb_hash(provider, path_or_url)
600    cache_filename = _thumb_cache_filename(thumb_hash, size, image_format, flatten_transparency)
601    if not _THUMB_FILENAME_RE.fullmatch(cache_filename):
602        # cache_filename is built from a sha256 + int + fixed extension, so this
603        # is unreachable in practice — it is here so a future change to either
604        # builder cannot silently let an unsafe value reach the filesystem path
605        msg = f"Refusing to use unexpected cache filename: {cache_filename!r}"
606        raise OSError(msg)
607
608    cache_filepath = _thumb_cache_filepath(mass, cache_filename)
609
610    # 1. Check in-memory FIFO cache
611    if (cached := _get_from_memory_cache(cache_filename)) is not None:
612        return cached, cache_filepath
613
614    # 2. Check on-disk cache
615    if await asyncio.to_thread(os.path.isfile, cache_filepath):
616        try:
617            async with aiofiles.open(cache_filepath, "rb") as f:
618                thumb_data = cast("bytes", await f.read())
619        except FileNotFoundError:
620            pass
621        else:
622            _put_in_memory_cache(cache_filename, thumb_data)
623            return thumb_data, cache_filepath
624
625    # 3. Generate thumbnail (de-duplicated across concurrent requests)
626    task: asyncio.Task[bytes] = mass.create_task(
627        _generate_and_cache_thumb,
628        mass,
629        path_or_url,
630        size,
631        provider,
632        image_format,
633        cache_filepath,
634        flatten_transparency,
635        task_id=f"thumb.{cache_filename}",
636        abort_existing=False,
637    )
638    thumb_data = await join_task(task)
639    _put_in_memory_cache(cache_filename, thumb_data)
640    return thumb_data, cache_filepath
641
642
643async def _generate_and_cache_thumb(
644    mass: MusicAssistant,
645    path_or_url: str,
646    size: int | None,
647    provider: str,
648    image_format: str,
649    cache_filepath: str,
650    flatten_transparency: bool = False,
651) -> bytes:
652    """
653    Generate a thumbnail, persist it on disk, and return the bytes.
654
655    :param mass: The MusicAssistant instance.
656    :param path_or_url: Path or URL to the source image.
657    :param size: Target thumbnail size (square), or None for original.
658    :param provider: Provider identifier for the image source.
659    :param image_format: Normalized output format (PNG or JPEG).
660    :param cache_filepath: Absolute path where the thumbnail will be stored.
661    :param flatten_transparency: When True, alpha is composited onto white and
662        kept as JPEG; when False, transparent sources are emitted as PNG.
663    """
664    img_data = await get_image_data(mass, path_or_url, provider)
665    if not img_data or not isinstance(img_data, bytes):
666        raise FileNotFoundError(f"Image not found: {path_or_url}")
667
668    if is_svg_data(img_data):
669        # Pillow can't decode SVG; pass it through unchanged.
670        thumb_data = img_data
671    elif not size and image_format.encode() in img_data:
672        thumb_data = img_data
673    else:
674
675        def _create_image() -> bytes:
676            data = BytesIO()
677            try:
678                img: ImageClass = Image.open(BytesIO(img_data))
679            except UnidentifiedImageError:
680                raise FileNotFoundError(f"Invalid image: {path_or_url}")
681            if size:
682                img.thumbnail((size, size), Image.Resampling.LANCZOS)
683            target_format = image_format
684            if target_format == "JPEG" and _has_alpha(img):
685                if flatten_transparency:
686                    # composite onto white, else the alpha would flatten to black
687                    background = Image.new("RGBA", img.size, (255, 255, 255, 255))
688                    background.alpha_composite(img.convert("RGBA"))
689                    img = background
690                else:
691                    target_format = "PNG"
692            mode = "RGBA" if target_format == "PNG" else "RGB"
693            converted = img.convert(mode)
694            if target_format == "JPEG":
695                converted.save(data, target_format, quality=95, optimize=False)
696            else:
697                converted.save(data, target_format, optimize=False)
698            return data.getvalue()
699
700        thumb_data = await asyncio.to_thread(_create_image)
701
702    # Persist to disk cache (best-effort, don't fail on I/O errors).
703    with contextlib.suppress(OSError):
704        await _write_thumb_to_disk(mass, cache_filepath, thumb_data)
705
706    return thumb_data
707
708
709async def cleanup_thumb_cache(cache_path: str, max_size_bytes: int) -> int:
710    """
711    Remove oldest cached thumbnails when total size exceeds the limit.
712
713    :param cache_path: The base cache directory (mass.cache_path).
714    :param max_size_bytes: Maximum allowed total size in bytes.
715    :returns: Number of files removed.
716    """
717    thumb_dir = os.path.join(cache_path, _THUMB_CACHE_DIR)
718
719    def _cleanup() -> int:
720        if not os.path.isdir(thumb_dir):
721            return 0
722        entries = []
723        for entry in os.scandir(thumb_dir):
724            if entry.is_file():
725                stat = entry.stat()
726                entries.append((entry.path, stat.st_size, stat.st_mtime))
727        entries.sort(key=lambda e: e[2])
728        total_size = sum(e[1] for e in entries)
729        removed = 0
730        for filepath, file_size, _ in entries:
731            if total_size <= max_size_bytes:
732                break
733            try:
734                Path(filepath).unlink()
735                total_size -= file_size
736                removed += 1
737            except OSError:
738                pass
739        return removed
740
741    return await asyncio.to_thread(_cleanup)
742
743
744async def invalidate_cached_image(mass: MusicAssistant, provider: str, path_or_url: str) -> None:
745    """
746    Remove all cached artifacts (thumbnails + source bytes) for an image.
747
748    Used when the image content behind an unchanged (provider, path) identity
749    has changed, e.g. a local file whose (embedded) artwork was replaced.
750
751    :param mass: The MusicAssistant instance.
752    :param provider: Provider id exactly as used when the image was requested.
753    :param path_or_url: Image path or URL exactly as used when the image was requested.
754    """
755    thumb_hash = create_thumb_hash(provider, path_or_url)
756    prefix = f"{thumb_hash}_"
757    for key in [key for key in _thumb_memory_cache if key.startswith(prefix)]:
758        _thumb_memory_cache.pop(key, None)
759    _source_memory_cache.pop(thumb_hash)
760    _failed_sources.pop(thumb_hash, None)
761
762    thumb_dir = os.path.realpath(os.path.join(mass.cache_path, _THUMB_CACHE_DIR))
763
764    def _remove_disk_entries() -> None:
765        # covers every size/format/flatten thumb variant plus the `_src` entry
766        if not os.path.isdir(thumb_dir):
767            return
768        for entry in os.scandir(thumb_dir):
769            if entry.name.startswith(prefix) and entry.is_file():
770                with contextlib.suppress(OSError):
771                    Path(entry.path).unlink()
772
773    await asyncio.to_thread(_remove_disk_entries)
774
775
776async def create_collage(
777    mass: MusicAssistant,
778    images: Iterable[MediaItemImage],
779    dimensions: tuple[int, int] = (1500, 1500),
780) -> bytes:
781    """Create a basic collage image from multiple image urls."""
782    image_size = 250
783
784    def _new_collage() -> ImageClass:
785        return Image.new("RGB", (dimensions[0], dimensions[1]), color=(255, 255, 255, 255))
786
787    collage = await asyncio.to_thread(_new_collage)
788
789    def _add_to_collage(img_data: bytes, coord_x: int, coord_y: int) -> None:
790        data = BytesIO(img_data)
791        photo = Image.open(data).convert("RGB")
792        photo = photo.resize((image_size, image_size))
793        collage.paste(photo, (coord_x, coord_y))
794        del data
795
796    # prevent duplicates with a set
797    images = list(set(images))
798    # warm the source cache with bounded concurrency and drop images that can't
799    # be fetched, so the (serial) tile loop below is served from cache
800    fetch_limiter = asyncio.Semaphore(8)
801
802    async def _warm_source_cache(img: MediaItemImage) -> MediaItemImage | None:
803        async with fetch_limiter:
804            try:
805                await get_image_data(mass, img.path, img.provider)
806            except FileNotFoundError, MusicAssistantError:
807                return None
808            return img
809
810    usable_images = [
811        img for img in await asyncio.gather(*map(_warm_source_cache, images)) if img is not None
812    ]
813    if not usable_images:
814        msg = "None of the collage images could be fetched"
815        raise FileNotFoundError(msg)
816    random.shuffle(usable_images)
817    iter_images = itertools.cycle(usable_images)
818
819    for x_co in range(0, dimensions[0], image_size):
820        for y_co in range(0, dimensions[1], image_size):
821            # try a few candidates per tile: a fetched image can still fail to decode
822            for _ in range(5):
823                img = next(iter_images)
824                try:
825                    img_data = await get_image_data(mass, img.path, img.provider)
826                    await asyncio.to_thread(_add_to_collage, img_data, x_co, y_co)
827                except FileNotFoundError, MusicAssistantError, UnidentifiedImageError:
828                    continue
829                del img_data
830                break
831
832    def _save_collage() -> bytes:
833        final_data = BytesIO()
834        collage.convert("RGB").save(final_data, "JPEG", optimize=True)
835        return final_data.getvalue()
836
837    return await asyncio.to_thread(_save_collage)
838
839
840async def load_provider_icon(icon_path: str) -> tuple[str, bytes]:
841    """
842    Load a provider icon file and return its mime type and bytes.
843
844    :param icon_path: Path to an svg or (transparent) png icon file.
845    """
846    ext = icon_path.rsplit(".", maxsplit=1)[-1].lower()
847    if ext == "svg":
848        async with aiofiles.open(icon_path, encoding="utf-8") as svg_file:
849            xml_data = (await svg_file.read()).replace("\n", "").strip()
850        return "image/svg+xml", xml_data.encode("utf-8")
851    if ext == "png":
852        async with aiofiles.open(icon_path, "rb") as png_file:
853            return "image/png", await png_file.read()
854    msg = f"Unsupported provider icon format: {ext}"
855    raise ValueError(msg)
856
857
858async def detect_provider_icons(
859    provider_path: str,
860) -> dict[ProviderIconVariant, tuple[str, bytes]]:
861    """
862    Detect the provider icon variants present in a provider directory.
863
864    Svg is preferred over png when both exist for the same variant.
865
866    :param provider_path: Path to the provider directory to scan.
867    """
868    variant_files = {
869        ProviderIconVariant.DEFAULT: ("icon.svg", "icon.png"),
870        ProviderIconVariant.DARK: ("icon_dark.svg", "icon_dark.png"),
871        ProviderIconVariant.MONOCHROME: ("icon_monochrome.svg", "icon_monochrome.png"),
872    }
873    icons: dict[ProviderIconVariant, tuple[str, bytes]] = {}
874    for variant, filenames in variant_files.items():
875        for filename in filenames:  # svg first -> preferred
876            icon_path = os.path.join(provider_path, filename)
877            if await aiofiles.os.path.isfile(icon_path):
878                icons[variant] = await load_provider_icon(icon_path)
879                break
880    return icons
881
882
883def _thumb_cache_filepath(mass: MusicAssistant, cache_filename: str) -> str:
884    """Return a validated absolute path for a thumbnail cache filename."""
885    thumb_dir = os.path.realpath(os.path.join(mass.cache_path, _THUMB_CACHE_DIR))
886    cache_filepath = os.path.realpath(os.path.join(thumb_dir, cache_filename))
887    if not cache_filepath.startswith(thumb_dir + os.sep):
888        msg = f"Cache path escapes thumbnail directory: {cache_filepath}"
889        raise OSError(msg)
890    return cache_filepath
891
892
893async def _ensure_thumb_on_disk(
894    mass: MusicAssistant, cache_filepath: str, thumb_data: bytes
895) -> None:
896    """
897    Ensure thumbnail bytes are fully persisted at their cache path.
898
899    :param mass: The MusicAssistant instance.
900    :param cache_filepath: Validated absolute thumbnail cache path.
901    :param thumb_data: Complete encoded thumbnail bytes.
902    """
903
904    def _is_complete() -> bool:
905        path = Path(cache_filepath)
906        try:
907            return path.is_file() and path.stat().st_size == len(thumb_data)
908        except OSError:
909            return False
910
911    if await asyncio.to_thread(_is_complete):
912        return
913    await _write_thumb_to_disk(mass, cache_filepath, thumb_data)
914    if not await asyncio.to_thread(_is_complete):
915        msg = f"Thumbnail cache file was not persisted: {cache_filepath}"
916        raise OSError(msg)
917
918
919async def _write_thumb_to_disk(
920    mass: MusicAssistant, cache_filepath: str, thumb_data: bytes
921) -> None:
922    """
923    Atomically persist thumbnail bytes to their validated cache path.
924
925    :param mass: The MusicAssistant instance.
926    :param cache_filepath: Validated absolute thumbnail cache path.
927    :param thumb_data: Complete encoded thumbnail bytes.
928    """
929    resolved = os.path.realpath(cache_filepath)
930    thumb_dir = os.path.realpath(os.path.join(mass.cache_path, _THUMB_CACHE_DIR))
931    if not resolved.startswith(thumb_dir + os.sep):
932        msg = f"Cache path escapes thumbnail directory: {resolved}"
933        raise OSError(msg)
934    await asyncio.to_thread(os.makedirs, thumb_dir, exist_ok=True)
935    file_descriptor, temp_filepath = await asyncio.to_thread(
936        tempfile.mkstemp, dir=thumb_dir, prefix=".thumb-"
937    )
938    await asyncio.to_thread(os.close, file_descriptor)
939    try:
940        async with aiofiles.open(temp_filepath, "wb") as temp_file:
941            await temp_file.write(thumb_data)
942        await asyncio.to_thread(os.replace, temp_filepath, resolved)
943    finally:
944        with contextlib.suppress(OSError):
945            await asyncio.to_thread(Path(temp_filepath).unlink)
946