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