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