/
/
/
1"""Helpers for normalizing and looking up external media identifiers."""
2
3from __future__ import annotations
4
5import re
6from collections.abc import Iterable
7from uuid import UUID
8
9from music_assistant_models.enums import ExternalID
10
11_GTIN_LENGTHS = (8, 12, 13, 14, 15)
12_ISRC_PATTERN = re.compile(r"^[A-Z]{2}[A-Z0-9]{3}\d{7}$")
13_MUSICBRAINZ_PREFIX = "musicbrainz_"
14
15_EXTERNAL_ID_PRIORITY = {
16 ExternalID.MB_RECORDING: 0,
17 ExternalID.MB_TRACK: 1,
18 ExternalID.MB_ALBUM: 2,
19 ExternalID.MB_ARTIST: 3,
20 ExternalID.DISCOGS: 4,
21 ExternalID.TADB: 5,
22 ExternalID.ACOUSTID: 6,
23 ExternalID.MB_RELEASEGROUP: 7,
24 ExternalID.ASIN: 20,
25 ExternalID.BARCODE: 21,
26 ExternalID.ISRC: 22,
27}
28
29
30def normalize_external_id(external_id_type: ExternalID, value: str) -> str:
31 """
32 Return the canonical representation of an external media identifier.
33
34 Invalid identifiers are returned stripped but otherwise unchanged so provider data
35 remains available for exact matching.
36
37 :param external_id_type: Type of the external identifier.
38 :param value: Identifier value supplied by a provider or file tag.
39 """
40 value = value.strip()
41 if external_id_type == ExternalID.BARCODE:
42 return _normalize_barcode(value)
43 if external_id_type == ExternalID.ISRC:
44 return _normalize_isrc(value)
45 if external_id_type.value.startswith(_MUSICBRAINZ_PREFIX):
46 return _normalize_mbid(value)
47 return value
48
49
50def normalize_external_ids(
51 external_ids: Iterable[tuple[ExternalID, str]],
52) -> set[tuple[ExternalID, str]]:
53 """
54 Return external identifiers with canonical values.
55
56 :param external_ids: External identifier type/value pairs.
57 """
58 return {
59 (external_id_type, normalize_external_id(external_id_type, value))
60 for external_id_type, value in external_ids
61 if value.strip()
62 }
63
64
65def external_id_lookup_values(external_id_type: ExternalID, value: str) -> tuple[str, ...]:
66 """
67 Return indexed lookup values compatible with current and legacy storage formats.
68
69 :param external_id_type: Type of the external identifier.
70 :param value: Identifier value to look up.
71 """
72 raw_value = value.strip()
73 normalized_value = normalize_external_id(external_id_type, raw_value)
74 values = {raw_value, normalized_value}
75
76 if external_id_type == ExternalID.BARCODE and _is_valid_gtin(normalized_value):
77 payload = normalized_value.lstrip("0")
78 for length in _GTIN_LENGTHS:
79 if len(payload) <= length:
80 values.add(payload.zfill(length))
81 elif external_id_type == ExternalID.ISRC and _ISRC_PATTERN.fullmatch(normalized_value):
82 values.add(
83 f"{normalized_value[:2]}-{normalized_value[2:5]}-"
84 f"{normalized_value[5:7]}-{normalized_value[7:]}"
85 )
86 elif external_id_type.value.startswith(_MUSICBRAINZ_PREFIX):
87 values.add(f"{{{normalized_value}}}")
88
89 return tuple(sorted(values))
90
91
92def is_valid_isrc(value: str) -> bool:
93 """
94 Return whether a value is a structurally valid (canonical) ISRC.
95
96 An invalid ISRC is still preserved as-is for storage/exact-match lookups, but
97 should be treated as absent by callers that rely on the ISRC being a genuine,
98 globally unique recording identifier (e.g. album/track fingerprint comparisons).
99
100 :param value: Identifier value to validate.
101 """
102 return bool(_ISRC_PATTERN.fullmatch(normalize_external_id(ExternalID.ISRC, value)))
103
104
105def is_valid_barcode(value: str) -> bool:
106 """
107 Return whether a value is a structurally valid (canonical) barcode/GTIN.
108
109 An invalid barcode is still preserved as-is for storage/exact-match lookups, but
110 should be treated as absent by callers that rely on the barcode being a genuine
111 GTIN (e.g. a MusicBrainz release lookup).
112
113 :param value: Identifier value to validate.
114 """
115 return _is_valid_gtin(normalize_external_id(ExternalID.BARCODE, value))
116
117
118def external_id_lookup_values_untyped(value: str) -> tuple[str, ...]:
119 """
120 Return lookup values for an identifier whose type is not known.
121
122 :param value: Identifier value to look up.
123 """
124 values = {value.strip()}
125 for external_id_type in ExternalID:
126 values.update(external_id_lookup_values(external_id_type, value))
127 return tuple(sorted(values))
128
129
130def barcode_to_upc(value: str) -> str:
131 """
132 Return the shortest UPC/EAN representation of a valid barcode.
133
134 :param value: Barcode in a provider or canonical storage format.
135 """
136 normalized_value = normalize_external_id(ExternalID.BARCODE, value)
137 if not _is_valid_gtin(normalized_value):
138 return value
139 payload = normalized_value.lstrip("0")
140 return payload.zfill(12)
141
142
143def external_id_sort_key(external_id: tuple[ExternalID, str]) -> tuple[int, str, str]:
144 """
145 Return a deterministic preference key for an external identifier.
146
147 :param external_id: External identifier type/value pair.
148 """
149 external_id_type, value = external_id
150 default_priority = 10 if external_id_type.is_unique else 30
151 return (
152 _EXTERNAL_ID_PRIORITY.get(external_id_type, default_priority),
153 external_id_type.value,
154 normalize_external_id(external_id_type, value),
155 )
156
157
158def _normalize_barcode(value: str) -> str:
159 """Return a valid UPC/EAN/GTIN as GTIN-14."""
160 compact_value = value.replace("-", "").replace(" ", "")
161 if len(compact_value) not in _GTIN_LENGTHS or not compact_value.isdigit():
162 return value
163 payload = compact_value.lstrip("0")
164 if not payload or len(payload) > 14:
165 return value
166 normalized_value = payload.zfill(14)
167 if _is_valid_gtin(normalized_value):
168 return normalized_value
169 if len(compact_value) == 13:
170 # Qobuz delivers part of its catalogue as a GTIN-14 with the final check digit
171 # chopped off: recover the full value by appending the recomputed check digit
172 return compact_value + _gtin_check_digit(compact_value)
173 return value
174
175
176def _normalize_isrc(value: str) -> str:
177 """Return a valid ISRC without separators."""
178 normalized_value = value.replace("-", "").replace(" ", "").upper()
179 return normalized_value if _ISRC_PATTERN.fullmatch(normalized_value) else value
180
181
182def _normalize_mbid(value: str) -> str:
183 """Return a MusicBrainz identifier as a lowercase UUID."""
184 try:
185 return str(UUID(value.strip("{}")))
186 except ValueError, AttributeError:
187 return value
188
189
190def _is_valid_gtin(value: str) -> bool:
191 """Return whether a value is a canonical GTIN-14 with a valid check digit."""
192 if len(value) != 14 or not value.isdigit():
193 return False
194 return value[-1] == _gtin_check_digit(value[:-1])
195
196
197def _gtin_check_digit(body: str) -> str:
198 """Return the GTIN check digit for a numeric identifier body."""
199 checksum = sum(
200 int(char) * (3 if index % 2 == 0 else 1) for index, char in enumerate(reversed(body))
201 )
202 return str((10 - checksum % 10) % 10)
203