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 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 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        _store_failed_source(cache_key, str(err))
413        LOGGER.warning("%s (not retrying for %s seconds)", err, _FAILED_SOURCE_TTL)
414        raise
415    _failed_sources.pop(cache_key, None)
416    _source_memory_cache.put(cache_key, img_data)
417    if disk_cacheable:
418        # persist to disk cache (best-effort, don't fail on I/O errors)
419        try:
420            await asyncio.to_thread(os.makedirs, os.path.dirname(filepath), exist_ok=True)
421            async with aiofiles.open(filepath, "wb") as _file:
422                await _file.write(img_data)
423        except OSError:
424            pass
425    return img_data
426
427
428async def _fetch_source_image(
429    mass: MusicAssistant, path_or_url: str, provider: str, depth: int
430) -> tuple[bytes, bool]:
431    """
432    Fetch image bytes from their origin.
433
434    Returns the raw bytes plus whether they may be persisted on disk under this
435    (provider, path) cache key. That is True for every origin except results
436    resolved under a different key (imageproxy URLs pointing at our own server):
437    an alias-keyed copy would keep being served after the underlying image is
438    invalidated.
439
440    :param mass: The MusicAssistant instance.
441    :param path_or_url: The image path or URL.
442    :param provider: The provider ID that can resolve the image.
443    :param depth: Recursion depth of the originating get_image_data call.
444    """
445    if prov := mass.get_provider(provider):
446        assert isinstance(prov, MusicProvider | MetadataProvider | PlayerProvider | PluginProvider)
447        if resolved_image := await prov.resolve_image(path_or_url):
448            if isinstance(resolved_image, bytes):
449                return resolved_image, True
450            if isinstance(resolved_image, str):
451                path_or_url = resolved_image
452    # handle HTTP location
453    if path_or_url.startswith("http"):
454        # handle imageproxy URLs pointing to our own server
455        if resolved := await _resolve_own_imageproxy_url(mass, path_or_url):
456            extracted_provider, extracted_path = resolved
457            # route through the public entrypoint so the result is cached under
458            # the resolved key; don't persist it under this alias key as well
459            return (
460                await get_image_data(mass, extracted_path, extracted_provider, _depth=depth + 1),
461                False,
462            )
463        try:
464            return await _fetch_remote_image(mass, path_or_url), True
465        except ClientError as err:
466            msg = f"Failed to fetch image from {path_or_url}: {err}"
467            raise FileNotFoundError(msg) from err
468    # handle base64 embedded images
469    if path_or_url.startswith("data:image"):
470        return b64decode(path_or_url.split(",")[-1]), True
471    # handle FILE location (of type image)
472    if path_or_url.endswith(("jpg", "JPG", "png", "PNG", "jpeg", "svg", "SVG")) and is_safe_path(
473        path_or_url
474    ):
475        if await asyncio.to_thread(os.path.isfile, path_or_url):
476            async with aiofiles.open(path_or_url, "rb") as _file:
477                return cast("bytes", await _file.read()), True
478    # use ffmpeg for embedded images
479    if is_safe_path(path_or_url) and (img_data := await get_embedded_image(path_or_url)):
480        return img_data, True
481    msg = f"Image not found: {path_or_url}"
482    raise FileNotFoundError(msg)
483
484
485async def _fetch_remote_image(mass: MusicAssistant, url: str) -> bytes:
486    """
487    Fetch raw image bytes over HTTP.
488
489    :param mass: The MusicAssistant instance.
490    :param url: The (http/https) image URL to fetch.
491    """
492    # Bot-protected CDNs (e.g. Akamai) reject our normal self-identifying User-Agent,
493    # and even regular browser User-Agents, while still serving well-known fetch tools.
494    # We keep identifying as Music Assistant but carry a Wget compatibility token, which
495    # such CDNs allowlist, so artwork is served on the first (and only) request.
496    user_agent = f"{APPLICATION_NAME}/{mass.version} (Wget/1.24.5; +https://music-assistant.io)"
497    async with mass.http_session_no_ssl.get(
498        url, raise_for_status=True, headers={"User-Agent": user_agent}
499    ) as resp:
500        return await resp.read()
501
502
503async def get_image_thumb(
504    mass: MusicAssistant,
505    path_or_url: str,
506    size: int | None,
507    provider: str,
508    image_format: str = "PNG",
509    flatten_transparency: bool = False,
510) -> bytes:
511    """
512    Get (optimized) thumbnail from image url.
513
514    Uses a two-tier cache (in-memory FIFO + on-disk) keyed by a hash of
515    provider + path so that repeated requests never trigger ffmpeg or
516    PIL processing again.  Concurrent requests for the same thumbnail
517    are de-duplicated via create_task.
518
519    :param mass: The MusicAssistant instance.
520    :param path_or_url: Path or URL to the source image.
521    :param size: Target thumbnail size (square), or None for original.
522    :param provider: Provider identifier for the image source.
523    :param image_format: Output format (PNG or JPEG/JPG).
524    :param flatten_transparency: When True, alpha is composited onto white and
525        kept as JPEG; when False, transparent sources are emitted as PNG.
526    """
527    thumb_data, _cache_filepath = await _get_image_thumb(
528        mass,
529        path_or_url,
530        size,
531        provider,
532        image_format,
533        flatten_transparency,
534    )
535    return thumb_data
536
537
538async def get_image_thumb_path(
539    mass: MusicAssistant,
540    path_or_url: str,
541    size: int | None,
542    provider: str,
543    image_format: str = "PNG",
544    flatten_transparency: bool = False,
545) -> str:
546    """
547    Get the absolute on-disk cache path for a thumbnail.
548
549    Unlike :func:`get_image_thumb`, cache persistence is required and any
550    filesystem error is raised to the caller.
551
552    :param mass: The MusicAssistant instance.
553    :param path_or_url: Path or URL to the source image.
554    :param size: Target thumbnail size (square), or None for original.
555    :param provider: Provider identifier for the image source.
556    :param image_format: Output format (PNG or JPEG/JPG).
557    :param flatten_transparency: When True, alpha is composited onto white and
558        kept as JPEG; when False, transparent sources are emitted as PNG.
559    """
560    thumb_data, cache_filepath = await _get_image_thumb(
561        mass,
562        path_or_url,
563        size,
564        provider,
565        image_format,
566        flatten_transparency,
567    )
568    await _ensure_thumb_on_disk(mass, cache_filepath, thumb_data)
569    return cache_filepath
570
571
572async def _get_image_thumb(
573    mass: MusicAssistant,
574    path_or_url: str,
575    size: int | None,
576    provider: str,
577    image_format: str,
578    flatten_transparency: bool,
579) -> tuple[bytes, str]:
580    """
581    Resolve thumbnail bytes and their validated cache path.
582
583    :param mass: The MusicAssistant instance.
584    :param path_or_url: Path or URL to the source image.
585    :param size: Target thumbnail size (square), or None for original.
586    :param provider: Provider identifier for the image source.
587    :param image_format: Output format (PNG or JPEG/JPG).
588    :param flatten_transparency: Whether to flatten alpha onto white for JPEG output.
589    """
590    image_format = image_format.upper()
591    if image_format == "JPG":
592        image_format = "JPEG"
593    if image_format not in _ALLOWED_THUMB_FORMATS:
594        msg = f"Unsupported thumbnail format: {image_format}"
595        raise ValueError(msg)
596
597    thumb_hash = create_thumb_hash(provider, path_or_url)
598    cache_filename = _thumb_cache_filename(thumb_hash, size, image_format, flatten_transparency)
599    if not _THUMB_FILENAME_RE.fullmatch(cache_filename):
600        # cache_filename is built from a sha256 + int + fixed extension, so this
601        # is unreachable in practice — it is here so a future change to either
602        # builder cannot silently let an unsafe value reach the filesystem path
603        msg = f"Refusing to use unexpected cache filename: {cache_filename!r}"
604        raise OSError(msg)
605
606    cache_filepath = _thumb_cache_filepath(mass, cache_filename)
607
608    # 1. Check in-memory FIFO cache
609    if (cached := _get_from_memory_cache(cache_filename)) is not None:
610        return cached, cache_filepath
611
612    # 2. Check on-disk cache
613    if await asyncio.to_thread(os.path.isfile, cache_filepath):
614        try:
615            async with aiofiles.open(cache_filepath, "rb") as f:
616                thumb_data = cast("bytes", await f.read())
617        except FileNotFoundError:
618            pass
619        else:
620            _put_in_memory_cache(cache_filename, thumb_data)
621            return thumb_data, cache_filepath
622
623    # 3. Generate thumbnail (de-duplicated across concurrent requests)
624    task: asyncio.Task[bytes] = mass.create_task(
625        _generate_and_cache_thumb,
626        mass,
627        path_or_url,
628        size,
629        provider,
630        image_format,
631        cache_filepath,
632        flatten_transparency,
633        task_id=f"thumb.{cache_filename}",
634        abort_existing=False,
635    )
636    thumb_data = await join_task(task)
637    _put_in_memory_cache(cache_filename, thumb_data)
638    return thumb_data, cache_filepath
639
640
641async def _generate_and_cache_thumb(
642    mass: MusicAssistant,
643    path_or_url: str,
644    size: int | None,
645    provider: str,
646    image_format: str,
647    cache_filepath: str,
648    flatten_transparency: bool = False,
649) -> bytes:
650    """
651    Generate a thumbnail, persist it on disk, and return the bytes.
652
653    :param mass: The MusicAssistant instance.
654    :param path_or_url: Path or URL to the source image.
655    :param size: Target thumbnail size (square), or None for original.
656    :param provider: Provider identifier for the image source.
657    :param image_format: Normalized output format (PNG or JPEG).
658    :param cache_filepath: Absolute path where the thumbnail will be stored.
659    :param flatten_transparency: When True, alpha is composited onto white and
660        kept as JPEG; when False, transparent sources are emitted as PNG.
661    """
662    img_data = await get_image_data(mass, path_or_url, provider)
663    if not img_data or not isinstance(img_data, bytes):
664        raise FileNotFoundError(f"Image not found: {path_or_url}")
665
666    if is_svg_data(img_data):
667        # Pillow can't decode SVG; pass it through unchanged.
668        thumb_data = img_data
669    elif not size and image_format.encode() in img_data:
670        thumb_data = img_data
671    else:
672
673        def _create_image() -> bytes:
674            data = BytesIO()
675            try:
676                img: ImageClass = Image.open(BytesIO(img_data))
677            except UnidentifiedImageError:
678                raise FileNotFoundError(f"Invalid image: {path_or_url}")
679            if size:
680                img.thumbnail((size, size), Image.Resampling.LANCZOS)
681            target_format = image_format
682            if target_format == "JPEG" and _has_alpha(img):
683                if flatten_transparency:
684                    # composite onto white, else the alpha would flatten to black
685                    background = Image.new("RGBA", img.size, (255, 255, 255, 255))
686                    background.alpha_composite(img.convert("RGBA"))
687                    img = background
688                else:
689                    target_format = "PNG"
690            mode = "RGBA" if target_format == "PNG" else "RGB"
691            converted = img.convert(mode)
692            if target_format == "JPEG":
693                converted.save(data, target_format, quality=95, optimize=False)
694            else:
695                converted.save(data, target_format, optimize=False)
696            return data.getvalue()
697
698        thumb_data = await asyncio.to_thread(_create_image)
699
700    # Persist to disk cache (best-effort, don't fail on I/O errors).
701    with contextlib.suppress(OSError):
702        await _write_thumb_to_disk(mass, cache_filepath, thumb_data)
703
704    return thumb_data
705
706
707async def cleanup_thumb_cache(cache_path: str, max_size_bytes: int) -> int:
708    """
709    Remove oldest cached thumbnails when total size exceeds the limit.
710
711    :param cache_path: The base cache directory (mass.cache_path).
712    :param max_size_bytes: Maximum allowed total size in bytes.
713    :returns: Number of files removed.
714    """
715    thumb_dir = os.path.join(cache_path, _THUMB_CACHE_DIR)
716
717    def _cleanup() -> int:
718        if not os.path.isdir(thumb_dir):
719            return 0
720        entries = []
721        for entry in os.scandir(thumb_dir):
722            if entry.is_file():
723                stat = entry.stat()
724                entries.append((entry.path, stat.st_size, stat.st_mtime))
725        entries.sort(key=lambda e: e[2])
726        total_size = sum(e[1] for e in entries)
727        removed = 0
728        for filepath, file_size, _ in entries:
729            if total_size <= max_size_bytes:
730                break
731            try:
732                Path(filepath).unlink()
733                total_size -= file_size
734                removed += 1
735            except OSError:
736                pass
737        return removed
738
739    return await asyncio.to_thread(_cleanup)
740
741
742async def invalidate_cached_image(mass: MusicAssistant, provider: str, path_or_url: str) -> None:
743    """
744    Remove all cached artifacts (thumbnails + source bytes) for an image.
745
746    Used when the image content behind an unchanged (provider, path) identity
747    has changed, e.g. a local file whose (embedded) artwork was replaced.
748
749    :param mass: The MusicAssistant instance.
750    :param provider: Provider id exactly as used when the image was requested.
751    :param path_or_url: Image path or URL exactly as used when the image was requested.
752    """
753    thumb_hash = create_thumb_hash(provider, path_or_url)
754    prefix = f"{thumb_hash}_"
755    for key in [key for key in _thumb_memory_cache if key.startswith(prefix)]:
756        _thumb_memory_cache.pop(key, None)
757    _source_memory_cache.pop(thumb_hash)
758    _failed_sources.pop(thumb_hash, None)
759
760    thumb_dir = os.path.realpath(os.path.join(mass.cache_path, _THUMB_CACHE_DIR))
761
762    def _remove_disk_entries() -> None:
763        # covers every size/format/flatten thumb variant plus the `_src` entry
764        if not os.path.isdir(thumb_dir):
765            return
766        for entry in os.scandir(thumb_dir):
767            if entry.name.startswith(prefix) and entry.is_file():
768                with contextlib.suppress(OSError):
769                    Path(entry.path).unlink()
770
771    await asyncio.to_thread(_remove_disk_entries)
772
773
774async def create_collage(
775    mass: MusicAssistant,
776    images: Iterable[MediaItemImage],
777    dimensions: tuple[int, int] = (1500, 1500),
778) -> bytes:
779    """Create a basic collage image from multiple image urls."""
780    image_size = 250
781
782    def _new_collage() -> ImageClass:
783        return Image.new("RGB", (dimensions[0], dimensions[1]), color=(255, 255, 255, 255))
784
785    collage = await asyncio.to_thread(_new_collage)
786
787    def _add_to_collage(img_data: bytes, coord_x: int, coord_y: int) -> None:
788        data = BytesIO(img_data)
789        photo = Image.open(data).convert("RGB")
790        photo = photo.resize((image_size, image_size))
791        collage.paste(photo, (coord_x, coord_y))
792        del data
793
794    # prevent duplicates with a set
795    images = list(set(images))
796    # warm the source cache with bounded concurrency and drop images that can't
797    # be fetched, so the (serial) tile loop below is served from cache
798    fetch_limiter = asyncio.Semaphore(8)
799
800    async def _warm_source_cache(img: MediaItemImage) -> MediaItemImage | None:
801        async with fetch_limiter:
802            try:
803                await get_image_data(mass, img.path, img.provider)
804            except FileNotFoundError, MusicAssistantError:
805                return None
806            return img
807
808    usable_images = [
809        img for img in await asyncio.gather(*map(_warm_source_cache, images)) if img is not None
810    ]
811    if not usable_images:
812        msg = "None of the collage images could be fetched"
813        raise FileNotFoundError(msg)
814    random.shuffle(usable_images)
815    iter_images = itertools.cycle(usable_images)
816
817    for x_co in range(0, dimensions[0], image_size):
818        for y_co in range(0, dimensions[1], image_size):
819            # try a few candidates per tile: a fetched image can still fail to decode
820            for _ in range(5):
821                img = next(iter_images)
822                try:
823                    img_data = await get_image_data(mass, img.path, img.provider)
824                    await asyncio.to_thread(_add_to_collage, img_data, x_co, y_co)
825                except FileNotFoundError, MusicAssistantError, UnidentifiedImageError:
826                    continue
827                del img_data
828                break
829
830    def _save_collage() -> bytes:
831        final_data = BytesIO()
832        collage.convert("RGB").save(final_data, "JPEG", optimize=True)
833        return final_data.getvalue()
834
835    return await asyncio.to_thread(_save_collage)
836
837
838async def load_provider_icon(icon_path: str) -> tuple[str, bytes]:
839    """
840    Load a provider icon file and return its mime type and bytes.
841
842    :param icon_path: Path to an svg or (transparent) png icon file.
843    """
844    ext = icon_path.rsplit(".", maxsplit=1)[-1].lower()
845    if ext == "svg":
846        async with aiofiles.open(icon_path, encoding="utf-8") as svg_file:
847            xml_data = (await svg_file.read()).replace("\n", "").strip()
848        return "image/svg+xml", xml_data.encode("utf-8")
849    if ext == "png":
850        async with aiofiles.open(icon_path, "rb") as png_file:
851            return "image/png", await png_file.read()
852    msg = f"Unsupported provider icon format: {ext}"
853    raise ValueError(msg)
854
855
856async def detect_provider_icons(
857    provider_path: str,
858) -> dict[ProviderIconVariant, tuple[str, bytes]]:
859    """
860    Detect the provider icon variants present in a provider directory.
861
862    Svg is preferred over png when both exist for the same variant.
863
864    :param provider_path: Path to the provider directory to scan.
865    """
866    variant_files = {
867        ProviderIconVariant.DEFAULT: ("icon.svg", "icon.png"),
868        ProviderIconVariant.DARK: ("icon_dark.svg", "icon_dark.png"),
869        ProviderIconVariant.MONOCHROME: ("icon_monochrome.svg", "icon_monochrome.png"),
870    }
871    icons: dict[ProviderIconVariant, tuple[str, bytes]] = {}
872    for variant, filenames in variant_files.items():
873        for filename in filenames:  # svg first -> preferred
874            icon_path = os.path.join(provider_path, filename)
875            if await aiofiles.os.path.isfile(icon_path):
876                icons[variant] = await load_provider_icon(icon_path)
877                break
878    return icons
879
880
881def _thumb_cache_filepath(mass: MusicAssistant, cache_filename: str) -> str:
882    """Return a validated absolute path for a thumbnail cache filename."""
883    thumb_dir = os.path.realpath(os.path.join(mass.cache_path, _THUMB_CACHE_DIR))
884    cache_filepath = os.path.realpath(os.path.join(thumb_dir, cache_filename))
885    if not cache_filepath.startswith(thumb_dir + os.sep):
886        msg = f"Cache path escapes thumbnail directory: {cache_filepath}"
887        raise OSError(msg)
888    return cache_filepath
889
890
891async def _ensure_thumb_on_disk(
892    mass: MusicAssistant, cache_filepath: str, thumb_data: bytes
893) -> None:
894    """
895    Ensure thumbnail bytes are fully persisted at their cache path.
896
897    :param mass: The MusicAssistant instance.
898    :param cache_filepath: Validated absolute thumbnail cache path.
899    :param thumb_data: Complete encoded thumbnail bytes.
900    """
901
902    def _is_complete() -> bool:
903        path = Path(cache_filepath)
904        try:
905            return path.is_file() and path.stat().st_size == len(thumb_data)
906        except OSError:
907            return False
908
909    if await asyncio.to_thread(_is_complete):
910        return
911    await _write_thumb_to_disk(mass, cache_filepath, thumb_data)
912    if not await asyncio.to_thread(_is_complete):
913        msg = f"Thumbnail cache file was not persisted: {cache_filepath}"
914        raise OSError(msg)
915
916
917async def _write_thumb_to_disk(
918    mass: MusicAssistant, cache_filepath: str, thumb_data: bytes
919) -> None:
920    """
921    Atomically persist thumbnail bytes to their validated cache path.
922
923    :param mass: The MusicAssistant instance.
924    :param cache_filepath: Validated absolute thumbnail cache path.
925    :param thumb_data: Complete encoded thumbnail bytes.
926    """
927    resolved = os.path.realpath(cache_filepath)
928    thumb_dir = os.path.realpath(os.path.join(mass.cache_path, _THUMB_CACHE_DIR))
929    if not resolved.startswith(thumb_dir + os.sep):
930        msg = f"Cache path escapes thumbnail directory: {resolved}"
931        raise OSError(msg)
932    await asyncio.to_thread(os.makedirs, thumb_dir, exist_ok=True)
933    file_descriptor, temp_filepath = await asyncio.to_thread(
934        tempfile.mkstemp, dir=thumb_dir, prefix=".thumb-"
935    )
936    await asyncio.to_thread(os.close, file_descriptor)
937    try:
938        async with aiofiles.open(temp_filepath, "wb") as temp_file:
939            await temp_file.write(thumb_data)
940        await asyncio.to_thread(os.replace, temp_filepath, resolved)
941    finally:
942        with contextlib.suppress(OSError):
943            await asyncio.to_thread(Path(temp_filepath).unlink)
944