{
- // Use IntelliSense to learn about possible attributes.
- // Hover to view descriptions of existing attributes.
- // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
- "version": "0.2.0",
- "configurations": [
- {
- "name": "Python: Module",
- "type": "debugpy",
- "request": "launch",
- "module": "music_assistant",
- "justMyCode": false,
- "args":[
- "--log-level", "debug"
- ],
- "env": {"PYTHONDEVMODE": "1"}
- }
- ]
+ // Use IntelliSense to learn about possible attributes.
+ // Hover to view descriptions of existing attributes.
+ // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+ "version": "0.2.0",
+ "configurations": [
+ {
+ "name": "Music Assistant: Server",
+ "type": "debugpy",
+ "request": "launch",
+ "module": "music_assistant",
+ "justMyCode": false,
+ "args": ["--log-level", "debug"],
+ "env": { "PYTHONDEVMODE": "1" }
+ },
+ {
+ "name": "Music Assistant: Tests",
+ "type": "debugpy",
+ "request": "launch",
+ "module": "pytest",
+ "justMyCode": false,
+ "args": ["tests"]
+ },
+ {
+ "name": "Python Debugger: Current File",
+ "type": "debugpy",
+ "request": "launch",
+ "program": "${file}",
+ "console": "integratedTerminal"
+ }
+ ]
}
performers: set[str] | None = None
preview: str | None = None
popularity: int | None = None
- # cache_checksum: optional value to (in)validate cache / detect changes (used for playlists)
- cache_checksum: str | None = None
# last_refresh: timestamp the (full) metadata was last collected
last_refresh: int | None = None
owner: str = ""
is_editable: bool = False
+ # cache_checksum: optional value to (in)validate cache
+ # detect changes to the playlist tracks listing
+ cache_checksum: str | None = None
+
@dataclass(kw_only=True)
class Radio(MediaItem):
API_SCHEMA_VERSION: Final[int] = 24
MIN_SCHEMA_VERSION: Final[int] = 24
-DB_SCHEMA_VERSION: Final[int] = 2
+
MASS_LOGGER_NAME: Final[str] = "music_assistant"
from music_assistant.common.helpers.json import json_dumps, json_loads
from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import ConfigEntryType
-from music_assistant.constants import (
- DB_SCHEMA_VERSION,
- DB_TABLE_CACHE,
- DB_TABLE_SETTINGS,
- MASS_LOGGER_NAME,
-)
+from music_assistant.constants import DB_TABLE_CACHE, DB_TABLE_SETTINGS, MASS_LOGGER_NAME
from music_assistant.server.helpers.database import DatabaseConnection
from music_assistant.server.models.core_controller import CoreController
LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.cache")
CONF_CLEAR_CACHE = "clear_cache"
+DB_SCHEMA_VERSION = 1
class CacheController(CoreController):
from music_assistant.server.helpers.compare import (
compare_album,
compare_artists,
+ compare_media_item,
loose_compare_strings,
)
prov_track = await self.mass.music.tracks.get_provider_item(
track_prov_map.item_id, track_prov_map.provider_instance, force_refresh=True
)
- if (
- prov_track.metadata.cache_checksum
- == prov_album_track.metadata.cache_checksum
- ):
- continue
await self.mass.music.tracks._update_library_item(
prov_album_track.item_id, prov_track, True
)
for search_result_item in search_result:
if not search_result_item.available:
continue
- if not compare_album(db_album, search_result_item):
+ if not compare_media_item(db_album, search_result_item):
continue
- # we must fetch the full album version, search results are simplified objects
+ # we must fetch the full album version, search results can be simplified objects
prov_album = await self.get_provider_item(
search_result_item.item_id,
search_result_item.provider,
lazy=not force_refresh,
)
prov_map = next(x for x in playlist.provider_mappings)
- cache_checksum = playlist.metadata.cache_checksum
+ cache_checksum = playlist.cache_checksum
tracks = await self._get_provider_playlist_tracks(
prov_map.item_id,
prov_map.provider_instance,
)
from music_assistant.server.helpers.compare import (
compare_artists,
+ compare_media_item,
compare_track,
loose_compare_strings,
)
if not search_result_item.available:
continue
# do a basic compare first
- if not compare_track(db_track, search_result_item, strict=False):
+ if not compare_media_item(db_track, search_result_item, strict=False):
continue
- # we must fetch the full version, search results are simplified objects
+ # we must fetch the full version, search results can be simplified objects
prov_track = await self.get_provider_item(
search_result_item.item_id,
search_result_item.provider,
from contextlib import suppress
from itertools import zip_longest
from math import inf
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Final
from music_assistant.common.helpers.datetime import utc_timestamp
from music_assistant.common.helpers.global_cache import get_global_cache_value
from music_assistant.common.models.provider import SyncTask
from music_assistant.common.models.streamdetails import LoudnessMeasurement
from music_assistant.constants import (
- DB_SCHEMA_VERSION,
DB_TABLE_ALBUM_ARTISTS,
DB_TABLE_ALBUM_TRACKS,
DB_TABLE_ALBUMS,
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
class MusicController(CoreController):
)
IGNORE_VERSIONS = (
- "remaster",
- "explicit",
+ "explicit", # explicit is matched separately
"music from and inspired by the motion picture",
"original soundtrack",
- "hi-res",
+ "hi-res", # quality is handled separately
)
# for strict matching we REQUIRE both items to be a real album object
assert isinstance(base_item, Album)
assert isinstance(compare_item, Album)
+ # compare year
+ if base_item.year and compare_item.year and base_item.year != compare_item.year:
+ return False
# compare explicitness
if compare_explicit(base_item.metadata, compare_item.metadata) is False:
return False
- # compare album artist
- return compare_artists(base_item.artists, compare_item.artists, True)
+ # compare album artist(s)
+ return compare_artists(base_item.artists, compare_item.artists, not strict)
def compare_track(
- base_item: Track | ItemMapping,
- compare_item: Track | ItemMapping,
+ base_item: Track,
+ compare_item: Track,
strict: bool = True,
- track_albums: list[Album | ItemMapping] | None = None,
+ track_albums: list[Album] | None = None,
) -> bool:
"""Compare two track items and return True if they match."""
if base_item is None or compare_item is None:
)
if external_id_match is not None:
return external_id_match
+ # return early on exact albumtrack match = 100% match
+ if (
+ base_item.album
+ and compare_item.album
+ and compare_album(base_item.album, compare_item.album, False)
+ and base_item.disc_number
+ and compare_item.disc_number
+ and base_item.track_number
+ and compare_item.track_number
+ and base_item.disc_number == compare_item.disc_number
+ and base_item.track_number == compare_item.track_number
+ ):
+ return True
+
## fallback to comparing on attributes
+
# compare name
if not compare_strings(base_item.name, compare_item.name, strict=True):
return False
compare_item.metadata.explicit = compare_item.album.metadata.explicit
if strict and compare_explicit(base_item.metadata, compare_item.metadata) is False:
return False
- if not strict and not (base_item.album or track_albums):
- # in non-strict mode, the album does not have to match (but duration needs to)
- return abs(base_item.duration - compare_item.duration) <= 2
- # exact albumtrack match = 100% match
- if (
- base_item.album
- and compare_item.album
- and compare_album(base_item.album, compare_item.album, False)
- and base_item.disc_number == compare_item.disc_number
- and base_item.track_number == compare_item.track_number
- ):
- return True
+
# fallback: exact album match and (near-exact) track duration match
if (
base_item.album is not None
and compare_item.album is not None
+ and (base_item.track_number == 0 or compare_item.track_number == 0)
and compare_album(base_item.album, compare_item.album, False)
and abs(base_item.duration - compare_item.duration) <= 3
):
return True
+
# fallback: additional compare albums provided for base track
if (
compare_item.album is not None
for track_album in track_albums:
if compare_album(track_album, compare_item.album, False):
return True
- # accept last resort: albumless track and (near) exact duration
- # otherwise fail all other cases
- return (
+
+ # fallback edge case: albumless track with same duration
+ if (
base_item.album is None
and compare_item.album is None
- and abs(base_item.duration - compare_item.duration) <= 1
- )
+ and base_item.disc_number == 0
+ and compare_item.disc_number == 0
+ and base_item.track_number == 0
+ and compare_item.track_number == 0
+ and base_item.duration == compare_item.duration
+ ):
+ return True
+
+ if strict:
+ # in strict mode, we require an exact album match so return False here
+ return False
+
+ # Accept last resort (in non strict mode): (near) exact duration,
+ # otherwise fail all other cases.
+ # Note that as this stage, all other info already matches,
+ # such as title artist etc.
+ return abs(base_item.duration - compare_item.duration) <= 2
def compare_playlist(
any_match: bool = True,
) -> bool:
"""Compare two lists of artist and return True if both lists match (exactly)."""
+ if not base_items and not compare_items:
+ return True
+ if not base_items or not compare_items:
+ return False
+ # match if first artist matches in both lists
+ if compare_artist(base_items[0], compare_items[0]):
+ return True
+ # compare the artist lists
matches = 0
for base_item in base_items:
for compare_item in compare_items:
if any_match:
return True
matches += 1
- return len(base_items) == matches
+ return len(base_items) == len(compare_items) == matches
def compare_albums(
if create_safe_string(str1) == create_safe_string(str2):
return True
# last resort: use difflib to compare strings
- required_accuracy = 0.91 if len(str1) > 8 else 0.85
+ required_accuracy = 0.9 if (len(str1) + len(str2)) > 18 else 0.8
return SequenceMatcher(a=str1_lower, b=str2).ratio() > required_accuracy
return False
if base_version and not compare_version:
return False
- if " " not in base_version:
- return compare_strings(base_version, compare_version)
+
+ if " " not in base_version and " " not in compare_version:
+ return compare_strings(base_version, compare_version, False)
+
# do this the hard way as sometimes the version string is in the wrong order
- base_versions = base_version.lower().split(" ").sort()
- compare_versions = compare_version.lower().split(" ").sort()
+ base_versions = sorted(base_version.lower().split(" "))
+ compare_versions = sorted(compare_version.lower().split(" "))
+ # filter out words we can ignore (such as 'version')
+ ignore_words = [*IGNORE_VERSIONS, "version", "edition", "variant", "versie", "versione"]
+ base_versions = [x for x in base_versions if x not in ignore_words]
+ compare_versions = [x for x in compare_versions if x not in ignore_words]
+
return base_versions == compare_versions
library_item = await controller.add_item_to_library(
prov_item, metadata_lookup=False
)
- elif library_item.metadata.cache_checksum != prov_item.metadata.cache_checksum:
- # existing dbitem checksum changed
+ elif getattr(library_item, "cache_checksum", None) != getattr(
+ prov_item, "cache_checksum", None
+ ):
+ # existing dbitem checksum changed (playlists only)
library_item = await controller.update_item_in_library(
library_item.item_id, prov_item
)
playlist.metadata.description = description.get("standard")
playlist.is_editable = attributes.get("canEdit", False)
if checksum := attributes.get("lastModifiedDate"):
- playlist.metadata.cache_checksum = checksum
+ playlist.cache_checksum = checksum
return playlist
async def _get_all_items(self, endpoint, key="data", **kwargs) -> list[dict]:
UniqueList,
)
from music_assistant.common.models.streamdetails import StreamDetails
-from music_assistant.constants import DB_SCHEMA_VERSION, MASS_LOGO, VARIOUS_ARTISTS_FANART
+from music_assistant.constants import MASS_LOGO, VARIOUS_ARTISTS_FANART
from music_assistant.server.helpers.tags import AudioTags, parse_tags
from music_assistant.server.models.music_provider import MusicProvider
},
owner="Music Assistant",
is_editable=False,
+ cache_checksum=str(int(time.time())),
metadata=MediaItemMetadata(
images=UniqueList([DEFAULT_THUMB])
if prov_playlist_id in COLLAGE_IMAGE_PLAYLISTS
else UniqueList([DEFAULT_THUMB, DEFAULT_FANART]),
- cache_checksum=str(int(time.time())),
),
)
# user created universal playlist
owner="Music Assistant",
is_editable=True,
)
- playlist.metadata.cache_checksum = f"{DB_SCHEMA_VERSION}.{stored_item.get('last_updated')}"
+ playlist.cache_checksum = str(stored_item.get("last_updated"))
if image_url := stored_item.get("image_url"):
playlist.metadata.images = UniqueList(
[
remotely_accessible=True,
)
],
- cache_checksum=playlist.checksum,
),
is_editable=creator.id == self.user.id,
owner=creator.name,
+ cache_checksum=playlist.checksum,
)
def get_playlist_creator(self, playlist: deezer.Playlist):
DB_TABLE_TRACK_ARTISTS,
VARIOUS_ARTISTS_NAME,
)
-from music_assistant.server.controllers.music import DB_SCHEMA_VERSION
from music_assistant.server.helpers.compare import compare_strings
from music_assistant.server.helpers.playlists import parse_m3u, parse_pls
from music_assistant.server.helpers.tags import parse_tags, split_items
cur_filenames.add(item.path)
try:
# continue if the item did not change (checksum still the same)
- if item.checksum == file_checksums.get(item.path):
+ prev_checksum = file_checksums.get(item.path)
+ if item.checksum == prev_checksum:
continue
self.logger.debug("Processing: %s", item.path)
if item.ext in TRACK_EXTENSIONS:
# when they are detected as changed
track = await self._parse_track(item)
await self.mass.music.tracks.add_item_to_library(
- track, metadata_lookup=False, overwrite_existing=True
+ track, metadata_lookup=False, overwrite_existing=prev_checksum is not None
)
elif item.ext in PLAYLIST_EXTENSIONS:
playlist = await self.get_playlist(item.path)
# add/update] playlist to db
- playlist.metadata.cache_checksum = item.checksum
+ playlist.cache_checksum = item.checksum
# playlist is always favorite
playlist.favorite = True
await self.mass.music.playlists.add_item_to_library(
- playlist, metadata_lookup=False, overwrite_existing=True
+ playlist,
+ metadata_lookup=False,
+ overwrite_existing=prev_checksum is not None,
)
except Exception as err: # pylint: disable=broad-except
# we don't want the whole sync to crash on one file so we catch all exceptions here
if file_item.ext == "pls":
playlist.is_editable = False
playlist.owner = self.name
- checksum = f"{DB_SCHEMA_VERSION}.{file_item.checksum}"
- playlist.metadata.cache_checksum = checksum
+ checksum = str(file_item.checksum)
+ playlist.cache_checksum = checksum
return playlist
async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
if acoustid := tags.get("acoustidid"):
track.external_ids.add((ExternalID.ACOUSTID, acoustid))
- album: Album | None
+ album: Album | None = None
album_artists: list[Artist] = []
# album
# track artist(s)
for index, track_artist_str in enumerate(tags.artists):
# reuse album artist details if possible
- if album and (
- album_artist := next((x for x in album_artists if x.name == track_artist_str), None)
- ):
+ if album_artist := next((x for x in album_artists if x.name == track_artist_str), None):
artist = album_artist
else:
artist = await self._parse_artist(track_artist_str)
album.year = tags.year
album.album_type = tags.album_type
album.metadata.explicit = track.metadata.explicit
- # set checksum to invalidate any cached listings
- track.metadata.cache_checksum = file_item.checksum
- if album:
- # use track checksum for album(artists) too
- album.metadata.cache_checksum = track.metadata.cache_checksum
- for artist in album_artists:
- artist.metadata.cache_checksum = track.metadata.cache_checksum
-
return track
async def _parse_artist(
from plexapi.myplex import MyPlexAccount, MyPlexPinLogin
from plexapi.server import PlexServer
+from music_assistant.common.helpers.util import parse_title_and_version
from music_assistant.common.models.config_entries import (
ConfigEntry,
ConfigValueOption,
),
ConfigEntry(
key=CONF_LOCAL_SERVER_PORT,
- type=ConfigEntryType.STRING,
+ type=ConfigEntryType.INTEGER,
label="Local server port",
description="The local server port (e.g. 32400)",
required=True,
return await self._run_async(self._plex_library.fetchItem, key, cls)
def _get_item_mapping(self, media_type: MediaType, key: str, name: str) -> ItemMapping:
+ name, version = parse_title_and_version(name)
+ if media_type in (MediaType.ALBUM, MediaType.TRACK):
+ name, version = parse_title_and_version(name)
+ else:
+ version = ""
return ItemMapping(
media_type=media_type,
item_id=key,
provider=self.instance_id,
name=name,
+ version=version,
)
async def _get_or_create_artist_by_name(self, artist_name) -> Artist:
)
]
playlist.is_editable = not plex_playlist.smart
- playlist.metadata.cache_checksum = str(plex_playlist.updatedAt.timestamp())
+ playlist.cache_checksum = str(plex_playlist.updatedAt.timestamp())
return playlist
]
if plex_track.parentKey:
track.album = self._get_item_mapping(
- MediaType.ALBUM, plex_track.parentKey, plex_track.parentKey
+ MediaType.ALBUM, plex_track.parentKey, plex_track.parentTitle
)
if plex_track.duration:
track.duration = int(plex_track.duration / 1000)
remotely_accessible=True,
)
]
- playlist.metadata.cache_checksum = str(playlist_obj["updated_at"])
+ playlist.cache_checksum = str(playlist_obj["updated_at"])
return playlist
async def _auth_token(self):
)
]
- liked_songs.metadata.cache_checksum = str(time.time())
+ liked_songs.cache_checksum = str(time.time())
return liked_songs
]
if playlist.owner is None:
playlist.owner = self._sp_user["display_name"]
- playlist.metadata.cache_checksum = str(playlist_obj["snapshot_id"])
+ playlist.cache_checksum = str(playlist_obj["snapshot_id"])
return playlist
async def login(self) -> dict:
is_editable = bool(creator_id and str(creator_id) == self._tidal_user_id)
playlist.is_editable = is_editable
# metadata
- playlist.metadata.cache_checksum = str(playlist_obj.last_updated)
+ playlist.cache_checksum = str(playlist_obj.last_updated)
playlist.metadata.popularity = playlist_obj.popularity
if picture := (playlist_obj.square_picture or playlist_obj.picture):
picture_id = picture.replace("-", "/")
playlist.owner = authors["name"]
else:
playlist.owner = self.instance_id
- playlist.metadata.cache_checksum = playlist_obj.get("checksum")
+ playlist.cache_checksum = playlist_obj.get("checksum")
return playlist
def _parse_track(self, track_obj: dict) -> Track:
'item_id': '70b7288088b42d318f75dbcc41fd0091',
'media_type': 'album',
'metadata': dict({
- 'cache_checksum': None,
'chapters': None,
'copyright': None,
'description': None,
'item_id': '32ed6a0091733dcff57eae67010f3d4b',
'media_type': 'album',
'metadata': dict({
- 'cache_checksum': None,
'chapters': None,
'copyright': None,
'description': None,
'item_id': '7c8d0bd55291c7fc0451d17ebef30017',
'media_type': 'album',
'metadata': dict({
- 'cache_checksum': None,
'chapters': None,
'copyright': None,
'description': None,
'item_id': 'dd954bbf54398e247d803186d3585b79',
'media_type': 'artist',
'metadata': dict({
- 'cache_checksum': None,
'chapters': None,
'copyright': None,
'description': None,
'item_id': 'b5319fb11cde39fca2023184fcfa9862',
'media_type': 'track',
'metadata': dict({
- 'cache_checksum': None,
'chapters': None,
'copyright': None,
'description': None,
'item_id': '54918f75ee8f6c8b8dc5efd680644f29',
'media_type': 'track',
'metadata': dict({
- 'cache_checksum': None,
'chapters': None,
'copyright': None,
'description': None,
'item_id': 'fb12a77f49616a9fc56a6fab2b94174c',
'media_type': 'track',
'metadata': dict({
- 'cache_checksum': None,
'chapters': None,
'copyright': None,
'description': None,
--- /dev/null
+"""Tests for mediaitem compare helper functions."""
+
+from music_assistant.common.models import media_items
+from music_assistant.server.helpers import compare
+
+
+def test_compare_version() -> None:
+ """Test the version compare helper."""
+ assert compare.compare_version("Remaster", "remaster") is True
+ assert compare.compare_version("Remastered", "remaster") is True
+ assert compare.compare_version("Remaster", "") is False
+ assert compare.compare_version("Remaster", "Remix") is False
+ assert compare.compare_version("", "Deluxe") is False
+ assert compare.compare_version("", "Live") is False
+ assert compare.compare_version("Live", "live") is True
+ assert compare.compare_version("Live", "live version") is True
+ assert compare.compare_version("Live version", "live") is True
+ assert compare.compare_version("Deluxe Edition", "Deluxe") is True
+ assert compare.compare_version("Deluxe Karaoke Edition", "Deluxe") is False
+ assert compare.compare_version("Deluxe Karaoke Edition", "Karaoke") is False
+ assert compare.compare_version("Deluxe Edition", "Edition Deluxe") is True
+ assert compare.compare_version("", "Karaoke Version") is False
+ assert compare.compare_version("Karaoke", "Karaoke Version") is True
+ assert compare.compare_version("Remaster", "Remaster Edition Deluxe") is False
+ assert compare.compare_version("Remastered Version", "Deluxe Version") is False
+
+
+def test_compare_artist() -> None:
+ """Test artist comparison."""
+ artist_a = media_items.Artist(
+ item_id="1",
+ provider="test1",
+ name="Artist A",
+ provider_mappings={
+ media_items.ProviderMapping(
+ item_id="1", provider_domain="test", provider_instance="test1"
+ )
+ },
+ )
+ artist_b = media_items.Artist(
+ item_id="1",
+ provider="test2",
+ name="Artist A",
+ provider_mappings={
+ media_items.ProviderMapping(
+ item_id="2", provider_domain="test", provider_instance="test2"
+ )
+ },
+ )
+ # test match on name match
+ assert compare.compare_artist(artist_a, artist_b) is True
+ # test match on name mismatch
+ artist_b.name = "Artist B"
+ assert compare.compare_artist(artist_a, artist_b) is False
+ # test on exact item_id match
+ artist_b.item_id = artist_a.item_id
+ artist_b.provider = artist_a.provider
+ assert compare.compare_artist(artist_a, artist_b) is True
+ # test on external id match
+ artist_b.name = "Artist B"
+ artist_b.item_id = "2"
+ artist_b.provider = "test2"
+ artist_a.external_ids = {(media_items.ExternalID.MUSICBRAINZ, "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")}
+ assert compare.compare_artist(artist_a, artist_b) is False
+
+
+def test_compare_album() -> None:
+ """Test album comparison."""
+ album_a = media_items.Album(
+ item_id="1",
+ provider="test1",
+ name="Album A",
+ provider_mappings={
+ media_items.ProviderMapping(
+ item_id="1", provider_domain="test", provider_instance="test1"
+ )
+ },
+ )
+ album_b = media_items.Album(
+ item_id="1",
+ provider="test2",
+ name="Album A",
+ provider_mappings={
+ media_items.ProviderMapping(
+ item_id="2", provider_domain="test", provider_instance="test2"
+ )
+ },
+ )
+ # test match on name match
+ assert compare.compare_album(album_a, album_b) is True
+ # test match on name mismatch
+ album_b.name = "Album B"
+ assert compare.compare_album(album_a, album_b) is False
+ # test on version mismatch
+ album_b.name = album_a.name
+ album_b.version = "Deluxe"
+ assert compare.compare_album(album_a, album_b) is False
+ album_b.version = "Remix"
+ assert compare.compare_album(album_a, album_b) is False
+ # test on version match
+ album_b.name = album_a.name
+ album_a.version = "Deluxe"
+ album_b.version = "Deluxe Edition"
+ assert compare.compare_album(album_a, album_b) is True
+ # test on exact item_id match
+ album_b.item_id = album_a.item_id
+ album_b.provider = album_a.provider
+ assert compare.compare_album(album_a, album_b) is True
+ # test on external id match
+ album_b.name = "Album B"
+ album_b.item_id = "2"
+ album_b.provider = "test2"
+ album_a.external_ids = {(media_items.ExternalID.MUSICBRAINZ, "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")}
+ assert compare.compare_album(album_a, album_b) is False
+ album_a.external_ids = set()
+ album_b.external_ids = set()
+ # fail on year mismatch
+ album_b.external_ids = set()
+ album_a.year = 2021
+ album_b.year = 2020
+ assert compare.compare_album(album_a, album_b) is False
+ # pass on year match
+ album_b.year = 2021
+ assert compare.compare_album(album_a, album_b) is True
+ # fail on artist mismatch
+ album_a.artists = media_items.UniqueList(
+ [media_items.ItemMapping(item_id="1", provider="test1", name="Artist A")]
+ )
+ album_b.artists = media_items.UniqueList(
+ [media_items.ItemMapping(item_id="2", provider="test1", name="Artist B")]
+ )
+ assert compare.compare_album(album_a, album_b) is False
+ # pass on partial artist match (if first artist matches)
+ album_a.artists = media_items.UniqueList(
+ [media_items.ItemMapping(item_id="1", provider="test1", name="Artist A")]
+ )
+ album_b.artists = media_items.UniqueList(
+ [
+ media_items.ItemMapping(item_id="1", provider="test1", name="Artist A"),
+ media_items.ItemMapping(item_id="2", provider="test1", name="Artist B"),
+ ]
+ )
+ assert compare.compare_album(album_a, album_b) is True
+ # fail on partial artist match in strict mode
+ album_b.artists = media_items.UniqueList(
+ [
+ media_items.ItemMapping(item_id="2", provider="test1", name="Artist B"),
+ media_items.ItemMapping(item_id="1", provider="test1", name="Artist A"),
+ ]
+ )
+ assert compare.compare_album(album_a, album_b) is False
+ # partial artist match is allowed in non-strict mode
+ assert compare.compare_album(album_a, album_b, False) is True
+
+
+def test_compare_track() -> None: # noqa: PLR0915
+ """Test track comparison."""
+ track_a = media_items.Track(
+ item_id="1",
+ provider="test1",
+ name="Track A",
+ provider_mappings={
+ media_items.ProviderMapping(
+ item_id="1", provider_domain="test", provider_instance="test1"
+ )
+ },
+ )
+ track_b = media_items.Track(
+ item_id="1",
+ provider="test2",
+ name="Track A",
+ provider_mappings={
+ media_items.ProviderMapping(
+ item_id="2", provider_domain="test", provider_instance="test2"
+ )
+ },
+ )
+ # test match on name match
+ assert compare.compare_track(track_a, track_b) is True
+ # test match on name mismatch
+ track_b.name = "Track B"
+ assert compare.compare_track(track_a, track_b) is False
+ # test on version mismatch
+ track_b.name = track_a.name
+ track_b.version = "Deluxe"
+ assert compare.compare_track(track_a, track_b) is False
+ track_b.version = "Remix"
+ assert compare.compare_track(track_a, track_b) is False
+ # test on version mismatch
+ track_b.name = track_a.name
+ track_a.version = ""
+ track_b.version = "Remaster"
+ assert compare.compare_track(track_a, track_b) is False
+ track_b.version = "Remix"
+ assert compare.compare_track(track_a, track_b) is False
+ # test on version match
+ track_b.name = track_a.name
+ track_a.version = "Deluxe"
+ track_b.version = "Deluxe Edition"
+ assert compare.compare_track(track_a, track_b) is True
+ # test on exact item_id match
+ track_b.item_id = track_a.item_id
+ track_b.provider = track_a.provider
+ assert compare.compare_track(track_a, track_b) is True
+ # test on external id match
+ track_b.name = "Track B"
+ track_b.item_id = "2"
+ track_b.provider = "test2"
+ track_a.external_ids = {(media_items.ExternalID.MUSICBRAINZ, "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")}
+ assert compare.compare_track(track_a, track_b) is False
+ track_a.external_ids = set()
+ track_b.external_ids = set()
+ # fail on artist mismatch
+ track_a.artists = media_items.UniqueList(
+ [media_items.ItemMapping(item_id="1", provider="test1", name="Artist A")]
+ )
+ track_b.artists = media_items.UniqueList(
+ [media_items.ItemMapping(item_id="2", provider="test1", name="Artist B")]
+ )
+ assert compare.compare_track(track_a, track_b) is False
+ # pass on partial artist match (if first artist matches)
+ track_a.artists = media_items.UniqueList(
+ [media_items.ItemMapping(item_id="1", provider="test1", name="Artist A")]
+ )
+ track_b.artists = media_items.UniqueList(
+ [
+ media_items.ItemMapping(item_id="1", provider="test1", name="Artist A"),
+ media_items.ItemMapping(item_id="2", provider="test1", name="Artist B"),
+ ]
+ )
+ assert compare.compare_track(track_a, track_b) is True
+ # fail on partial artist match in strict mode
+ track_b.artists = media_items.UniqueList(
+ [
+ media_items.ItemMapping(item_id="2", provider="test1", name="Artist B"),
+ media_items.ItemMapping(item_id="1", provider="test1", name="Artist A"),
+ ]
+ )
+ assert compare.compare_track(track_a, track_b) is False
+ # partial artist match is allowed in non-strict mode
+ assert compare.compare_track(track_a, track_b, False) is True
+ track_b.artists = track_a.artists
+ # fail on album mismatch
+ track_a.album = media_items.ItemMapping(item_id="1", provider="test1", name="Album A")
+ track_b.album = media_items.ItemMapping(item_id="2", provider="test1", name="Album B")
+ assert compare.compare_track(track_a, track_b) is False
+ # pass on exact album(track) match (regardless duration)
+ track_b.album = track_a.album
+ track_a.disc_number = 1
+ track_a.track_number = 1
+ track_b.disc_number = track_a.disc_number
+ track_b.track_number = track_a.track_number
+ track_a.duration = 300
+ track_b.duration = 310
+ assert compare.compare_track(track_a, track_b) is True
+ # pass on album(track) mismatch
+ track_b.album = track_a.album
+ track_a.disc_number = 1
+ track_a.track_number = 1
+ track_b.disc_number = track_a.disc_number
+ track_b.track_number = 2
+ track_b.duration = track_a.duration
+ assert compare.compare_track(track_a, track_b) is False
+ # test special case - ISRC match but MusicBrainz ID mismatch
+ # this can happen for some classical music albums
+ track_a.external_ids = {
+ (media_items.ExternalID.ISRC, "123"),
+ (media_items.ExternalID.MUSICBRAINZ, "abc"),
+ }
+ track_b.external_ids = {
+ (media_items.ExternalID.ISRC, "123"),
+ (media_items.ExternalID.MUSICBRAINZ, "abcd"),
+ }
+ assert compare.compare_track(track_a, track_b) is False
def test_version_extract() -> None:
"""Test the extraction of version from title."""
+ test_str = "Bam Bam (feat. Ed Sheeran)"
+ title, version = util.parse_title_and_version(test_str)
+ assert title == "Bam Bam"
+ assert version == ""
test_str = "Bam Bam (feat. Ed Sheeran) - Karaoke Version"
title, version = util.parse_title_and_version(test_str)
assert title == "Bam Bam"
assert version == "Karaoke Version"
+ test_str = "SuperSong (2011 Remaster)"
+ title, version = util.parse_title_and_version(test_str)
+ assert title == "SuperSong"
+ assert version == "Remaster"
+ test_str = "SuperSong (Live at Wembley)"
+ title, version = util.parse_title_and_version(test_str)
+ assert title == "SuperSong"
+ assert version == "Live At Wembley"
+ test_str = "SuperSong (Instrumental)"
+ title, version = util.parse_title_and_version(test_str)
+ assert title == "SuperSong"
+ assert version == "Instrumental"
async def test_uri_parsing() -> None:
projects / music-assistant-server.git / commitdiff