/
/
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.models.metadata_provider import MetadataProvider
15
16if TYPE_CHECKING:
17 from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
18 from music_assistant_models.media_items import Album
19 from music_assistant_models.provider import ProviderManifest
20
21 from music_assistant.mass import MusicAssistant
22 from music_assistant.models import ProviderInstanceType
23
24SUPPORTED_FEATURES = {
25 ProviderFeature.ALBUM_METADATA,
26}
27
28ITUNES_LOOKUP_URL = "https://itunes.apple.com/lookup"
29
30
31async def setup(
32 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
33) -> ProviderInstanceType:
34 """Initialize provider(instance) with given configuration."""
35 return ITunesArtworkMetadataProvider(mass, manifest, config, SUPPORTED_FEATURES)
36
37
38class ITunesArtworkMetadataProvider(MetadataProvider):
39 """
40 iTunes Artwork Metadata provider.
41
42 Fetches high-resolution album artwork from the iTunes catalog using UPC barcode lookup.
43 """
44
45 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
46 """Return Config entries to setup this provider."""
47 return ()
48
49 @property
50 def priority(self) -> int:
51 """Priority for this provider (lower = more preferred)."""
52 return 30
53
54 async def get_album_metadata(self, album: Album) -> MediaItemMetadata | None:
55 """
56 Retrieve metadata for an album.
57
58 :param album: Album to retrieve metadata for.
59 """
60 barcode = album.get_external_id(ExternalID.BARCODE)
61 if not barcode:
62 self.logger.debug(
63 "No barcode available for album %s, skipping iTunes lookup", album.name
64 )
65 return None
66
67 artwork_url = await self._get_artwork_url(barcode)
68 if not artwork_url:
69 return None
70
71 self.logger.debug("Found artwork for album %s on iTunes", album.name)
72 return MediaItemMetadata(
73 images=UniqueList(
74 [
75 MediaItemImage(
76 type=ImageType.THUMB,
77 path=artwork_url,
78 provider=self.domain,
79 remotely_accessible=True,
80 )
81 ]
82 )
83 )
84
85 @use_cache(86400 * 30)
86 async def _get_artwork_url(self, barcode: str) -> str | None:
87 """
88 Look up album artwork URL from iTunes by UPC barcode.
89
90 :param barcode: UPC/EAN barcode for the album.
91 """
92 # iTunes expects a UPC (12 digits), strip leading zero from EAN-13 if present
93 upc = barcode.lstrip("0").zfill(12) if len(barcode) == 13 else 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