music-assistant-server

1"""Several helper/utils to compare objects."""
2
3from __future__ import annotations
4
5import re
6from difflib import SequenceMatcher
7from functools import lru_cache
8
9from music_assistant_models.enums import ExternalID, MediaType
10from music_assistant_models.helpers import create_safe_string
11from music_assistant_models.media_items import (
12    Album,
13    Artist,
14    Audiobook,
15    ItemMapping,
16    MediaItem,
17    MediaItemMetadata,
18    MediaItemType,
19    Playlist,
20    Podcast,
21    Radio,
22    Track,
23)
24
25from music_assistant.helpers.external_ids import normalize_external_id
26
27IGNORE_VERSIONS = (
28    "explicit",  # explicit is matched separately
29    "music from and inspired by the motion picture",
30    "original soundtrack",
31    "hi-res",  # quality is handled separately
32)
33
34_VERSION_IGNORE_WORDS = {
35    "album",
36    "edition",
37    "variant",
38    "versie",
39    "version",
40    "versione",
41}
42_VERSION_WORD_ALIASES = {
43    "remastered": "remaster",
44}
45_IGNORE_VERSION_KEYS = {create_safe_string(value) for value in IGNORE_VERSIONS}
46
47
48def compare_media_item(
49    base_item: MediaItemType | ItemMapping,
50    compare_item: MediaItemType | ItemMapping,
51    strict: bool = True,
52) -> bool | None:
53    """Compare two media items and return True if they match."""
54    if base_item.media_type == MediaType.ARTIST and compare_item.media_type == MediaType.ARTIST:
55        assert isinstance(base_item, Artist | ItemMapping)  # for type checking
56        assert isinstance(compare_item, Artist | ItemMapping)  # for type checking
57        return compare_artist(base_item, compare_item, strict)
58    if base_item.media_type == MediaType.ALBUM and compare_item.media_type == MediaType.ALBUM:
59        assert isinstance(base_item, Album | ItemMapping)  # for type checking
60        assert isinstance(compare_item, Album | ItemMapping)  # for type checking
61        return compare_album(base_item, compare_item, strict)
62    if base_item.media_type == MediaType.TRACK and compare_item.media_type == MediaType.TRACK:
63        assert isinstance(base_item, Track)  # for type checking
64        assert isinstance(compare_item, Track)  # for type checking
65        return compare_track(base_item, compare_item, strict)
66    if base_item.media_type == MediaType.PLAYLIST and compare_item.media_type == MediaType.PLAYLIST:
67        assert isinstance(base_item, Playlist | ItemMapping)  # for type checking
68        assert isinstance(compare_item, Playlist | ItemMapping)  # for type checking
69        return compare_playlist(base_item, compare_item, strict)
70    if base_item.media_type == MediaType.RADIO and compare_item.media_type == MediaType.RADIO:
71        assert isinstance(base_item, Radio | ItemMapping)  # for type checking
72        assert isinstance(compare_item, Radio | ItemMapping)  # for type checking
73        return compare_radio(base_item, compare_item, strict)
74    if (
75        base_item.media_type == MediaType.AUDIOBOOK
76        and compare_item.media_type == MediaType.AUDIOBOOK
77    ):
78        assert isinstance(base_item, Audiobook | ItemMapping)  # for type checking
79        assert isinstance(compare_item, Audiobook | ItemMapping)  # for type checking
80        return compare_audiobook(base_item, compare_item, strict)
81    if base_item.media_type == MediaType.PODCAST and compare_item.media_type == MediaType.PODCAST:
82        assert isinstance(base_item, Podcast | ItemMapping)  # for type checking
83        assert isinstance(compare_item, Podcast | ItemMapping)  # for type checking
84        return compare_podcast(base_item, compare_item, strict)
85    assert isinstance(base_item, ItemMapping)  # for type checking
86    assert isinstance(compare_item, ItemMapping)  # for type checking
87    return compare_item_mapping(base_item, compare_item, strict)
88
89
90def compare_artist(
91    base_item: Artist | ItemMapping,
92    compare_item: Artist | ItemMapping,
93    strict: bool = True,
94) -> bool | None:
95    """Compare two artist items and return True if they match."""
96    # return early on exact item_id match
97    if compare_item_ids(base_item, compare_item):
98        return True
99    # return early on (un)matched external id
100    for ext_id in (ExternalID.MB_ARTIST, ExternalID.DISCOGS, ExternalID.TADB):
101        external_id_match = compare_external_ids(
102            base_item.external_ids, compare_item.external_ids, ext_id
103        )
104        if external_id_match is not None:
105            return external_id_match
106    # return early if artist_types don't match
107    if (
108        isinstance(base_item, Artist)
109        and isinstance(compare_item, Artist)
110        and base_item.artist_type != compare_item.artist_type
111    ):
112        return False
113    # finally comparing on (exact) name match
114    return compare_strings(base_item.name, compare_item.name, strict=strict)
115
116
117def compare_album(
118    base_item: Album | ItemMapping,
119    compare_item: Album | ItemMapping,
120    strict: bool = True,
121) -> bool | None:
122    """Compare two album items and return True if they match."""
123    # return early on exact item_id match
124    if compare_item_ids(base_item, compare_item):
125        return True
126
127    # return early on (un)matched authoritative external id
128    for ext_id in (
129        ExternalID.MB_ALBUM,
130        ExternalID.DISCOGS,
131        ExternalID.TADB,
132    ):
133        external_id_match = compare_external_ids(
134            base_item.external_ids, compare_item.external_ids, ext_id
135        )
136        if external_id_match is not None:
137            return external_id_match
138
139    secondary_external_id_match = any(
140        compare_external_ids(base_item.external_ids, compare_item.external_ids, ext_id) is True
141        for ext_id in (ExternalID.ASIN, ExternalID.BARCODE)
142    )
143
144    # compare version
145    if not compare_version(base_item.version, compare_item.version):
146        return False
147    # compare name
148    if not compare_strings(base_item.name, compare_item.name, strict=True):
149        return False
150    if not strict and (isinstance(base_item, ItemMapping) or isinstance(compare_item, ItemMapping)):
151        return True
152    # for strict matching we REQUIRE both items to be a real album object
153    assert isinstance(base_item, Album)
154    assert isinstance(compare_item, Album)
155    # compare year
156    if (
157        base_item.year
158        and compare_item.year
159        and base_item.year != compare_item.year
160        and not secondary_external_id_match
161    ):
162        return False
163    # compare explicitness
164    if compare_explicit(base_item.metadata, compare_item.metadata) is False:
165        return False
166    # compare album artist(s)
167    return compare_artists(base_item.artists, compare_item.artists, not strict)
168
169
170def compare_track(
171    base_item: Track,
172    compare_item: Track,
173    strict: bool = True,
174    track_albums: list[Album] | None = None,
175) -> bool:
176    """Compare two track items and return True if they match."""
177    # return early on exact item_id match
178    if compare_item_ids(base_item, compare_item):
179        return True
180    # tracks on the same album but different discs are always distinct,
181    # even if they share external IDs (e.g. same recording on multiple discs)
182    if (
183        base_item.album
184        and compare_item.album
185        and base_item.disc_number
186        and compare_item.disc_number
187        and base_item.disc_number != compare_item.disc_number
188        and compare_album(base_item.album, compare_item.album, False)
189    ):
190        return False
191    # return early on (un)matched primary/unique external id
192    for ext_id in (
193        ExternalID.MB_RECORDING,
194        ExternalID.MB_TRACK,
195        ExternalID.ACOUSTID,
196    ):
197        external_id_match = compare_external_ids(
198            base_item.external_ids, compare_item.external_ids, ext_id
199        )
200        if external_id_match is not None:
201            return external_id_match
202    # check secondary external id matches
203    for ext_id in (
204        ExternalID.DISCOGS,
205        ExternalID.TADB,
206        ExternalID.ISRC,
207        ExternalID.ASIN,
208    ):
209        external_id_match = compare_external_ids(
210            base_item.external_ids, compare_item.external_ids, ext_id
211        )
212        if external_id_match is True:
213            # we got a 'soft-match' on a secondary external id (like ISRC)
214            # but we do a double check on duration
215            if abs(base_item.duration - compare_item.duration) <= 8:
216                return True
217
218    # compare name
219    if not compare_strings(base_item.name, compare_item.name, strict=True):
220        return False
221    # track artist(s) must match
222    if not compare_artists(base_item.artists, compare_item.artists, any_match=not strict):
223        return False
224    # track version must match
225    if strict and not compare_version(base_item.version, compare_item.version):
226        return False
227    # check if both tracks are (not) explicit
228    if base_item.metadata.explicit is None and isinstance(base_item.album, Album):
229        base_item.metadata.explicit = base_item.album.metadata.explicit
230    if compare_item.metadata.explicit is None and isinstance(compare_item.album, Album):
231        compare_item.metadata.explicit = compare_item.album.metadata.explicit
232    if strict and compare_explicit(base_item.metadata, compare_item.metadata) is False:
233        return False
234
235    # exact albumtrack match = 100% match
236    if (
237        base_item.album
238        and compare_item.album
239        and compare_album(base_item.album, compare_item.album, False)
240        and base_item.disc_number
241        and compare_item.disc_number
242        and base_item.track_number
243        and compare_item.track_number
244        and base_item.disc_number == compare_item.disc_number
245        and base_item.track_number == compare_item.track_number
246    ):
247        return True
248
249    # fallback: exact album match and (near-exact) track duration match
250    if (
251        base_item.album is not None
252        and compare_item.album is not None
253        and (base_item.track_number == 0 or compare_item.track_number == 0)
254        and compare_album(base_item.album, compare_item.album, False)
255        and abs(base_item.duration - compare_item.duration) <= 3
256    ):
257        return True
258
259    # fallback: additional compare albums provided for base track
260    if (
261        compare_item.album is not None
262        and track_albums
263        and abs(base_item.duration - compare_item.duration) <= 3
264    ):
265        for track_album in track_albums:
266            if compare_album(track_album, compare_item.album, False):
267                return True
268
269    # fallback edge case: albumless track with same duration
270    if (
271        base_item.album is None
272        and compare_item.album is None
273        and base_item.disc_number == 0
274        and compare_item.disc_number == 0
275        and base_item.track_number == 0
276        and compare_item.track_number == 0
277        and base_item.duration == compare_item.duration
278    ):
279        return True
280
281    if strict:
282        # in strict mode, we require an exact album match so return False here
283        return False
284
285    # Accept last resort (in non strict mode): (near) exact duration,
286    # otherwise fail all other cases.
287    # Note that as this stage, all other info already matches,
288    # such as title, artist etc.
289    return abs(base_item.duration - compare_item.duration) <= 2
290
291
292def compare_playlist(
293    base_item: Playlist | ItemMapping,
294    compare_item: Playlist | ItemMapping,
295    strict: bool = True,
296) -> bool | None:
297    """Compare two Playlist items and return True if they match."""
298    # require (exact) name match
299    if not compare_strings(base_item.name, compare_item.name, strict=strict):
300        return False
301    # require exact owner match (if not ItemMapping)
302    if isinstance(base_item, Playlist) and isinstance(compare_item, Playlist):
303        if not compare_strings(base_item.owner, compare_item.owner):
304            return False
305    # a playlist is always unique - so do a strict compare on item id(s)
306    return compare_item_ids(base_item, compare_item)
307
308
309def compare_radio(
310    base_item: Radio | ItemMapping,
311    compare_item: Radio | ItemMapping,
312    strict: bool = True,
313) -> bool | None:
314    """Compare two Radio items and return True if they match."""
315    # return early on exact item_id match
316    if compare_item_ids(base_item, compare_item):
317        return True
318    # a dynamic station is its provider's own, so a same-named station is a different one
319    if _is_dynamic_radio(base_item) or _is_dynamic_radio(compare_item):
320        return False
321    # compare version
322    if not compare_version(base_item.version, compare_item.version):
323        return False
324    # finally comparing on (exact) name match
325    return compare_strings(base_item.name, compare_item.name, strict=strict)
326
327
328def compare_audiobook(
329    base_item: Audiobook | ItemMapping,
330    compare_item: Audiobook | ItemMapping,
331    strict: bool = True,
332) -> bool | None:
333    """Compare two Audiobook items and return True if they match."""
334    # return early on exact item_id match
335    if compare_item_ids(base_item, compare_item):
336        return True
337
338    # return early on (un)matched external id
339    for ext_id in (
340        ExternalID.ASIN,
341        ExternalID.BARCODE,
342    ):
343        external_id_match = compare_external_ids(
344            base_item.external_ids, compare_item.external_ids, ext_id
345        )
346        if external_id_match is not None:
347            return external_id_match
348
349    # compare version
350    if not compare_version(base_item.version, compare_item.version):
351        return False
352    # compare name
353    if not compare_strings(base_item.name, compare_item.name, strict=True):
354        return False
355    if not strict and (isinstance(base_item, ItemMapping) or isinstance(compare_item, ItemMapping)):
356        return True
357    # for strict matching we REQUIRE both items to be a real Audiobook object
358    assert isinstance(base_item, Audiobook)
359    assert isinstance(compare_item, Audiobook)
360    # compare publisher
361    if (
362        base_item.publisher
363        and compare_item.publisher
364        and not compare_strings(base_item.publisher, compare_item.publisher, strict=True)
365    ):
366        return False
367
368    def _audiobook_artist_name(value: str | Artist | ItemMapping) -> str:
369        return value.name if isinstance(value, Artist | ItemMapping) else value
370
371    # compare narrator(s) — different narrators indicate different recordings and must not be merged
372    if base_item.narrators and compare_item.narrators:
373        base_narrators = {
374            create_safe_string(_audiobook_artist_name(n)) for n in base_item.narrators
375        }
376        compare_narrators = {
377            create_safe_string(_audiobook_artist_name(n)) for n in compare_item.narrators
378        }
379        if base_narrators.isdisjoint(compare_narrators):
380            return False
381    # compare author(s)
382    for author in base_item.authors:
383        author_safe = create_safe_string(_audiobook_artist_name(author))
384        if author_safe in [
385            create_safe_string(_audiobook_artist_name(x)) for x in compare_item.authors
386        ]:
387            return True
388    return False
389
390
391def compare_podcast(
392    base_item: Podcast | ItemMapping,
393    compare_item: Podcast | ItemMapping,
394    strict: bool = True,
395) -> bool | None:
396    """Compare two Podcast items and return True if they match."""
397    # return early on exact item_id match
398    if compare_item_ids(base_item, compare_item):
399        return True
400
401    # return early on (un)matched external id
402    for ext_id in (
403        ExternalID.ASIN,
404        ExternalID.BARCODE,
405    ):
406        external_id_match = compare_external_ids(
407            base_item.external_ids, compare_item.external_ids, ext_id
408        )
409        if external_id_match is not None:
410            return external_id_match
411
412    # compare version
413    if not compare_version(base_item.version, compare_item.version):
414        return False
415    # compare name
416    if not compare_strings(base_item.name, compare_item.name, strict=True):
417        return False
418    if not strict and (isinstance(base_item, ItemMapping) or isinstance(compare_item, ItemMapping)):
419        return True
420    # for strict matching we REQUIRE both items to be a real Podcast object
421    assert isinstance(base_item, Podcast)
422    assert isinstance(compare_item, Podcast)
423    # compare publisher
424    return not (
425        base_item.publisher
426        and compare_item.publisher
427        and not compare_strings(base_item.publisher, compare_item.publisher, strict=True)
428    )
429
430
431def compare_item_mapping(
432    base_item: ItemMapping,
433    compare_item: ItemMapping,
434    strict: bool = True,
435) -> bool | None:
436    """Compare two ItemMapping items and return True if they match."""
437    # return early on exact item_id match
438    if compare_item_ids(base_item, compare_item):
439        return True
440    # return early on (un)matched external id
441    # check all ExternalID, as ItemMapping is a minimized obj for all MediaItems
442    for ext_id in ExternalID:
443        external_id_match = compare_external_ids(
444            base_item.external_ids, compare_item.external_ids, ext_id
445        )
446        if external_id_match is not None:
447            return external_id_match
448    # compare version
449    if not compare_version(base_item.version, compare_item.version):
450        return False
451    # finally comparing on (exact) name match
452    return compare_strings(base_item.name, compare_item.name, strict=strict)
453
454
455def compare_artists(
456    base_items: list[Artist | ItemMapping],
457    compare_items: list[Artist | ItemMapping],
458    any_match: bool = True,
459) -> bool:
460    """Compare two lists of artist and return True if both lists match (exactly)."""
461    if not base_items or not compare_items:
462        return False
463    # match if first artist matches in both lists
464    if compare_artist(base_items[0], compare_items[0]):
465        return True
466    # compare the artist lists
467    matches = 0
468    for base_item in base_items:
469        for compare_item in compare_items:
470            if compare_artist(base_item, compare_item):
471                if any_match:
472                    return True
473                matches += 1
474    return len(base_items) == len(compare_items) == matches
475
476
477def compare_albums(
478    base_items: list[Album | ItemMapping],
479    compare_items: list[Album | ItemMapping],
480    any_match: bool = True,
481) -> bool:
482    """Compare two lists of albums and return True if a match was found."""
483    matches = 0
484    for base_item in base_items:
485        for compare_item in compare_items:
486            if compare_album(base_item, compare_item):
487                if any_match:
488                    return True
489                matches += 1
490    return len(base_items) == matches
491
492
493def compare_item_ids(
494    base_item: MediaItem | ItemMapping, compare_item: MediaItem | ItemMapping
495) -> bool:
496    """Compare item_id(s) of two media items."""
497    if not base_item.provider or not compare_item.provider:
498        return False
499    if not base_item.item_id or not compare_item.item_id:
500        return False
501    if base_item.provider == compare_item.provider and base_item.item_id == compare_item.item_id:
502        return True
503
504    base_prov_ids = getattr(base_item, "provider_mappings", None)
505    compare_prov_ids = getattr(compare_item, "provider_mappings", None)
506
507    if base_prov_ids is not None:
508        assert isinstance(base_item, MediaItem)  # for type checking
509        for prov_l in base_item.provider_mappings:
510            if (
511                prov_l.provider_instance == compare_item.provider
512                and prov_l.item_id == compare_item.item_id
513            ):
514                return True
515
516    if compare_prov_ids is not None:
517        assert isinstance(compare_item, MediaItem)  # for type checking
518        for prov_r in compare_item.provider_mappings:
519            if (
520                prov_r.provider_instance == base_item.provider
521                and prov_r.item_id == base_item.item_id
522            ):
523                return True
524
525    if base_prov_ids is not None and compare_prov_ids is not None:
526        assert isinstance(base_item, MediaItem)  # for type checking
527        assert isinstance(compare_item, MediaItem)  # for type checking
528        for prov_l in base_item.provider_mappings:
529            for prov_r in compare_item.provider_mappings:
530                if prov_l.provider_domain != prov_r.provider_domain:
531                    continue
532                if (
533                    prov_l.is_unique or prov_r.is_unique
534                ) and prov_l.provider_instance != prov_r.provider_instance:
535                    continue
536                if prov_l.item_id == prov_r.item_id:
537                    return True
538    return False
539
540
541def compare_external_ids(
542    external_ids_base: set[tuple[ExternalID, str]],
543    external_ids_compare: set[tuple[ExternalID, str]],
544    external_id_type: ExternalID,
545) -> bool | None:
546    """Compare external ids and return True if a match was found."""
547    base_ids = {
548        normalize_external_id(external_id_type, value)
549        for current_type, value in external_ids_base
550        if current_type == external_id_type
551    }
552    if not base_ids:
553        # return early if the requested external id type is not present in the base set
554        return None
555    compare_ids = {
556        normalize_external_id(external_id_type, value)
557        for current_type, value in external_ids_compare
558        if current_type == external_id_type
559    }
560    if not compare_ids:
561        # return early if the requested external id type is not present in the compare set
562        return None
563    if base_ids.intersection(compare_ids):
564        return True
565    if external_id_type.is_unique:
566        return False
567    return None
568
569
570def loose_compare_strings(base: str, alt: str) -> bool:
571    """Compare strings and return True even on partial match."""
572    # this is used to display 'versions' of the same track/album
573    # where we account for other spelling or some additional wording in the title
574    if len(base) <= 3 or len(alt) <= 3:
575        return compare_strings(base, alt, True)
576    word_count = len(base.strip().split(" "))
577    if word_count == 1 and len(base) < 10:
578        return compare_strings(base, alt, False)
579    base_comp = create_safe_string(base)
580    alt_comp = create_safe_string(alt)
581    if base_comp in alt_comp:
582        return True
583    return base_comp in alt_comp
584
585
586def compare_strings(str1: str, str2: str, strict: bool = True) -> bool:
587    """Compare strings and return True if we have an (almost) perfect match."""
588    if not str1 or not str2:
589        return False
590    str1_lower = str1.lower()
591    str2_lower = str2.lower()
592    if strict:
593        return str1_lower == str2_lower
594    # return early if total length mismatch
595    if abs(len(str1) - len(str2)) > 4:
596        return False
597    # handle '&' vs 'And'
598    if " & " in str1_lower and " and " in str2_lower:
599        str2 = str2_lower.replace(" and ", " & ")
600    elif " and " in str1_lower and " & " in str2:
601        str2 = str2_lower.replace(" & ", " and ")
602    if create_safe_string(str1) == create_safe_string(str2):
603        return True
604    # last resort: use difflib to compare strings
605    required_accuracy = 0.9 if (len(str1) + len(str2)) > 18 else 0.8
606    return SequenceMatcher(a=str1_lower, b=str2_lower).ratio() > required_accuracy
607
608
609def compare_version(base_version: str, compare_version: str) -> bool:
610    """Compare version string."""
611    return _normalize_version_tokens(base_version) == _normalize_version_tokens(compare_version)
612
613
614def compare_explicit(base: MediaItemMetadata, compare: MediaItemMetadata) -> bool | None:
615    """Compare if explicit is same in metadata."""
616    if base.explicit is not None and compare.explicit is not None:
617        # explicitness info is not always present in metadata
618        # only strict compare them if both have the info set
619        return base.explicit == compare.explicit
620    return None
621
622
623@lru_cache(maxsize=1024)
624def _normalize_version_tokens(value: str) -> tuple[str, ...]:
625    """Return meaningful version tokens in stable order."""
626    if not value or create_safe_string(value) in _IGNORE_VERSION_KEYS:
627        return ()
628    tokens = (
629        _VERSION_WORD_ALIASES.get(token, token)
630        for token in re.findall(r"[^\W_]+", value.casefold())
631    )
632    return tuple(sorted(token for token in tokens if token not in _VERSION_IGNORE_WORDS))
633
634
635def _is_dynamic_radio(item: Radio | ItemMapping) -> bool:
636    """Return True if the item is a dynamic radio station."""
637    return isinstance(item, Radio) and item.is_dynamic
638