/
/
/
1"""iTunes Artwork Metadata provider for Music Assistant."""
2
3from __future__ import annotations
4
5from json import JSONDecodeError
6from typing import TYPE_CHECKING
7
8import aiohttp
9from music_assistant_models.enums import ExternalID, ImageType, ProviderFeature
10from music_assistant_models.errors import ResourceTemporarilyUnavailable
11from music_assistant_models.media_items import MediaItemImage, MediaItemMetadata, UniqueList
12
13from music_assistant.controllers.cache import use_cache
14from music_assistant.helpers.external_ids import barcode_to_upc
15from music_assistant.models.metadata_provider import MetadataProvider
16
17if TYPE_CHECKING:
18 from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
19 from music_assistant_models.media_items import Album
20 from music_assistant_models.provider import ProviderManifest
21
22 from music_assistant.mass import MusicAssistant
23 from music_assistant.models import ProviderInstanceType
24
25SUPPORTED_FEATURES = {
26 ProviderFeature.ALBUM_METADATA,
27}
28
29ITUNES_LOOKUP_URL = "https://itunes.apple.com/lookup"
30
31
32async def setup(
33 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
34) -> ProviderInstanceType:
35 """Initialize provider(instance) with given configuration."""
36 return ITunesArtworkMetadataProvider(mass, manifest, config, SUPPORTED_FEATURES)
37
38
39class ITunesArtworkMetadataProvider(MetadataProvider):
40 """
41 iTunes Artwork Metadata provider.
42
43 Fetches high-resolution album artwork from the iTunes catalog using UPC barcode lookup.
44 """
45
46 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
47 """Return Config entries to setup this provider."""
48 return ()
49
50 @property
51 def priority(self) -> int:
52 """Priority for this provider (lower = more preferred)."""
53 return 30
54
55 async def get_album_metadata(self, album: Album) -> MediaItemMetadata | None:
56 """
57 Retrieve metadata for an album.
58
59 :param album: Album to retrieve metadata for.
60 """
61 barcode = album.get_external_id(ExternalID.BARCODE)
62 if not barcode:
63 self.logger.debug(
64 "No barcode available for album %s, skipping iTunes lookup", album.name
65 )
66 return None
67
68 artwork_url = await self._get_artwork_url(barcode)
69 if not artwork_url:
70 return None
71
72 self.logger.debug("Found artwork for album %s on iTunes", album.name)
73 return MediaItemMetadata(
74 images=UniqueList(
75 [
76 MediaItemImage(
77 type=ImageType.THUMB,
78 path=artwork_url,
79 provider=self.domain,
80 remotely_accessible=True,
81 )
82 ]
83 )
84 )
85
86 @use_cache(86400 * 30)
87 async def _get_artwork_url(self, barcode: str) -> str | None:
88 """
89 Look up album artwork URL from iTunes by UPC barcode.
90
91 :param barcode: UPC/EAN barcode for the album.
92 """
93 upc = barcode_to_upc(barcode)
94 try:
95 async with self.mass.http_session.get(
96 ITUNES_LOOKUP_URL, params={"upc": upc}
97 ) as response:
98 response.raise_for_status()
99 data = await response.json(content_type=None)
100 except (aiohttp.ClientError, TimeoutError, JSONDecodeError) as err:
101 # non-2xx / network / parse failure is transient â surface it as
102 # ResourceTemporarilyUnavailable so callers degrade instead of caching "no artwork"
103 raise ResourceTemporarilyUnavailable("iTunes request failed") from err
104
105 if not data.get("resultCount"):
106 self.logger.debug("No results from iTunes for barcode %s", barcode)
107 return None
108
109 result = data["results"][0]
110 artwork_url = result.get("artworkUrl100")
111 if not artwork_url:
112 self.logger.debug("No artwork URL in iTunes result for barcode %s", barcode)
113 return None
114
115 # Replace 100x100 with high-resolution 1500x1500
116 return str(artwork_url).replace("100x100bb", "1500x1500bb")
117