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