"""Check if uuid string is a valid UUID."""
try:
uuid_obj = UUID(uuid_to_test)
- except ValueError:
+ except (ValueError, TypeError):
return False
return str(uuid_obj) == uuid_to_test
class ExternalID(StrEnum):
"""Enum with External ID types."""
- # musicbrainz:
- # for tracks this is the RecordingID
- # for albums this is the ReleaseGroupID (NOT the release ID!)
- # for artists this is the ArtistID
- MUSICBRAINZ = "musicbrainz"
+ MB_ARTIST = "musicbrainz_artistid" # MusicBrainz Artist ID (or AlbumArtist ID)
+ MB_ALBUM = "musicbrainz_albumid" # MusicBrainz Album ID
+ MB_RELEASEGROUP = "musicbrainz_releasegroupid" # MusicBrainz ReleaseGroupID
+ MB_TRACK = "musicbrainz_trackid" # MusicBrainz Track ID
+ MB_RECORDING = "musicbrainz_recordingid" # MusicBrainz Recording ID
+
ISRC = "isrc" # used to identify unique recordings
BARCODE = "barcode" # EAN-13 barcode for identifying albums
ACOUSTID = "acoustid" # unique fingerprint (id) for a recording
"""Set default enum member if an unknown value is provided."""
return cls.UNKNOWN
+ @property
+ def is_unique(self) -> bool:
+ """Return if the ExternalID is unique."""
+ return self.is_musicbrainz or self in (
+ ExternalID.ACOUSTID,
+ ExternalID.DISCOGS,
+ ExternalID.TADB,
+ )
+
+ @property
+ def is_musicbrainz(self) -> bool:
+ """Return if the ExternalID is a MusicBrainz identifier."""
+ return self in (
+ ExternalID.MB_RELEASEGROUP,
+ ExternalID.MB_ALBUM,
+ ExternalID.MB_TRACK,
+ ExternalID.MB_ARTIST,
+ ExternalID.MB_RECORDING,
+ )
+
class LinkType(StrEnum):
"""Enum with link types."""
if self.sort_name is None:
self.sort_name = create_sort_name(self.name)
+ def get_external_id(self, external_id_type: ExternalID) -> str | None:
+ """Get (the first instance) of given External ID or None if not found."""
+ for ext_id in self.external_ids:
+ if ext_id[0] != external_id_type:
+ continue
+ return ext_id[1]
+ return None
+
+ def add_external_id(self, external_id_type: ExternalID, value: str) -> None:
+ """Add ExternalID."""
+ if external_id_type.is_musicbrainz and not is_valid_uuid(value):
+ msg = f"Invalid MusicBrainz identifier: {value}"
+ raise InvalidDataError(msg)
+ if external_id_type.is_unique and (
+ existing := next((x for x in self.external_ids if x[0] == external_id_type), None)
+ ):
+ self.external_ids.remove(existing)
+ self.external_ids.add((external_id_type, value))
+
@property
def mbid(self) -> str | None:
"""Return MusicBrainz ID."""
- return self.get_external_id(ExternalID.MUSICBRAINZ)
+ if self.media_type == MediaType.ARTIST:
+ return self.get_external_id(ExternalID.MB_ARTIST)
+ if self.media_type == MediaType.ALBUM:
+ return self.get_external_id(ExternalID.MB_ALBUM)
+ if self.media_type == MediaType.TRACK:
+ return self.get_external_id(ExternalID.MB_RECORDING)
+ return None
@mbid.setter
def mbid(self, value: str) -> None:
"""Set MusicBrainz External ID."""
- if not value:
+ if self.media_type == MediaType.ARTIST:
+ self.add_external_id(ExternalID.MB_ARTIST, value)
+ elif self.media_type == MediaType.ALBUM:
+ self.add_external_id(ExternalID.MB_ALBUM, value)
+ elif self.media_type == MediaType.TRACK:
+ # NOTE: for tracks we use the recording id to
+ # differentiate a unique recording
+ # and not the track id (as that is just the reference
+ # of the recording on a specific album)
+ self.add_external_id(ExternalID.MB_RECORDING, value)
return
- if not is_valid_uuid(value):
- msg = f"Invalid MusicBrainz identifier: {value}"
- raise InvalidDataError(msg)
- if existing := next((x for x in self.external_ids if x[0] == ExternalID.MUSICBRAINZ), None):
- # Musicbrainz ID is unique so remove existing entry
- self.external_ids.remove(existing)
- self.external_ids.add((ExternalID.MUSICBRAINZ, value))
-
- def get_external_id(self, external_id_type: ExternalID) -> str | None:
- """Get (the first instance) of given External ID or None if not found."""
- for ext_id in self.external_ids:
- if ext_id[0] != external_id_type:
- continue
- return ext_id[1]
- return None
def __hash__(self) -> int:
"""Return custom hash."""
LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.cache")
CONF_CLEAR_CACHE = "clear_cache"
-DB_SCHEMA_VERSION = 1
+DB_SCHEMA_VERSION = 3
class CacheController(CoreController):
CONF_SYNC_INTERVAL = "sync_interval"
CONF_DELETED_PROVIDERS = "deleted_providers"
CONF_ADD_LIBRARY_ON_PLAY = "add_library_on_play"
-DB_SCHEMA_VERSION: Final[int] = 2
+DB_SCHEMA_VERSION: Final[int] = 3
class MusicController(CoreController):
self.logger.info(
"Migrating database from version %s to %s", prev_version, DB_SCHEMA_VERSION
)
- if prev_version == 1:
- # migrate from version 1 to 2
+ if prev_version == 2:
+ # migrate from version 2 to 3
+ # convert musicbrainz external id's
await self.database.execute(
- f"DELETE FROM {DB_TABLE_PLAYLOG} WHERE provider = 'builtin'"
+ f"UPDATE {DB_TABLE_ARTISTS} SET external_ids = "
+ "replace(external_ids, 'musicbrainz', 'musicbrainz_artistid')"
+ )
+ # convert musicbrainz external id's
+ await self.database.execute(
+ f"UPDATE {DB_TABLE_ALBUMS} SET external_ids = "
+ "replace(external_ids, 'musicbrainz', 'musicbrainz_releasegroupid')"
+ )
+ await self.database.execute(
+ f"UPDATE {DB_TABLE_TRACKS} SET external_ids = "
+ "replace(external_ids, 'musicbrainz', 'musicbrainz_recordingid')"
)
await self.database.commit()
return
if compare_item_ids(base_item, compare_item):
return True
# return early on (un)matched external id
- for ext_id in (ExternalID.DISCOGS, ExternalID.MUSICBRAINZ, ExternalID.TADB):
+ for ext_id in (ExternalID.DISCOGS, ExternalID.MB_ARTIST, ExternalID.TADB):
external_id_match = compare_external_ids(
base_item.external_ids, compare_item.external_ids, ext_id
)
# return early on (un)matched external id
for ext_id in (
ExternalID.DISCOGS,
- ExternalID.MUSICBRAINZ,
+ ExternalID.MB_ALBUM,
ExternalID.TADB,
ExternalID.ASIN,
ExternalID.BARCODE,
return True
# return early on (un)matched external id
for ext_id in (
- ExternalID.MUSICBRAINZ,
+ ExternalID.MB_RECORDING,
ExternalID.DISCOGS,
ExternalID.ACOUSTID,
ExternalID.TADB,
if base_id[1:] in compare_ids:
return True
# return false if the identifier is unique (e.g. musicbrainz id)
- if external_id_type in (ExternalID.DISCOGS, ExternalID.MUSICBRAINZ, ExternalID.TADB):
+ if external_id_type.is_unique:
return False
return None
return self.tags.get("musicbrainzreleasegroupid")
@property
- def musicbrainz_releaseid(self) -> str | None:
- """Return musicbrainz_releaseid tag if present."""
+ def musicbrainz_albumid(self) -> str | None:
+ """Return musicbrainz_albumid tag if present."""
return self.tags.get("musicbrainzreleaseid", self.tags.get("musicbrainzalbumid"))
@property
import aiohttp.client_exceptions
from asyncio_throttle import Throttler
-from music_assistant.common.models.enums import ProviderFeature
+from music_assistant.common.models.enums import ExternalID, ProviderFeature
from music_assistant.common.models.media_items import ImageType, MediaItemImage, MediaItemMetadata
from music_assistant.server.controllers.cache import use_cache
from music_assistant.server.helpers.app_vars import app_var # pylint: disable=no-name-in-module
async def get_album_metadata(self, album: Album) -> MediaItemMetadata | None:
"""Retrieve metadata for album on fanart.tv."""
- if not album.mbid:
+ if (mbid := album.get_external_id(ExternalID.MB_RELEASEGROUP)) is None:
return None
self.logger.debug("Fetching metadata for Album %s on Fanart.tv", album.name)
- if data := await self._get_data(f"music/albums/{album.mbid}"):
+ if data := await self._get_data(f"music/albums/{mbid}"):
if data and data.get("albums"):
- data = data["albums"][album.mbid]
+ data = data["albums"][mbid]
metadata = MediaItemMetadata()
metadata.images = []
for key, img_type in IMG_MAPPING.items():
track.mbid = tags.musicbrainz_recordingid
track.metadata.chapters = UniqueList(tags.chapters)
if album:
- if not album.mbid:
- album.mbid = tags.musicbrainz_releasegroupid
+ if not album.mbid and tags.musicbrainz_albumid:
+ album.mbid = tags.musicbrainz_albumid
+ if tags.musicbrainz_releasegroupid:
+ album.add_external_id(ExternalID.MB_RELEASEGROUP, tags.musicbrainz_releasegroupid)
if not album.year:
album.year = tags.year
album.album_type = tags.album_type
album.name = info.get("title", info.get("name", name))
if sort_name := info.get("sortname"):
album.sort_name = sort_name
- if mbid := info.get("musicbrainzreleasegroupid"):
- album.mbid = mbid
+ if releasegroup_id := info.get("musicbrainzreleasegroupid"):
+ album.add_external_id(ExternalID.MB_RELEASEGROUP, releasegroup_id)
+ if album_id := info.get("musicbrainzalbumid"):
+ album.add_external_id(ExternalID.MB_ALBUM, album_id)
if mb_artist_id := info.get("musicbrainzalbumartistid"):
if album.artists and not album.artists[0].mbid:
album.artists[0].mbid = mb_artist_id
ITEM_KEY_OVERVIEW: Final = "Overview"
ITEM_KEY_MUSICBRAINZ_RELEASE_GROUP: Final = "MusicBrainzReleaseGroup"
ITEM_KEY_MUSICBRAINZ_ARTIST: Final = "MusicBrainzArtist"
+ITEM_KEY_MUSICBRAINZ_ALBUM: Final = "MusicBrainzAlbum"
ITEM_KEY_MUSICBRAINZ_TRACK: Final = "MusicBrainzTrack"
ITEM_KEY_SORT_NAME: Final = "SortName"
ITEM_KEY_ALBUM_ARTIST: Final = "AlbumArtist"
from aiojellyfin.const import ImageType as JellyImageType
-from music_assistant.common.models.enums import ContentType, ImageType, MediaType
+from music_assistant.common.models.enums import ContentType, ExternalID, ImageType, MediaType
from music_assistant.common.models.errors import InvalidDataError
from music_assistant.common.models.media_items import (
Album,
ITEM_KEY_IMAGE_TAGS,
ITEM_KEY_MEDIA_CODEC,
ITEM_KEY_MEDIA_STREAMS,
+ ITEM_KEY_MUSICBRAINZ_ALBUM,
ITEM_KEY_MUSICBRAINZ_ARTIST,
ITEM_KEY_MUSICBRAINZ_RELEASE_GROUP,
ITEM_KEY_MUSICBRAINZ_TRACK,
album.metadata.images = _get_artwork(instance_id, connection, jellyfin_album)
if ITEM_KEY_OVERVIEW in jellyfin_album:
album.metadata.description = jellyfin_album[ITEM_KEY_OVERVIEW]
+ if ITEM_KEY_MUSICBRAINZ_ALBUM in jellyfin_album[ITEM_KEY_PROVIDER_IDS]:
+ try:
+ album.add_external_id(
+ ExternalID.MB_ALBUM,
+ jellyfin_album[ITEM_KEY_PROVIDER_IDS][ITEM_KEY_MUSICBRAINZ_ALBUM],
+ )
+ except InvalidDataError as error:
+ logger.warning(
+ "Jellyfin has an invalid musicbrainz album id for album %s",
+ album.name,
+ exc_info=error if logger.isEnabledFor(logging.DEBUG) else None,
+ )
if ITEM_KEY_MUSICBRAINZ_RELEASE_GROUP in jellyfin_album[ITEM_KEY_PROVIDER_IDS]:
try:
- album.mbid = jellyfin_album[ITEM_KEY_PROVIDER_IDS][ITEM_KEY_MUSICBRAINZ_RELEASE_GROUP]
+ album.add_external_id(
+ ExternalID.MB_RELEASEGROUP,
+ jellyfin_album[ITEM_KEY_PROVIDER_IDS][ITEM_KEY_MUSICBRAINZ_RELEASE_GROUP],
+ )
except InvalidDataError as error:
logger.warning(
"Jellyfin has an invalid musicbrainz id for album %s",
from music_assistant.common.helpers.json import json_loads
from music_assistant.common.helpers.util import parse_title_and_version
-from music_assistant.common.models.enums import ProviderFeature
+from music_assistant.common.models.enums import ExternalID, ProviderFeature
from music_assistant.common.models.errors import (
InvalidDataError,
MediaNotFoundError,
msg = "Invalid MusicBrainz recording ID provided"
raise InvalidDataError(msg)
+ async def get_release_details(self, album_id: str) -> MusicBrainzRelease:
+ """Get Release/Album details by providing a MusicBrainz Album id."""
+ endpoint = f"release/{album_id}?inc=artist-credits+aliases+labels"
+ if result := await self.get_data(endpoint):
+ if "id" not in result:
+ result["id"] = album_id
+ try:
+ return MusicBrainzRelease.from_dict(replace_hyphens(result))
+ except MissingField as err:
+ raise InvalidDataError from err
+ msg = "Invalid MusicBrainz Album ID provided"
+ raise InvalidDataError(msg)
+
async def get_releasegroup_details(self, releasegroup_id: str) -> MusicBrainzReleaseGroup:
"""Get ReleaseGroup details by providing a MusicBrainz ReleaseGroup id."""
endpoint = f"release-group/{releasegroup_id}?inc=artists+aliases"
return MusicBrainzReleaseGroup.from_dict(replace_hyphens(result))
except MissingField as err:
raise InvalidDataError from err
- msg = "Invalid MusicBrainz ReleaseGroup ID or barcode provided"
+ msg = "Invalid MusicBrainz ReleaseGroup ID provided"
raise InvalidDataError(msg)
async def get_artist_details_by_album(
MusicBrainzArtist object that is returned does not contain the optional data.
"""
- if not ref_album.mbid:
- return None
result = None
- with suppress(InvalidDataError):
- result = await self.get_releasegroup_details(ref_album.mbid)
+ if mb_id := ref_album.get_external_id(ExternalID.MB_RELEASEGROUP):
+ with suppress(InvalidDataError):
+ result = await self.get_releasegroup_details(mb_id)
+ elif mb_id := ref_album.get_external_id(ExternalID.MB_ALBUM):
+ with suppress(InvalidDataError):
+ result = await self.get_release_details(mb_id)
+ else:
+ return None
if not (result and result.artist_credit):
return None
for strict in (True, False):
import aiohttp.client_exceptions
from asyncio_throttle import Throttler
-from music_assistant.common.models.enums import ProviderFeature
+from music_assistant.common.models.enums import ExternalID, ProviderFeature
from music_assistant.common.models.media_items import (
Album,
AlbumType,
async def get_album_metadata(self, album: Album) -> MediaItemMetadata | None:
"""Retrieve metadata for album on theaudiodb."""
- if not album.mbid:
- # for 100% accuracy we require the musicbrainz id for all lookups
+ if (mbid := album.get_external_id(ExternalID.MB_RELEASEGROUP)) is None:
return None
- result = await self._get_data("album-mb.php", i=album.mbid)
+ result = await self._get_data("album-mb.php", i=mbid)
if result and result.get("album"):
adb_album = result["album"][0]
# fill in some missing album info if needed
continue
if (
track.album
- and track.album.mbid
- and track.album.mbid != item["strMusicBrainzAlbumID"]
+ and (mb_rgid := track.album.get_external_id(ExternalID.MB_RELEASEGROUP))
+ # AudioDb swapped MB Album ID and ReleaseGroup ID ?!
+ and mb_rgid != item["strMusicBrainzAlbumID"]
):
continue
if not compare_strings(track_artist.name, item["strArtist"]):
]),
'external_ids': list([
list([
- 'musicbrainz',
+ 'musicbrainz_albumid',
+ 'bf25b030-0cbb-495a-8d79-6c7fee20a089',
+ ]),
+ list([
+ 'musicbrainz_releasegroupid',
'0193355a-cdfb-3936-afd2-44d651eb006d',
]),
]),
]),
'external_ids': list([
list([
- 'musicbrainz',
+ 'musicbrainz_albumid',
+ 'b13a174d-527d-44a1-b8f8-a4c78b03b7d9',
+ ]),
+ list([
+ 'musicbrainz_releasegroupid',
'f002d6b7-17af-4f9e-8d30-5486548ffe6f',
]),
]),
dict({
'external_ids': list([
list([
- 'musicbrainz',
+ 'musicbrainz_artistid',
'99164692-c02d-407c-81c9-25d338dd21f4',
]),
]),
'duration': 224,
'external_ids': list([
list([
- 'musicbrainz',
+ 'musicbrainz_recordingid',
'17d1019d-d4f4-326c-b4bb-d8aec2607bd7',
]),
]),
"""Test we can parse artists."""
async with aiofiles.open(example) as fp:
raw_data = ARTIST_DECODER.decode(await fp.read())
- parsed = parse_artist(_LOGGER, "xx-instance-id-xx", connection, raw_data)
- assert snapshot == parsed.to_dict()
+ parsed = parse_artist(_LOGGER, "xx-instance-id-xx", connection, raw_data).to_dict()
+ # sort external Ids to ensure they are always in the same order for snapshot testing
+ parsed["external_ids"].sort()
+ assert snapshot == parsed
@pytest.mark.parametrize("example", ALBUM_FIXTURES, ids=lambda val: str(val.stem))
"""Test we can parse albums."""
async with aiofiles.open(example) as fp:
raw_data = ARTIST_DECODER.decode(await fp.read())
- parsed = parse_album(_LOGGER, "xx-instance-id-xx", connection, raw_data)
- assert snapshot == parsed.to_dict()
+ parsed = parse_album(_LOGGER, "xx-instance-id-xx", connection, raw_data).to_dict()
+ # sort external Ids to ensure they are always in the same order for snapshot testing
+ parsed["external_ids"].sort()
+ assert snapshot == parsed
@pytest.mark.parametrize("example", TRACK_FIXTURES, ids=lambda val: str(val.stem))
"""Test we can parse tracks."""
async with aiofiles.open(example) as fp:
raw_data = ARTIST_DECODER.decode(await fp.read())
- parsed = parse_track(_LOGGER, "xx-instance-id-xx", connection, raw_data)
- assert snapshot == parsed.to_dict()
+ parsed = parse_track(_LOGGER, "xx-instance-id-xx", connection, raw_data).to_dict()
+ # sort external Ids to ensure they are always in the same order for snapshot testing
+ parsed["external_ids"]
+ assert snapshot == parsed
artist_b.name = "Artist B"
artist_b.item_id = "2"
artist_b.provider = "test2"
- artist_a.external_ids = {(media_items.ExternalID.MUSICBRAINZ, "123")}
+ artist_a.external_ids = {(media_items.ExternalID.MB_ARTIST, "123")}
artist_b.external_ids = artist_a.external_ids
assert compare.compare_artist(artist_a, artist_b) is True
# test on external id mismatch
artist_b.name = artist_a.name
- artist_b.external_ids = {(media_items.ExternalID.MUSICBRAINZ, "1234")}
+ artist_b.external_ids = {(media_items.ExternalID.MB_ARTIST, "1234")}
assert compare.compare_artist(artist_a, artist_b) is False
album_b.name = "Album B"
album_b.item_id = "2"
album_b.provider = "test2"
- album_a.external_ids = {(media_items.ExternalID.MUSICBRAINZ, "123")}
+ album_a.external_ids = {(media_items.ExternalID.MB_ALBUM, "123")}
album_b.external_ids = album_a.external_ids
assert compare.compare_album(album_a, album_b) is True
# test on external id mismatch
album_b.name = album_a.name
- album_b.external_ids = {(media_items.ExternalID.MUSICBRAINZ, "1234")}
+ album_b.external_ids = {(media_items.ExternalID.MB_ALBUM, "1234")}
assert compare.compare_album(album_a, album_b) is False
album_a.external_ids = set()
album_b.external_ids = set()
track_b.name = "Track B"
track_b.item_id = "2"
track_b.provider = "test2"
- track_a.external_ids = {(media_items.ExternalID.MUSICBRAINZ, "123")}
+ track_a.external_ids = {(media_items.ExternalID.MB_RECORDING, "123")}
track_b.external_ids = track_a.external_ids
assert compare.compare_track(track_a, track_b) is True
# test on external id mismatch
track_b.name = track_a.name
- track_b.external_ids = {(media_items.ExternalID.MUSICBRAINZ, "1234")}
+ track_b.external_ids = {(media_items.ExternalID.MB_RECORDING, "1234")}
assert compare.compare_track(track_a, track_b) is False
track_a.external_ids = set()
track_b.external_ids = set()
# this can happen for some classical music albums
track_a.external_ids = {
(media_items.ExternalID.ISRC, "123"),
- (media_items.ExternalID.MUSICBRAINZ, "abc"),
+ (media_items.ExternalID.MB_RECORDING, "abc"),
}
track_b.external_ids = {
(media_items.ExternalID.ISRC, "123"),
- (media_items.ExternalID.MUSICBRAINZ, "abcd"),
+ (media_items.ExternalID.MB_RECORDING, "abcd"),
}
assert compare.compare_track(track_a, track_b) is False
projects / music-assistant-server.git / commitdiff