diff --git a/music_assistant/helpers/images.py b/music_assistant/helpers/images.py index dca001a5c2..598bebcac1 100644 --- a/music_assistant/helpers/images.py +++ b/music_assistant/helpers/images.py @@ -6,6 +6,8 @@ import itertools import os import random +import re +import urllib.parse from base64 import b64decode from collections.abc import Iterable from io import BytesIO @@ -28,8 +30,52 @@ from music_assistant.mass import MusicAssistant -async def get_image_data(mass: MusicAssistant, path_or_url: str, provider: str) -> bytes: - """Create thumbnail from image url.""" +def _extract_imageproxy_params(url: str) -> tuple[str, str] | None: + """ + Extract path and provider from an imageproxy URL. + + :param url: The URL to check for imageproxy format. + :return: Tuple of (path, provider) if this is an imageproxy URL, None otherwise. + """ + if "/imageproxy?" not in url: + return None + + try: + parsed = urllib.parse.urlparse(url) + query_params = urllib.parse.parse_qs(parsed.query) + + path = query_params.get("path", [None])[0] + provider = query_params.get("provider", ["builtin"])[0] + + if path: + # The path may be double URL-encoded, decode it + decoded_path = urllib.parse.unquote_plus(path) + if re.search(r"%[0-9A-Fa-f]{2}", decoded_path): + decoded_path = urllib.parse.unquote_plus(decoded_path) + return (decoded_path, provider) + except (KeyError, ValueError, IndexError): + # URL parsing failed, not an imageproxy URL + pass + + return None + + +async def get_image_data( + mass: MusicAssistant, path_or_url: str, provider: str, *, _depth: int = 0 +) -> bytes: + """ + Retrieve image data from a path or URL. + + :param mass: The MusicAssistant instance. + :param path_or_url: The image path, URL, or base64 data URI. + :param provider: The provider ID that can resolve the image. + :param _depth: Internal recursion depth counter (do not set manually). + """ + max_recursion_depth = 5 + if _depth > max_recursion_depth: + msg = f"Maximum recursion depth exceeded when fetching image: {path_or_url}" + raise FileNotFoundError(msg) + # TODO: add local cache here ! if prov := mass.get_provider(provider): assert isinstance(prov, MusicProvider | MetadataProvider | PluginProvider) @@ -40,11 +86,22 @@ async def get_image_data(mass: MusicAssistant, path_or_url: str, provider: str) path_or_url = resolved_image # handle HTTP location if path_or_url.startswith("http"): + # Handle imageproxy URLs pointing to our own server + parsed_url = urllib.parse.urlparse(path_or_url) + url_host = f"{parsed_url.scheme}://{parsed_url.netloc}" + server_base_urls = {mass.webserver.base_url, mass.streams.base_url} + if url_host in server_base_urls: + if imageproxy_params := _extract_imageproxy_params(path_or_url): + extracted_path, extracted_provider = imageproxy_params + return await get_image_data( + mass, extracted_path, extracted_provider, _depth=_depth + 1 + ) try: async with mass.http_session_no_ssl.get(path_or_url, raise_for_status=True) as resp: return await resp.read() except ClientError as err: - raise FileNotFoundError from err + msg = f"Failed to fetch image from {path_or_url}: {err}" + raise FileNotFoundError(msg) from err # handle base64 embedded images if path_or_url.startswith("data:image"): return b64decode(path_or_url.split(",")[-1])