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    # compare version
294    if not compare_version(base_item.version, compare_item.version):
295        return False
296    # finally comparing on (exact) name match
297    return compare_strings(base_item.name, compare_item.name, strict=strict)
298
299
300def compare_audiobook(
301    base_item: Audiobook | ItemMapping,
302    compare_item: Audiobook | ItemMapping,
303    strict: bool = True,
304) -> bool | None:
305    """Compare two Audiobook items and return True if they match."""
306    # return early on exact item_id match
307    if compare_item_ids(base_item, compare_item):
308        return True
309
310    # return early on (un)matched external id
311    for ext_id in (
312        ExternalID.ASIN,
313        ExternalID.BARCODE,
314    ):
315        external_id_match = compare_external_ids(
316            base_item.external_ids, compare_item.external_ids, ext_id
317        )
318        if external_id_match is not None:
319            return external_id_match
320
321    # compare version
322    if not compare_version(base_item.version, compare_item.version):
323        return False
324    # compare name
325    if not compare_strings(base_item.name, compare_item.name, strict=True):
326        return False
327    if not strict and (isinstance(base_item, ItemMapping) or isinstance(compare_item, ItemMapping)):
328        return True
329    # for strict matching we REQUIRE both items to be a real Audiobook object
330    assert isinstance(base_item, Audiobook)
331    assert isinstance(compare_item, Audiobook)
332    # compare publisher
333    if (
334        base_item.publisher
335        and compare_item.publisher
336        and not compare_strings(base_item.publisher, compare_item.publisher, strict=True)
337    ):
338        return False
339
340    def _audiobook_artist_name(value: str | Artist | ItemMapping) -> str:
341        return value.name if isinstance(value, Artist | ItemMapping) else value
342
343    # compare narrator(s) — different narrators indicate different recordings and must not be merged
344    if base_item.narrators and compare_item.narrators:
345        base_narrators = {
346            create_safe_string(_audiobook_artist_name(n)) for n in base_item.narrators
347        }
348        compare_narrators = {
349            create_safe_string(_audiobook_artist_name(n)) for n in compare_item.narrators
350        }
351        if base_narrators.isdisjoint(compare_narrators):
352            return False
353    # compare author(s)
354    for author in base_item.authors:
355        author_safe = create_safe_string(_audiobook_artist_name(author))
356        if author_safe in [
357            create_safe_string(_audiobook_artist_name(x)) for x in compare_item.authors
358        ]:
359            return True
360    return False
361
362
363def compare_podcast(
364    base_item: Podcast | ItemMapping,
365    compare_item: Podcast | ItemMapping,
366    strict: bool = True,
367) -> bool | None:
368    """Compare two Podcast items and return True if they match."""
369    # return early on exact item_id match
370    if compare_item_ids(base_item, compare_item):
371        return True
372
373    # return early on (un)matched external id
374    for ext_id in (
375        ExternalID.ASIN,
376        ExternalID.BARCODE,
377    ):
378        external_id_match = compare_external_ids(
379            base_item.external_ids, compare_item.external_ids, ext_id
380        )
381        if external_id_match is not None:
382            return external_id_match
383
384    # compare version
385    if not compare_version(base_item.version, compare_item.version):
386        return False
387    # compare name
388    if not compare_strings(base_item.name, compare_item.name, strict=True):
389        return False
390    if not strict and (isinstance(base_item, ItemMapping) or isinstance(compare_item, ItemMapping)):
391        return True
392    # for strict matching we REQUIRE both items to be a real Podcast object
393    assert isinstance(base_item, Podcast)
394    assert isinstance(compare_item, Podcast)
395    # compare publisher
396    return not (
397        base_item.publisher
398        and compare_item.publisher
399        and not compare_strings(base_item.publisher, compare_item.publisher, strict=True)
400    )
401
402
403def compare_item_mapping(
404    base_item: ItemMapping,
405    compare_item: ItemMapping,
406    strict: bool = True,
407) -> bool | None:
408    """Compare two ItemMapping items and return True if they match."""
409    # return early on exact item_id match
410    if compare_item_ids(base_item, compare_item):
411        return True
412    # return early on (un)matched external id
413    # check all ExternalID, as ItemMapping is a minimized obj for all MediaItems
414    for ext_id in ExternalID:
415        external_id_match = compare_external_ids(
416            base_item.external_ids, compare_item.external_ids, ext_id
417        )
418        if external_id_match is not None:
419            return external_id_match
420    # compare version
421    if not compare_version(base_item.version, compare_item.version):
422        return False
423    # finally comparing on (exact) name match
424    return compare_strings(base_item.name, compare_item.name, strict=strict)
425
426
427def compare_artists(
428    base_items: list[Artist | ItemMapping],
429    compare_items: list[Artist | ItemMapping],
430    any_match: bool = True,
431) -> bool:
432    """Compare two lists of artist and return True if both lists match (exactly)."""
433    if not base_items or not compare_items:
434        return False
435    # match if first artist matches in both lists
436    if compare_artist(base_items[0], compare_items[0]):
437        return True
438    # compare the artist lists
439    matches = 0
440    for base_item in base_items:
441        for compare_item in compare_items:
442            if compare_artist(base_item, compare_item):
443                if any_match:
444                    return True
445                matches += 1
446    return len(base_items) == len(compare_items) == matches
447
448
449def compare_albums(
450    base_items: list[Album | ItemMapping],
451    compare_items: list[Album | ItemMapping],
452    any_match: bool = True,
453) -> bool:
454    """Compare two lists of albums and return True if a match was found."""
455    matches = 0
456    for base_item in base_items:
457        for compare_item in compare_items:
458            if compare_album(base_item, compare_item):
459                if any_match:
460                    return True
461                matches += 1
462    return len(base_items) == matches
463
464
465def compare_item_ids(
466    base_item: MediaItem | ItemMapping, compare_item: MediaItem | ItemMapping
467) -> bool:
468    """Compare item_id(s) of two media items."""
469    if not base_item.provider or not compare_item.provider:
470        return False
471    if not base_item.item_id or not compare_item.item_id:
472        return False
473    if base_item.provider == compare_item.provider and base_item.item_id == compare_item.item_id:
474        return True
475
476    base_prov_ids = getattr(base_item, "provider_mappings", None)
477    compare_prov_ids = getattr(compare_item, "provider_mappings", None)
478
479    if base_prov_ids is not None:
480        assert isinstance(base_item, MediaItem)  # for type checking
481        for prov_l in base_item.provider_mappings:
482            if (
483                prov_l.provider_instance == compare_item.provider
484                and prov_l.item_id == compare_item.item_id
485            ):
486                return True
487
488    if compare_prov_ids is not None:
489        assert isinstance(compare_item, MediaItem)  # for type checking
490        for prov_r in compare_item.provider_mappings:
491            if (
492                prov_r.provider_instance == base_item.provider
493                and prov_r.item_id == base_item.item_id
494            ):
495                return True
496
497    if base_prov_ids is not None and compare_prov_ids is not None:
498        assert isinstance(base_item, MediaItem)  # for type checking
499        assert isinstance(compare_item, MediaItem)  # for type checking
500        for prov_l in base_item.provider_mappings:
501            for prov_r in compare_item.provider_mappings:
502                if prov_l.provider_domain != prov_r.provider_domain:
503                    continue
504                if (
505                    prov_l.is_unique or prov_r.is_unique
506                ) and prov_l.provider_instance != prov_r.provider_instance:
507                    continue
508                if prov_l.item_id == prov_r.item_id:
509                    return True
510    return False
511
512
513def compare_external_ids(
514    external_ids_base: set[tuple[ExternalID, str]],
515    external_ids_compare: set[tuple[ExternalID, str]],
516    external_id_type: ExternalID,
517) -> bool | None:
518    """Compare external ids and return True if a match was found."""
519    base_ids = {x[1] for x in external_ids_base if x[0] == external_id_type}
520    if not base_ids:
521        # return early if the requested external id type is not present in the base set
522        return None
523    compare_ids = {x[1] for x in external_ids_compare if x[0] == external_id_type}
524    if not compare_ids:
525        # return early if the requested external id type is not present in the compare set
526        return None
527    for base_id in base_ids:
528        if base_id in compare_ids:
529            return True
530        # handle upc stored as EAN-13 barcode
531        if external_id_type == ExternalID.BARCODE and len(base_id) == 12:
532            if f"0{base_id}" in compare_ids:
533                return True
534        # handle EAN-13 stored as UPC barcode
535        if external_id_type == ExternalID.BARCODE and len(base_id) == 13:
536            if base_id[1:] in compare_ids:
537                return True
538        # return false if the identifier is unique (e.g. musicbrainz id)
539        if external_id_type.is_unique:
540            return False
541    return None
542
543
544def loose_compare_strings(base: str, alt: str) -> bool:
545    """Compare strings and return True even on partial match."""
546    # this is used to display 'versions' of the same track/album
547    # where we account for other spelling or some additional wording in the title
548    if len(base) <= 3 or len(alt) <= 3:
549        return compare_strings(base, alt, True)
550    word_count = len(base.strip().split(" "))
551    if word_count == 1 and len(base) < 10:
552        return compare_strings(base, alt, False)
553    base_comp = create_safe_string(base)
554    alt_comp = create_safe_string(alt)
555    if base_comp in alt_comp:
556        return True
557    return base_comp in alt_comp
558
559
560def compare_strings(str1: str, str2: str, strict: bool = True) -> bool:
561    """Compare strings and return True if we have an (almost) perfect match."""
562    if not str1 or not str2:
563        return False
564    str1_lower = str1.lower()
565    str2_lower = str2.lower()
566    if strict:
567        return str1_lower == str2_lower
568    # return early if total length mismatch
569    if abs(len(str1) - len(str2)) > 4:
570        return False
571    # handle '&' vs 'And'
572    if " & " in str1_lower and " and " in str2_lower:
573        str2 = str2_lower.replace(" and ", " & ")
574    elif " and " in str1_lower and " & " in str2:
575        str2 = str2_lower.replace(" & ", " and ")
576    if create_safe_string(str1) == create_safe_string(str2):
577        return True
578    # last resort: use difflib to compare strings
579    required_accuracy = 0.9 if (len(str1) + len(str2)) > 18 else 0.8
580    return SequenceMatcher(a=str1_lower, b=str2_lower).ratio() > required_accuracy
581
582
583def compare_version(base_version: str, compare_version: str) -> bool:
584    """Compare version string."""
585    if not base_version and not compare_version:
586        return True
587    if not base_version and compare_version.lower() in IGNORE_VERSIONS:
588        return True
589    if not compare_version and base_version.lower() in IGNORE_VERSIONS:
590        return True
591    if not base_version and compare_version:
592        return False
593    if base_version and not compare_version:
594        return False
595
596    if " " not in base_version and " " not in compare_version:
597        return compare_strings(base_version, compare_version, False)
598
599    # do this the hard way as sometimes the version string is in the wrong order
600    base_versions = sorted(base_version.lower().split(" "))
601    compare_versions = sorted(compare_version.lower().split(" "))
602    # filter out words we can ignore (such as 'version')
603    ignore_words = [
604        *IGNORE_VERSIONS,
605        "version",
606        "edition",
607        "variant",
608        "versie",
609        "versione",
610    ]
611    base_versions = [x for x in base_versions if x not in ignore_words]
612    compare_versions = [x for x in compare_versions if x not in ignore_words]
613
614    return base_versions == compare_versions
615
616
617def compare_explicit(base: MediaItemMetadata, compare: MediaItemMetadata) -> bool | None:
618    """Compare if explicit is same in metadata."""
619    if base.explicit is not None and compare.explicit is not None:
620        # explicitness info is not always present in metadata
621        # only strict compare them if both have the info set
622        return base.explicit == compare.explicit
623    return None
624