music-assistant-server

1"""Several helper/utils to compare objects."""
2
3from __future__ import annotations
4
5import re
6import unicodedata
7from collections.abc import Sequence
8from difflib import SequenceMatcher
9from enum import Enum
10from functools import lru_cache
11from typing import Final
12
13from music_assistant_models.enums import ExternalID, MediaType
14from music_assistant_models.helpers import create_safe_string
15from music_assistant_models.media_items import (
16    Album,
17    Artist,
18    Audiobook,
19    ItemMapping,
20    MediaItem,
21    MediaItemMetadata,
22    MediaItemType,
23    Playlist,
24    Podcast,
25    Radio,
26    Track,
27)
28
29from music_assistant.helpers.external_ids import is_valid_isrc, normalize_external_id
30
31IGNORE_VERSIONS = (
32    "explicit",  # explicit is matched separately
33    "music from and inspired by the motion picture",
34    "original soundtrack",
35    "hi-res",  # quality is handled separately
36)
37
38_VERSION_IGNORE_WORDS = {
39    "album",
40    "at",
41    "edition",
42    "variant",
43    "versie",
44    "version",
45    "versione",
46}
47_VERSION_WORD_ALIASES = {
48    "remastered": "remaster",
49}
50# phrases stripped from a version before tokenizing: they may contain punctuation
51# ("hi-res") that the tokenizer would otherwise split into meaningful-looking tokens
52_IGNORE_VERSION_PATTERNS = tuple(
53    re.compile(rf"\b{re.escape(phrase)}\b", re.IGNORECASE) for phrase in IGNORE_VERSIONS
54)
55
56# version tokens that signal a fundamentally different recording (not just packaging),
57# so they must never be treated as an ambiguous/mergeable edition difference
58_RECORDING_CONFLICT_VERSION_TOKENS = {
59    "acoustic",
60    "cover",
61    "demo",
62    "instrumental",
63    "karaoke",
64    "live",
65    "remix",
66    "session",
67}
68
69# retail suffixes a provider (notably Apple Music) appends to an EP/single title
70_ALBUM_RETAIL_SUFFIXES: Final = ("EP", "Single")
71# the trailing retail suffix as it appears in a raw album title: set off by a dash
72# (any style, and only the space in front of it counts, so "K-EP" keeps its name) or
73# wrapped in brackets, which need no space to be unambiguous. A bare trailing word is
74# deliberately not accepted, as it is just as likely part of the title itself
75# ("The SL2 EP", "Saturday Night Single")
76_ALBUM_SUFFIX_PATTERN = re.compile(
77    rf"\s+[-\u2013\u2014]\s*(?P<suffix>{'|'.join(_ALBUM_RETAIL_SUFFIXES)})\s*$"
78    rf"|\s*[(\[](?P<bracketed>{'|'.join(_ALBUM_RETAIL_SUFFIXES)})[)\]]\s*$",
79    re.IGNORECASE,
80)
81# normalizing a title drops the separator, so the suffix survives as a plain trailing
82# fragment of the name key ("Foo - EP" -> "fooep"): appending one of these to a key
83# yields the key the same album is stored under when a provider spells out the suffix
84ALBUM_RETAIL_SUFFIX_KEYS: Final = tuple(
85    create_safe_string(suffix, True, True) for suffix in _ALBUM_RETAIL_SUFFIXES
86)
87
88# duration tolerances (seconds) for track comparisons: an external-id corroborated
89# match allows more duration drift than a bare title/version fallback
90_ISRC_DURATION_TOLERANCE = 8
91_FALLBACK_DURATION_TOLERANCE = 2
92
93
94class AlbumMatchEvidence(Enum):
95    """Confidence level for an album identity comparison."""
96
97    MATCH = "match"
98    NO_MATCH = "no_match"
99    INSUFFICIENT = "insufficient"
100
101
102def compare_media_item(
103    base_item: MediaItemType | ItemMapping,
104    compare_item: MediaItemType | ItemMapping,
105    strict: bool = True,
106) -> bool | None:
107    """Compare two media items and return True if they match."""
108    if base_item.media_type == MediaType.ARTIST and compare_item.media_type == MediaType.ARTIST:
109        assert isinstance(base_item, Artist | ItemMapping)  # for type checking
110        assert isinstance(compare_item, Artist | ItemMapping)  # for type checking
111        return compare_artist(base_item, compare_item, strict)
112    if base_item.media_type == MediaType.ALBUM and compare_item.media_type == MediaType.ALBUM:
113        assert isinstance(base_item, Album | ItemMapping)  # for type checking
114        assert isinstance(compare_item, Album | ItemMapping)  # for type checking
115        return compare_album(base_item, compare_item, strict)
116    if base_item.media_type == MediaType.TRACK and compare_item.media_type == MediaType.TRACK:
117        assert isinstance(base_item, Track)  # for type checking
118        assert isinstance(compare_item, Track)  # for type checking
119        return compare_track(base_item, compare_item, strict)
120    if base_item.media_type == MediaType.PLAYLIST and compare_item.media_type == MediaType.PLAYLIST:
121        assert isinstance(base_item, Playlist | ItemMapping)  # for type checking
122        assert isinstance(compare_item, Playlist | ItemMapping)  # for type checking
123        return compare_playlist(base_item, compare_item, strict)
124    if base_item.media_type == MediaType.RADIO and compare_item.media_type == MediaType.RADIO:
125        assert isinstance(base_item, Radio | ItemMapping)  # for type checking
126        assert isinstance(compare_item, Radio | ItemMapping)  # for type checking
127        return compare_radio(base_item, compare_item, strict)
128    if (
129        base_item.media_type == MediaType.AUDIOBOOK
130        and compare_item.media_type == MediaType.AUDIOBOOK
131    ):
132        assert isinstance(base_item, Audiobook | ItemMapping)  # for type checking
133        assert isinstance(compare_item, Audiobook | ItemMapping)  # for type checking
134        return compare_audiobook(base_item, compare_item, strict)
135    if base_item.media_type == MediaType.PODCAST and compare_item.media_type == MediaType.PODCAST:
136        assert isinstance(base_item, Podcast | ItemMapping)  # for type checking
137        assert isinstance(compare_item, Podcast | ItemMapping)  # for type checking
138        return compare_podcast(base_item, compare_item, strict)
139    assert isinstance(base_item, ItemMapping)  # for type checking
140    assert isinstance(compare_item, ItemMapping)  # for type checking
141    return compare_item_mapping(base_item, compare_item, strict)
142
143
144def compare_artist(
145    base_item: Artist | ItemMapping,
146    compare_item: Artist | ItemMapping,
147    strict: bool = True,
148) -> bool | None:
149    """Compare two artist items and return True if they match."""
150    # return early on exact item_id match
151    if compare_item_ids(base_item, compare_item):
152        return True
153    # return early on (un)matched external id
154    for ext_id in (ExternalID.MB_ARTIST, ExternalID.DISCOGS, ExternalID.TADB):
155        external_id_match = compare_external_ids(
156            base_item.external_ids, compare_item.external_ids, ext_id
157        )
158        if external_id_match is not None:
159            return external_id_match
160    # return early if artist_types don't match
161    if (
162        isinstance(base_item, Artist)
163        and isinstance(compare_item, Artist)
164        and base_item.artist_type != compare_item.artist_type
165    ):
166        return False
167    # finally comparing on (exact) name match
168    return compare_strings(base_item.name, compare_item.name, strict=strict)
169
170
171def compare_album(
172    base_item: Album | ItemMapping,
173    compare_item: Album | ItemMapping,
174    strict: bool = True,
175) -> bool | None:
176    """Compare two album items and return True if they match."""
177    return compare_album_evidence(base_item, compare_item, strict) == AlbumMatchEvidence.MATCH
178
179
180def compare_album_evidence(
181    base_item: Album | ItemMapping,
182    compare_item: Album | ItemMapping,
183    strict: bool = True,
184    base_tracks: Sequence[Track] | None = None,
185    compare_tracks: Sequence[Track] | None = None,
186) -> AlbumMatchEvidence:
187    """
188    Return the match evidence for two album items.
189
190    Unlike `compare_album`, this distinguishes a confident non-match from
191    insufficient metadata (e.g. an edition difference that cannot be resolved from
192    the album's own fields), so a caller that can fetch tracklists knows when doing
193    so may still resolve the comparison. If `base_tracks`/`compare_tracks` are
194    supplied, an ordered track fingerprint comparison is used to resolve that
195    remaining ambiguity, and a conflicting fingerprint overrides an otherwise
196    nominally-matching album (e.g. identical title/version/year but a different
197    number of tracks).
198
199    :param base_tracks: Ordered tracklist for base_item, if already available to the caller.
200    :param compare_tracks: Ordered tracklist for compare_item, if already available.
201    """
202    # return early on exact item_id match
203    if compare_item_ids(base_item, compare_item):
204        return AlbumMatchEvidence.MATCH
205
206    # return early on (un)matched authoritative external id
207    for ext_id in (
208        ExternalID.MB_ALBUM,
209        ExternalID.DISCOGS,
210        ExternalID.TADB,
211    ):
212        external_id_match = compare_external_ids(
213            base_item.external_ids, compare_item.external_ids, ext_id
214        )
215        if external_id_match is not None:
216            return AlbumMatchEvidence.MATCH if external_id_match else AlbumMatchEvidence.NO_MATCH
217
218    # barcode/ASIN are shared across pressings and are non-unique corroboration only,
219    # so they are never used on their own, only to resolve a year or edition ambiguity below
220    secondary_external_id_match = any(
221        compare_external_ids(base_item.external_ids, compare_item.external_ids, ext_id) is True
222        for ext_id in (ExternalID.ASIN, ExternalID.BARCODE)
223    )
224
225    # a real edition conflict (e.g. deluxe vs. live) is decisive, an ambiguous
226    # subset/superset wording (e.g. "2022 Remaster" vs "Deluxe 2022 Remaster") is not
227    version_evidence = _compare_album_version(base_item.version, compare_item.version)
228    if version_evidence == AlbumMatchEvidence.NO_MATCH:
229        return AlbumMatchEvidence.NO_MATCH
230    # compare name
231    if not compare_album_name(base_item.name, compare_item.name):
232        return AlbumMatchEvidence.NO_MATCH
233
234    ambiguous = version_evidence == AlbumMatchEvidence.INSUFFICIENT
235    if ambiguous and secondary_external_id_match:
236        # a shared barcode/ASIN identifies the same retail product, which resolves an
237        # ambiguous edition wording; when the caller supplies tracklists, a conflicting
238        # fingerprint still overrides
239        ambiguous = False
240    if not strict and (isinstance(base_item, ItemMapping) or isinstance(compare_item, ItemMapping)):
241        return _finalize_album_evidence(ambiguous, base_tracks, compare_tracks)
242    # for strict matching we REQUIRE both items to be a real album object
243    assert isinstance(base_item, Album)
244    assert isinstance(compare_item, Album)
245    # compare year: without corroboration this is provider drift, not proof either way
246    if (
247        base_item.year
248        and compare_item.year
249        and base_item.year != compare_item.year
250        and not secondary_external_id_match
251    ):
252        ambiguous = True
253    # compare explicitness
254    if compare_explicit(base_item.metadata, compare_item.metadata) is False:
255        return AlbumMatchEvidence.NO_MATCH
256    # compare album artist(s)
257    if not compare_artists(base_item.artists, compare_item.artists, not strict):
258        return AlbumMatchEvidence.NO_MATCH
259    return _finalize_album_evidence(ambiguous, base_tracks, compare_tracks)
260
261
262def compare_album_track_fingerprint(
263    base_tracks: Sequence[Track] | None,
264    compare_tracks: Sequence[Track] | None,
265) -> AlbumMatchEvidence:
266    """
267    Compare two album tracklists position-by-position and return match evidence.
268
269    Requires an identical disc/track shape to consider two tracklists the same
270    edition; a tracklist that never reports a disc number is treated as insufficient
271    (not assumed disc 1) when compared against a genuinely multi-disc tracklist. At
272    each position, a shared (normalized) ISRC with a compatible duration is preferred
273    as identity evidence; conflicting ISRCs indicate a different recording/remaster.
274    Positions without a usable ISRC on either side fall back to a normalized
275    title/version match with a tight duration tolerance.
276
277    :param base_tracks: Ordered tracklist for the base album.
278    :param compare_tracks: Ordered tracklist for the album being compared.
279    """
280    if not base_tracks or not compare_tracks:
281        return AlbumMatchEvidence.INSUFFICIENT
282    base_positions = _track_positions(base_tracks)
283    compare_positions = _track_positions(compare_tracks)
284    if not base_positions or not compare_positions:
285        return AlbumMatchEvidence.INSUFFICIENT
286    base_is_multi_disc = any(disc_number > 1 for disc_number, _ in base_positions)
287    compare_is_multi_disc = any(disc_number > 1 for disc_number, _ in compare_positions)
288    if (base_is_multi_disc and _has_unknown_disc_layout(compare_tracks)) or (
289        compare_is_multi_disc and _has_unknown_disc_layout(base_tracks)
290    ):
291        # one side never reports a disc number while the other is genuinely multi-disc:
292        # assuming disc 1 for the unknown side would produce a false shape conflict
293        return AlbumMatchEvidence.INSUFFICIENT
294    if base_positions.keys() != compare_positions.keys():
295        # different disc/track shape (e.g. a bonus disc or missing tracks): different edition
296        return AlbumMatchEvidence.NO_MATCH
297
298    evidence = AlbumMatchEvidence.MATCH
299    for position, base_track in base_positions.items():
300        position_evidence = _compare_track_fingerprint(base_track, compare_positions[position])
301        if position_evidence == AlbumMatchEvidence.NO_MATCH:
302            return AlbumMatchEvidence.NO_MATCH
303        if position_evidence == AlbumMatchEvidence.INSUFFICIENT:
304            evidence = AlbumMatchEvidence.INSUFFICIENT
305    return evidence
306
307
308def album_tracks_have_positions(tracks: Sequence[Track] | None) -> bool:
309    """
310    Return True if a tracklist has a trustworthy, unambiguous disc/track layout.
311
312    A caller choosing a base tracklist for album-track fingerprinting can use this to
313    reject a tracklist whose positions cannot be trusted (a missing disc or track number,
314    or a duplicate position) and fall back to another source instead.
315
316    :param tracks: Tracklist to inspect.
317    """
318    if not tracks:
319        return False
320    # a missing disc or track number is treated as unknown rather than silently assumed,
321    # so such a tracklist is not trusted as a shape reference
322    if any(not track.disc_number or not track.track_number for track in tracks):
323        return False
324    return bool(_track_positions(tracks))
325
326
327def compare_track(
328    base_item: Track,
329    compare_item: Track,
330    strict: bool = True,
331    track_albums: list[Album] | None = None,
332) -> bool:
333    """Compare two track 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    # tracks on the same album but different discs are always distinct,
338    # even if they share external IDs (e.g. same recording on multiple discs)
339    if (
340        base_item.album
341        and compare_item.album
342        and base_item.disc_number
343        and compare_item.disc_number
344        and base_item.disc_number != compare_item.disc_number
345        and compare_album(base_item.album, compare_item.album, False)
346    ):
347        return False
348    # return early on (un)matched primary/unique external id
349    for ext_id in (
350        ExternalID.MB_RECORDING,
351        ExternalID.MB_TRACK,
352        ExternalID.ACOUSTID,
353    ):
354        external_id_match = compare_external_ids(
355            base_item.external_ids, compare_item.external_ids, ext_id
356        )
357        if external_id_match is not None:
358            return external_id_match
359    # check secondary external id matches
360    for ext_id in (
361        ExternalID.DISCOGS,
362        ExternalID.TADB,
363        ExternalID.ISRC,
364        ExternalID.ASIN,
365    ):
366        external_id_match = compare_external_ids(
367            base_item.external_ids, compare_item.external_ids, ext_id
368        )
369        if external_id_match is True:
370            # we got a 'soft-match' on a secondary external id (like ISRC)
371            # but we do a double check on duration
372            if abs(base_item.duration - compare_item.duration) <= _ISRC_DURATION_TOLERANCE:
373                return True
374
375    # compare name
376    if not compare_strings(base_item.name, compare_item.name, strict=True):
377        return False
378    # track artist(s) must match
379    if not compare_artists(base_item.artists, compare_item.artists, any_match=not strict):
380        return False
381    # track version must match
382    if strict and not compare_version(base_item.version, compare_item.version):
383        return False
384    # check if both tracks are (not) explicit
385    if base_item.metadata.explicit is None and isinstance(base_item.album, Album):
386        base_item.metadata.explicit = base_item.album.metadata.explicit
387    if compare_item.metadata.explicit is None and isinstance(compare_item.album, Album):
388        compare_item.metadata.explicit = compare_item.album.metadata.explicit
389    if strict and compare_explicit(base_item.metadata, compare_item.metadata) is False:
390        return False
391
392    # exact albumtrack match = 100% match
393    # a missing disc number means unknown: assume disc 1 (local files often omit the tag)
394    if (
395        base_item.album
396        and compare_item.album
397        and compare_album(base_item.album, compare_item.album, False)
398        and base_item.track_number
399        and compare_item.track_number
400        and (base_item.disc_number or 1) == (compare_item.disc_number or 1)
401        and base_item.track_number == compare_item.track_number
402    ):
403        return True
404
405    # fallback: exact album match and (near-exact) track duration match
406    if (
407        base_item.album is not None
408        and compare_item.album is not None
409        and (base_item.track_number == 0 or compare_item.track_number == 0)
410        and compare_album(base_item.album, compare_item.album, False)
411        and abs(base_item.duration - compare_item.duration) <= 3
412    ):
413        return True
414
415    # fallback: additional compare albums provided for base track
416    if (
417        compare_item.album is not None
418        and track_albums
419        and abs(base_item.duration - compare_item.duration) <= 3
420    ):
421        for track_album in track_albums:
422            if compare_album(track_album, compare_item.album, False):
423                return True
424
425    # fallback edge case: albumless track with same duration
426    if (
427        base_item.album is None
428        and compare_item.album is None
429        and base_item.disc_number == 0
430        and compare_item.disc_number == 0
431        and base_item.track_number == 0
432        and compare_item.track_number == 0
433        and base_item.duration == compare_item.duration
434    ):
435        return True
436
437    if strict:
438        # in strict mode, we require an exact album match so return False here
439        return False
440
441    # Accept last resort (in non strict mode): (near) exact duration,
442    # otherwise fail all other cases.
443    # Note that as this stage, all other info already matches,
444    # such as title, artist etc.
445    return abs(base_item.duration - compare_item.duration) <= 2
446
447
448def compare_playlist(
449    base_item: Playlist | ItemMapping,
450    compare_item: Playlist | ItemMapping,
451    strict: bool = True,
452) -> bool | None:
453    """Compare two Playlist items and return True if they match."""
454    # require (exact) name match
455    if not compare_strings(base_item.name, compare_item.name, strict=strict):
456        return False
457    # require exact owner match (if not ItemMapping)
458    if isinstance(base_item, Playlist) and isinstance(compare_item, Playlist):
459        if not compare_strings(base_item.owner, compare_item.owner):
460            return False
461    # a playlist is always unique - so do a strict compare on item id(s)
462    return compare_item_ids(base_item, compare_item)
463
464
465def compare_radio(
466    base_item: Radio | ItemMapping,
467    compare_item: Radio | ItemMapping,
468    strict: bool = True,
469) -> bool | None:
470    """Compare two Radio items and return True if they match."""
471    # return early on exact item_id match
472    if compare_item_ids(base_item, compare_item):
473        return True
474    # a dynamic station is its provider's own, so a same-named station is a different one
475    if _is_dynamic_radio(base_item) or _is_dynamic_radio(compare_item):
476        return False
477    # compare version
478    if not compare_version(base_item.version, compare_item.version):
479        return False
480    # finally comparing on (exact) name match
481    return compare_strings(base_item.name, compare_item.name, strict=strict)
482
483
484def compare_audiobook(
485    base_item: Audiobook | ItemMapping,
486    compare_item: Audiobook | ItemMapping,
487    strict: bool = True,
488) -> bool | None:
489    """Compare two Audiobook items and return True if they match."""
490    # return early on exact item_id match
491    if compare_item_ids(base_item, compare_item):
492        return True
493
494    # return early on (un)matched external id
495    for ext_id in (
496        ExternalID.ASIN,
497        ExternalID.BARCODE,
498    ):
499        external_id_match = compare_external_ids(
500            base_item.external_ids, compare_item.external_ids, ext_id
501        )
502        if external_id_match is not None:
503            return external_id_match
504
505    # compare version
506    if not compare_version(base_item.version, compare_item.version):
507        return False
508    # compare name
509    if not compare_strings(base_item.name, compare_item.name, strict=True):
510        return False
511    if not strict and (isinstance(base_item, ItemMapping) or isinstance(compare_item, ItemMapping)):
512        return True
513    # for strict matching we REQUIRE both items to be a real Audiobook object
514    assert isinstance(base_item, Audiobook)
515    assert isinstance(compare_item, Audiobook)
516    # compare publisher
517    if (
518        base_item.publisher
519        and compare_item.publisher
520        and not compare_strings(base_item.publisher, compare_item.publisher, strict=True)
521    ):
522        return False
523
524    def _audiobook_artist_name(value: str | Artist | ItemMapping) -> str:
525        return value.name if isinstance(value, Artist | ItemMapping) else value
526
527    # compare narrator(s) — different narrators indicate different recordings and must not be merged
528    if base_item.narrators and compare_item.narrators:
529        base_narrators = {
530            create_safe_string(_audiobook_artist_name(n)) for n in base_item.narrators
531        }
532        compare_narrators = {
533            create_safe_string(_audiobook_artist_name(n)) for n in compare_item.narrators
534        }
535        if base_narrators.isdisjoint(compare_narrators):
536            return False
537    # compare author(s)
538    for author in base_item.authors:
539        author_safe = create_safe_string(_audiobook_artist_name(author))
540        if author_safe in [
541            create_safe_string(_audiobook_artist_name(x)) for x in compare_item.authors
542        ]:
543            return True
544    return False
545
546
547def compare_podcast(
548    base_item: Podcast | ItemMapping,
549    compare_item: Podcast | ItemMapping,
550    strict: bool = True,
551) -> bool | None:
552    """Compare two Podcast items and return True if they match."""
553    # return early on exact item_id match
554    if compare_item_ids(base_item, compare_item):
555        return True
556
557    # return early on (un)matched external id
558    for ext_id in (
559        ExternalID.ASIN,
560        ExternalID.BARCODE,
561    ):
562        external_id_match = compare_external_ids(
563            base_item.external_ids, compare_item.external_ids, ext_id
564        )
565        if external_id_match is not None:
566            return external_id_match
567
568    # compare version
569    if not compare_version(base_item.version, compare_item.version):
570        return False
571    # compare name
572    if not compare_strings(base_item.name, compare_item.name, strict=True):
573        return False
574    if not strict and (isinstance(base_item, ItemMapping) or isinstance(compare_item, ItemMapping)):
575        return True
576    # for strict matching we REQUIRE both items to be a real Podcast object
577    assert isinstance(base_item, Podcast)
578    assert isinstance(compare_item, Podcast)
579    # compare publisher
580    return not (
581        base_item.publisher
582        and compare_item.publisher
583        and not compare_strings(base_item.publisher, compare_item.publisher, strict=True)
584    )
585
586
587def compare_item_mapping(
588    base_item: ItemMapping,
589    compare_item: ItemMapping,
590    strict: bool = True,
591) -> bool | None:
592    """Compare two ItemMapping items and return True if they match."""
593    # return early on exact item_id match
594    if compare_item_ids(base_item, compare_item):
595        return True
596    # return early on (un)matched external id
597    # check all ExternalID, as ItemMapping is a minimized obj for all MediaItems
598    for ext_id in ExternalID:
599        external_id_match = compare_external_ids(
600            base_item.external_ids, compare_item.external_ids, ext_id
601        )
602        if external_id_match is not None:
603            return external_id_match
604    # compare version
605    if not compare_version(base_item.version, compare_item.version):
606        return False
607    # finally comparing on (exact) name match
608    return compare_strings(base_item.name, compare_item.name, strict=strict)
609
610
611def compare_artists(
612    base_items: list[Artist | ItemMapping],
613    compare_items: list[Artist | ItemMapping],
614    any_match: bool = True,
615) -> bool:
616    """Compare two lists of artist and return True if both lists match (exactly)."""
617    if not base_items or not compare_items:
618        return False
619    # match if first artist matches in both lists
620    if compare_artist(base_items[0], compare_items[0]):
621        return True
622    # compare the artist lists
623    matches = 0
624    for base_item in base_items:
625        for compare_item in compare_items:
626            if compare_artist(base_item, compare_item):
627                if any_match:
628                    return True
629                matches += 1
630    return len(base_items) == len(compare_items) == matches
631
632
633def compare_item_ids(
634    base_item: MediaItem | ItemMapping, compare_item: MediaItem | ItemMapping
635) -> bool:
636    """Compare item_id(s) of two media items."""
637    if not base_item.provider or not compare_item.provider:
638        return False
639    if not base_item.item_id or not compare_item.item_id:
640        return False
641    if base_item.provider == compare_item.provider and base_item.item_id == compare_item.item_id:
642        return True
643
644    base_prov_ids = getattr(base_item, "provider_mappings", None)
645    compare_prov_ids = getattr(compare_item, "provider_mappings", None)
646
647    if base_prov_ids is not None:
648        assert isinstance(base_item, MediaItem)  # for type checking
649        for prov_l in base_item.provider_mappings:
650            if (
651                prov_l.provider_instance == compare_item.provider
652                and prov_l.item_id == compare_item.item_id
653            ):
654                return True
655
656    if compare_prov_ids is not None:
657        assert isinstance(compare_item, MediaItem)  # for type checking
658        for prov_r in compare_item.provider_mappings:
659            if (
660                prov_r.provider_instance == base_item.provider
661                and prov_r.item_id == base_item.item_id
662            ):
663                return True
664
665    if base_prov_ids is not None and compare_prov_ids is not None:
666        assert isinstance(base_item, MediaItem)  # for type checking
667        assert isinstance(compare_item, MediaItem)  # for type checking
668        for prov_l in base_item.provider_mappings:
669            for prov_r in compare_item.provider_mappings:
670                if prov_l.provider_domain != prov_r.provider_domain:
671                    continue
672                if (
673                    prov_l.is_unique or prov_r.is_unique
674                ) and prov_l.provider_instance != prov_r.provider_instance:
675                    continue
676                if prov_l.item_id == prov_r.item_id:
677                    return True
678    return False
679
680
681def compare_external_ids(
682    external_ids_base: set[tuple[ExternalID, str]],
683    external_ids_compare: set[tuple[ExternalID, str]],
684    external_id_type: ExternalID,
685) -> bool | None:
686    """Compare external ids and return True if a match was found."""
687    base_ids = {
688        normalize_external_id(external_id_type, value)
689        for current_type, value in external_ids_base
690        if current_type == external_id_type
691    }
692    if not base_ids:
693        # return early if the requested external id type is not present in the base set
694        return None
695    compare_ids = {
696        normalize_external_id(external_id_type, value)
697        for current_type, value in external_ids_compare
698        if current_type == external_id_type
699    }
700    if not compare_ids:
701        # return early if the requested external id type is not present in the compare set
702        return None
703    if base_ids.intersection(compare_ids):
704        return True
705    if external_id_type.is_unique:
706        return False
707    return None
708
709
710def loose_compare_strings(base: str, alt: str) -> bool:
711    """Compare strings and return True even on partial match."""
712    # this is used to display 'versions' of the same track/album
713    # where we account for other spelling or some additional wording in the title
714    if len(base) <= 3 or len(alt) <= 3:
715        return compare_strings(base, alt, True)
716    word_count = len(base.strip().split(" "))
717    if word_count == 1 and len(base) < 10:
718        return compare_strings(base, alt, False)
719    base_comp = create_safe_string(base)
720    alt_comp = create_safe_string(alt)
721    if base_comp in alt_comp:
722        return True
723    return alt_comp in base_comp
724
725
726def compare_strings(str1: str, str2: str, strict: bool = True) -> bool:
727    """Compare strings and return True if we have an (almost) perfect match."""
728    if not str1 or not str2:
729        return False
730    str1_lower = str1.lower()
731    str2_lower = str2.lower()
732    if strict:
733        # fall back to the same normalization the (search_name) candidate lookup uses,
734        # so an item that selection surfaces is never rejected here on formatting alone
735        return str1_lower == str2_lower or _compare_safe_strings(str1, str2)
736    # return early if total length mismatch
737    if abs(len(str1) - len(str2)) > 4:
738        return False
739    # handle '&' vs 'And'
740    if " & " in str1_lower and " and " in str2_lower:
741        str2 = str2_lower.replace(" and ", " & ")
742    elif " and " in str1_lower and " & " in str2:
743        str2 = str2_lower.replace(" & ", " and ")
744    if create_safe_string(str1) == create_safe_string(str2):
745        return True
746    # last resort: use difflib to compare strings
747    required_accuracy = 0.9 if (len(str1) + len(str2)) > 18 else 0.8
748    return SequenceMatcher(a=str1_lower, b=str2_lower).ratio() > required_accuracy
749
750
751def compare_version(base_version: str, compare_version: str) -> bool:
752    """Compare version string."""
753    return _normalize_version_tokens(base_version) == _normalize_version_tokens(compare_version)
754
755
756def compare_album_name(base_name: str, compare_name: str) -> bool:
757    """Return True if two album titles are the same identity, ignoring formatting drift."""
758    base_suffix = _album_retail_suffix(base_name)
759    compare_suffix = _album_retail_suffix(compare_name)
760    if base_suffix and compare_suffix and base_suffix != compare_suffix:
761        # both titles name their format and they disagree: an EP is not the single of
762        # the same name, however much of the title the two share
763        return False
764    return compare_strings(
765        strip_album_retail_suffix(base_name), strip_album_retail_suffix(compare_name)
766    )
767
768
769def strip_album_retail_suffix(name: str) -> str:
770    """Return an album title without its retail suffix ("Foo - EP" -> "Foo")."""
771    # the suffix carries no identity information: Apple Music appends it to EP/single
772    # titles while already setting album_type
773    return _ALBUM_SUFFIX_PATTERN.sub("", name)
774
775
776def album_retail_suffix_sql_match(name_column: str, suffix_key: str) -> str:
777    """
778    Return a SQL condition that holds when a raw album title spells out a retail suffix.
779
780    :param name_column: SQL expression yielding the raw album title.
781    :param suffix_key: One of :data:`ALBUM_RETAIL_SUFFIX_KEYS`.
782    """
783    # any non-alphanumeric in front of the word sets it off, so an ordinary title that
784    # merely ends in those letters ("Step", "Singles") is left alone. Trailing brackets are
785    # trimmed first, which lets one condition cover every separator a provider may use.
786    # Deliberately looser than the pattern above, as this only selects the pairs the album
787    # comparison is then held to
788    return f"upper(rtrim({name_column}, ' )]')) GLOB '*[^A-Z0-9]{suffix_key.upper()}'"
789
790
791def compare_explicit(base: MediaItemMetadata, compare: MediaItemMetadata) -> bool | None:
792    """Compare if explicit is same in metadata."""
793    if base.explicit is not None and compare.explicit is not None:
794        # explicitness info is not always present in metadata
795        # only strict compare them if both have the info set
796        return base.explicit == compare.explicit
797    return None
798
799
800@lru_cache(maxsize=1024)
801def _normalize_version_tokens(value: str) -> tuple[str, ...]:
802    """Return meaningful, deduplicated version tokens in stable order."""
803    if not value:
804        return ()
805    stripped_value = value.casefold()
806    for pattern in _IGNORE_VERSION_PATTERNS:
807        stripped_value = pattern.sub(" ", stripped_value)
808    tokens = (
809        _VERSION_WORD_ALIASES.get(token, token) for token in re.findall(r"[^\W_]+", stripped_value)
810    )
811    return tuple(sorted({token for token in tokens if token not in _VERSION_IGNORE_WORDS}))
812
813
814def _album_retail_suffix(name: str) -> str:
815    """Return the retail suffix an album title spells out, or an empty string."""
816    match = _ALBUM_SUFFIX_PATTERN.search(name)
817    if not match:
818        return ""
819    return (match.group("suffix") or match.group("bracketed")).casefold()
820
821
822def _is_dynamic_radio(item: Radio | ItemMapping) -> bool:
823    """Return True if the item is a dynamic radio station."""
824    return isinstance(item, Radio) and item.is_dynamic
825
826
827def _compare_album_version(base_version: str, compare_version: str) -> AlbumMatchEvidence:
828    """Return match evidence for an album version/edition comparison."""
829    base_tokens = set(_normalize_version_tokens(base_version))
830    compare_tokens = set(_normalize_version_tokens(compare_version))
831    if base_tokens == compare_tokens:
832        return AlbumMatchEvidence.MATCH
833    # a recording-changing qualifier (live, karaoke, remix, ...) makes an otherwise
834    # unequal pair of editions unsafe to merge, wherever it appears in either wording,
835    # not only when it is the token that happens to differ between the two, and even
836    # when the other side omits version metadata entirely
837    if (base_tokens | compare_tokens) & _RECORDING_CONFLICT_VERSION_TOKENS:
838        return AlbumMatchEvidence.NO_MATCH
839    if not base_tokens or not compare_tokens:
840        # a provider commonly omits edition metadata entirely (e.g. a remaster tagged
841        # without a version string), so a blank version next to a real one is
842        # undecided rather than a proven conflict: let a tracklist resolve it
843        return AlbumMatchEvidence.INSUFFICIENT
844    if base_tokens < compare_tokens or compare_tokens < base_tokens:
845        # one version's wording is a strict subset of the other's (e.g. "2022 Remaster"
846        # vs. "Deluxe 2022 Remaster"): an ambiguous packaging difference a tracklist can resolve
847        return AlbumMatchEvidence.INSUFFICIENT
848    return AlbumMatchEvidence.NO_MATCH
849
850
851def _compare_safe_strings(base: str, compare: str) -> bool:
852    """Return True if two names are equal ignoring case, diacritics, punctuation and spacing."""
853    base_safe = _normalize_name(base)
854    compare_safe = _normalize_name(compare)
855    if base_safe and compare_safe:
856        return base_safe == compare_safe
857    if base_safe or compare_safe:
858        return False
859    # both names collapse to nothing under normalization (e.g. the band "!!!"): fall back
860    # to a raw comparison with all whitespace removed, so spacing drift ("( )" vs "()")
861    # still matches while unrelated symbol-only names don't
862    return "".join(base.split()).casefold() == "".join(compare.split()).casefold()
863
864
865@lru_cache(maxsize=1024)
866def _normalize_name(name: str) -> str:
867    """Return a punctuation/diacritic/whitespace-insensitive name for identity checks."""
868    core = create_safe_string(name, True, True)
869    if not core:
870        # a name made up entirely of symbols is decided on its complete raw spelling
871        return core
872    stripped = name.strip()
873    # a symbol bordering the title belongs to it ("MOTOMAMI +"), however it is spaced,
874    # while punctuation and symbols between words are drift two spellings may differ on
875    return f"{_edge_symbols(stripped)}{core}{_edge_symbols(stripped[::-1])[::-1]}"
876
877
878def _edge_symbols(name: str) -> str:
879    """Return the run of identity-bearing symbols at the start of a title."""
880    # only a mathematical symbol is a title's own wording (Ed Sheeran's operators);
881    # currency and modifier symbols stand in for letters ("bbno$", a backtick for an
882    # apostrophe), which normalization folds away like the punctuation they replace
883    for index, char in enumerate(name):
884        # a symbol anyascii spells out (∂ -> d) already sits in the normalized name
885        if unicodedata.category(char) != "Sm" or create_safe_string(char, True, True):
886            return name[:index].casefold()
887    return name.casefold()
888
889
890def _track_positions(tracks: Sequence[Track]) -> dict[tuple[int, int], Track]:
891    """Return tracks keyed by their (disc_number, track_number) position."""
892    if len({bool(track.disc_number) for track in tracks}) > 1:
893        # some tracks report a disc number and others don't: the shape can't be trusted
894        return {}
895    positions: dict[tuple[int, int], Track] = {}
896    for track in tracks:
897        if not track.track_number:
898            return {}
899        key = (track.disc_number or 1, track.track_number)
900        if key in positions:
901            # duplicate position: the tracklist shape cannot be trusted
902            return {}
903        positions[key] = track
904    return positions
905
906
907def _has_unknown_disc_layout(tracks: Sequence[Track]) -> bool:
908    """Return True if a tracklist reports no disc number at all (an assumed single disc)."""
909    return all(not track.disc_number for track in tracks)
910
911
912def _compare_track_fingerprint(base_track: Track, compare_track: Track) -> AlbumMatchEvidence:
913    """Return match evidence for a single album-track position."""
914    base_isrcs = _track_isrcs(base_track)
915    compare_isrcs = _track_isrcs(compare_track)
916    if base_isrcs and compare_isrcs:
917        if base_isrcs.isdisjoint(compare_isrcs):
918            # both sides tagged an ISRC and they disagree: a different recording/remaster
919            return AlbumMatchEvidence.NO_MATCH
920        if not base_track.duration or not compare_track.duration:
921            return AlbumMatchEvidence.INSUFFICIENT
922        if _duration_close(base_track.duration, compare_track.duration, _ISRC_DURATION_TOLERANCE):
923            return AlbumMatchEvidence.MATCH
924        return AlbumMatchEvidence.INSUFFICIENT
925
926    # no usable ISRC on (at least) one side: fall back to title/version + duration
927    if not base_track.name or not compare_track.name:
928        return AlbumMatchEvidence.INSUFFICIENT
929    if not compare_strings(base_track.name, compare_track.name, strict=True):
930        return AlbumMatchEvidence.NO_MATCH
931    if not compare_version(base_track.version, compare_track.version):
932        return AlbumMatchEvidence.NO_MATCH
933    if not base_track.duration or not compare_track.duration:
934        return AlbumMatchEvidence.INSUFFICIENT
935    if _duration_close(base_track.duration, compare_track.duration, _FALLBACK_DURATION_TOLERANCE):
936        return AlbumMatchEvidence.MATCH
937    return AlbumMatchEvidence.NO_MATCH
938
939
940def _track_isrcs(track: Track) -> set[str]:
941    """Return the structurally valid, normalized ISRCs tagged on a track."""
942    return {
943        normalize_external_id(ExternalID.ISRC, value)
944        for current_type, value in track.external_ids
945        if current_type == ExternalID.ISRC and is_valid_isrc(value)
946    }
947
948
949def _duration_close(base_duration: int, compare_duration: int, tolerance: int) -> bool:
950    """Return True if two track durations (in seconds) are within tolerance."""
951    return abs(base_duration - compare_duration) <= tolerance
952
953
954def _finalize_album_evidence(
955    ambiguous: bool,
956    base_tracks: Sequence[Track] | None,
957    compare_tracks: Sequence[Track] | None,
958) -> AlbumMatchEvidence:
959    """Combine an album's metadata ambiguity with an optional track fingerprint override."""
960    fingerprint_evidence = compare_album_track_fingerprint(base_tracks, compare_tracks)
961    if fingerprint_evidence == AlbumMatchEvidence.NO_MATCH:
962        # a conflicting tracklist is decisive even if the album's own metadata looked fine
963        return AlbumMatchEvidence.NO_MATCH
964    if not ambiguous:
965        return AlbumMatchEvidence.MATCH
966    return fingerprint_evidence
967