music-assistant-server

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