music-assistant-server

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