import pytube
import ytmusicapi
+from music_assistant.common.helpers.uri import create_uri
+from music_assistant.common.helpers.util import create_sort_name
from music_assistant.common.models.config_entries import ConfigEntry
from music_assistant.common.models.enums import ConfigEntryType, ProviderFeature
from music_assistant.common.models.errors import (
Artist,
ContentType,
ImageType,
+ ItemMapping,
MediaItemImage,
MediaType,
Playlist,
YT_DOMAIN = "https://www.youtube.com"
YTM_DOMAIN = "https://music.youtube.com"
YTM_BASE_URL = f"{YTM_DOMAIN}/youtubei/v1/"
+VARIOUS_ARTISTS_YTM_ID = "UCUTXlgdcKU5vfzFqHOWIvkA"
SUPPORTED_FEATURES = (
ProviderFeature.LIBRARY_ARTISTS,
album.metadata.explicit = album_obj["isExplicit"]
if "artists" in album_obj:
album.artists = [
- await self._parse_artist(artist)
+ self._get_artist_item_mapping(artist)
for artist in album_obj["artists"]
- # artist object may be missing an id
- # in that case its either a performer (like the composer) OR this
- # is a Various artists compilation album...
- if (artist.get("id") or artist["name"] == "Various Artists")
+ if artist.get("id")
+ or artist.get("channelId")
+ or artist.get("name") == "Various Artists"
]
if "type" in album_obj:
if album_obj["type"] == "Single":
elif "id" in artist_obj and artist_obj["id"]:
artist_id = artist_obj["id"]
elif artist_obj["name"] == "Various Artists":
- artist_id = "UCUTXlgdcKU5vfzFqHOWIvkA"
+ artist_id = VARIOUS_ARTISTS_YTM_ID
if not artist_id:
raise InvalidDataError("Artist does not have a valid ID")
artist = Artist(item_id=artist_id, name=artist_obj["name"], provider=self.domain)
track = Track(item_id=track_obj["videoId"], provider=self.domain, name=track_obj["title"])
if "artists" in track_obj:
track.artists = [
- await self._parse_artist(artist)
+ self._get_artist_item_mapping(artist)
for artist in track_obj["artists"]
if artist.get("id")
or artist.get("channelId")
track.metadata.images = await self._parse_thumbnails(track_obj["thumbnails"])
if (
track_obj.get("album")
- and track_obj.get("artists")
and isinstance(track_obj.get("album"), dict)
and track_obj["album"].get("id")
):
album = track_obj["album"]
- album["artists"] = track_obj["artists"]
- track.album = await self._parse_album(album, album["id"])
+ track.album = self._get_item_mapping(MediaType.ALBUM, album["id"], album["name"])
if "isExplicit" in track_obj:
track.metadata.explicit = track_obj["isExplicit"]
if "duration" in track_obj and str(track_obj["duration"]).isdigit():
async with self.mass.http_session.head(url) as response:
return response.status == 200
+ def _get_item_mapping(self, media_type: MediaType, key: str, name: str) -> ItemMapping:
+ return ItemMapping(
+ media_type,
+ key,
+ self.instance_id,
+ name,
+ create_uri(media_type, self.instance_id, key),
+ create_sort_name(self.name),
+ )
+
+ def _get_artist_item_mapping(self, artist_obj: dict) -> ItemMapping:
+ artist_id = artist_obj.get("id") or artist_obj.get("channelId")
+ if not artist_id and artist_obj["name"] == "Various Artists":
+ artist_id = VARIOUS_ARTISTS_YTM_ID
+ return self._get_item_mapping(MediaType.ARTIST, artist_id, artist_obj.get("name"))
+
@classmethod
async def _parse_thumbnails(cls, thumbnails_obj: dict) -> list[MediaItemImage]:
"""Parse and sort a list of thumbnails and return the highest quality."""