/
/
/
1"""
2CUE sheet integration for the filesystem_local provider.
3
4A CUE sheet describes multiple logical tracks within a single audio file
5(typically a whole-album rip). This module provides:
6
7- Synthetic ``item_id`` construction/parsing for CUE-derived tracks.
8- A :class:`CueSheetHandler` that turns a CUE sheet + its audio file into
9 :class:`Track` objects, produces :class:`StreamDetails` for playback of a
10 single logical track, and streams the requested segment via FFmpeg.
11"""
12
13from __future__ import annotations
14
15import os
16from collections.abc import AsyncGenerator
17from dataclasses import asdict, dataclass
18from datetime import UTC, datetime
19from typing import TYPE_CHECKING, Any
20
21from music_assistant_models.enums import (
22 ContentType,
23 ExternalID,
24 ImageType,
25 MediaType,
26 StreamType,
27)
28from music_assistant_models.errors import InvalidDataError, MediaNotFoundError
29from music_assistant_models.media_items import (
30 Album,
31 Artist,
32 AudioFormat,
33 ItemMapping,
34 MediaItemImage,
35 ProviderMapping,
36 Track,
37 UniqueList,
38)
39from music_assistant_models.streamdetails import StreamDetails
40
41from music_assistant.constants import UNKNOWN_ARTIST, UNKNOWN_ARTIST_ID_MBID
42from music_assistant.helpers.cue_sheet import CueSheet, CueTrack, parse_cue_sheet
43from music_assistant.helpers.ffmpeg import get_ffmpeg_stream
44from music_assistant.helpers.tags import AudioTags, async_parse_tags, clean_mbid
45from music_assistant.helpers.util import detect_charset
46
47from .constants import CACHE_CATEGORY_CUE_SHEETS, TRACK_EXTENSIONS
48from .helpers import FileSystemItem
49
50if TYPE_CHECKING:
51 from . import LocalFileSystemProvider
52
53CUE_TRACK_ID_DELIMITER = "::track"
54
55# Bump when CUE decoding or parsing changes so cached and library metadata are refreshed.
56_CUE_METADATA_VERSION = 1
57
58
59@dataclass(frozen=True, slots=True)
60class _TrackBuildContext:
61 """Shared state used to build each Track from a CUE sheet."""
62
63 audio_format: AudioFormat
64 disc_number: int
65 date_added: datetime | None
66 embedded_image: MediaItemImage | None
67 track_genres: set[str] | None # fallback genres from the audio file
68 album: Album | None
69 album_performers: tuple[str, ...] # sheet-level PERFORMER values, fallback for track artists
70
71
72def cue_metadata_checksum(file_checksum: str | None) -> str:
73 """
74 Return the checksum for metadata derived from a CUE sheet.
75
76 :param file_checksum: The checksum of the source CUE file.
77 """
78 return f"{_CUE_METADATA_VERSION}:{file_checksum}"
79
80
81def make_cue_track_id(cue_relative_path: str, track_number: int) -> str:
82 """Build the synthetic provider item_id for a CUE-derived track."""
83 return f"{cue_relative_path}{CUE_TRACK_ID_DELIMITER}{track_number:02d}"
84
85
86def cue_referenced_audio_stem(cue_item: FileSystemItem, cue_sheet: CueSheet) -> str | None:
87 """
88 Return the absolute path (without extension) of the CUE's referenced audio file.
89
90 Returns None if the CUE names no file. The named file may differ from the CUE's
91 own filename.
92
93 :param cue_item: The CUE file's FileSystemItem.
94 :param cue_sheet: The parsed CUE sheet.
95 """
96 if not cue_sheet.file_path:
97 return None
98 companion = os.path.join(os.path.dirname(cue_item.absolute_path), cue_sheet.file_path)
99 return companion.rsplit(".", 1)[0]
100
101
102def parse_cue_track_id(item_id: str) -> tuple[str, int] | None:
103 """Return (cue_relative_path, track_number) if item_id is a CUE track id, else None."""
104 if CUE_TRACK_ID_DELIMITER not in item_id:
105 return None
106 cue_path, track_num_str = item_id.rsplit(CUE_TRACK_ID_DELIMITER, 1)
107 if not cue_path:
108 return None
109 try:
110 track_num = int(track_num_str)
111 except ValueError:
112 return None
113 return cue_path, track_num
114
115
116def _cue_sheet_from_dict(data: dict[str, Any]) -> CueSheet:
117 """Rebuild a :class:`CueSheet` from its ``asdict`` representation."""
118 tracks = [CueTrack(**track_data) for track_data in data.get("tracks", [])]
119 return CueSheet(**{**data, "tracks": tracks})
120
121
122class CueSheetHandler:
123 """CUE sheet integration bound to a :class:`LocalFileSystemProvider` instance."""
124
125 def __init__(self, provider: LocalFileSystemProvider) -> None:
126 """
127 Initialize the handler.
128
129 :param provider: The provider that owns this handler; used for shared state
130 (cache, logger, mass) and helpers (``_parse_album``, ``_parse_artist``).
131 """
132 self.provider = provider
133
134 async def read_cue_file(self, cue_item: FileSystemItem) -> str:
135 """
136 Read CUE file content, decoded with the detected charset.
137
138 :param cue_item: The CUE file's FileSystemItem.
139 """
140 # route through provider._read_file so non-mounted providers (WebDAV)
141 # use their own transport instead of aiofiles
142 raw = await self.provider._read_file(cue_item.relative_path)
143 encoding = await detect_charset(raw)
144 return raw.decode(encoding, errors="replace")
145
146 async def load_cue_sheet(self, cue_item: FileSystemItem) -> CueSheet:
147 """
148 Return the parsed CUE sheet for the given file.
149
150 :param cue_item: The CUE file's FileSystemItem.
151 """
152 # cached by (path, checksum) so unchanged CUE files skip the file read
153 provider = self.provider
154 metadata_checksum = cue_metadata_checksum(cue_item.checksum)
155 cached = await provider.mass.cache.get(
156 key=cue_item.relative_path,
157 provider=provider.instance_id,
158 category=CACHE_CATEGORY_CUE_SHEETS,
159 checksum=metadata_checksum,
160 default=None,
161 )
162 if cached is not None:
163 return _cue_sheet_from_dict(cached)
164 content = await self.read_cue_file(cue_item)
165 sheet = parse_cue_sheet(content)
166 await provider.mass.cache.set(
167 key=cue_item.relative_path,
168 data=asdict(sheet),
169 provider=provider.instance_id,
170 category=CACHE_CATEGORY_CUE_SHEETS,
171 checksum=metadata_checksum,
172 expiration=3600 * 24 * 365,
173 )
174 return sheet
175
176 async def find_audio_file(self, cue_item: FileSystemItem, cue_sheet: CueSheet) -> str | None:
177 """
178 Locate the audio file referenced by a CUE sheet.
179
180 Returns the provider-relative path of the audio file, or ``None`` if it
181 cannot be located. Routing through ``provider.exists`` keeps this working
182 for every filesystem provider (local, SMB/NFS mounts, WebDAV).
183
184 :param cue_item: The CUE file's FileSystemItem.
185 :param cue_sheet: The parsed CUE sheet data.
186 """
187 cue_dir = os.path.dirname(cue_item.relative_path)
188
189 def _join(name: str) -> str:
190 return os.path.join(cue_dir, name) if cue_dir else name
191
192 # 1. try the filename from the CUE FILE command
193 if cue_sheet.file_path:
194 candidate = _join(cue_sheet.file_path)
195 if await self.provider.exists(candidate):
196 return candidate
197
198 # 2. same-name matching: album.cue -> album.{flac,mp3,...}
199 cue_stem = cue_item.filename.rsplit(".", 1)[0]
200 for ext in TRACK_EXTENSIONS:
201 candidate = _join(f"{cue_stem}.{ext}")
202 if await self.provider.exists(candidate):
203 return candidate
204
205 return None
206
207 async def parse_tracks(self, cue_item: FileSystemItem) -> list[Track]:
208 """
209 Parse a CUE sheet and return individual :class:`Track` objects.
210
211 :param cue_item: The CUE file's FileSystemItem.
212 """
213 with self.provider._ondemand_listing_scope():
214 return await self._parse_tracks_impl(cue_item)
215
216 async def get_stream_details(self, item_id: str) -> StreamDetails:
217 """
218 Return the streamdetails for a CUE-sheet-derived track.
219
220 :param item_id: Track ID in format "path/to/file.cue::trackNN".
221 """
222 parsed = parse_cue_track_id(item_id)
223 if parsed is None:
224 msg = f"Invalid CUE track id: {item_id}"
225 raise InvalidDataError(msg)
226 cue_path, track_number = parsed
227
228 # audio format + duration were persisted at sync time; reuse them here
229 provider = self.provider
230 library_track = await provider.mass.music.tracks.get_library_item_by_prov_id(
231 item_id, provider.instance_id
232 )
233 if library_track is None:
234 msg = f"CUE track not in library: {item_id}"
235 raise MediaNotFoundError(msg)
236 prov_mapping = next(x for x in library_track.provider_mappings if x.item_id == item_id)
237 original_format = prov_mapping.audio_format
238
239 # re-parse to read the track's start offset
240 cue_item = await provider.resolve(cue_path)
241 cue_sheet = await self.load_cue_sheet(cue_item)
242 cue_track = next((t for t in cue_sheet.tracks if t.number == track_number), None)
243 if cue_track is None:
244 msg = f"Track {track_number} not found in CUE sheet: {cue_path}"
245 raise MediaNotFoundError(msg)
246
247 audio_relative_path = await self.find_audio_file(cue_item, cue_sheet)
248 if audio_relative_path is None:
249 msg = f"Audio file not found for CUE sheet: {cue_path}"
250 raise MediaNotFoundError(msg)
251
252 # CUE tracks need StreamType.CUSTOM: they are a segment of a larger file and
253 # require -ss/-t at the track offset. Core appends its own -ss for user seeks
254 # after extra_input_args, and a second input -ss overrides the first, so
255 # LOCAL_FILE with a base offset cannot coexist with user seeking.
256 output_format = AudioFormat(
257 content_type=ContentType.PCM_F32LE,
258 sample_rate=original_format.sample_rate,
259 bit_depth=32,
260 channels=original_format.channels,
261 )
262 # store the relative path so get_audio_stream re-resolves at stream time;
263 # keeps WebDAV auth fresh and keeps credentials out of persisted StreamDetails
264 return StreamDetails(
265 provider=provider.instance_id,
266 item_id=item_id,
267 audio_format=output_format,
268 media_type=MediaType.TRACK,
269 stream_type=StreamType.CUSTOM,
270 duration=library_track.duration,
271 can_seek=True,
272 allow_seek=True,
273 data={
274 "audio_relative_path": audio_relative_path,
275 "start_seconds": cue_track.start_position,
276 "original_format": original_format.to_dict(),
277 },
278 )
279
280 async def get_audio_stream(
281 self, streamdetails: StreamDetails, seek_position: int = 0
282 ) -> AsyncGenerator[bytes]:
283 """
284 Yield the segment of the underlying audio file for a CUE-derived track.
285
286 :param streamdetails: Streamdetails previously built by :meth:`get_stream_details`.
287 :param seek_position: Position (seconds) within the track to start from.
288 """
289 # streamdetails was built by get_stream_details; asserts narrow for mypy
290 assert streamdetails.data is not None
291 assert streamdetails.duration is not None
292 audio_relative_path: str = streamdetails.data["audio_relative_path"]
293 base_start: float = streamdetails.data["start_seconds"]
294 original_format = AudioFormat.from_dict(streamdetails.data["original_format"])
295
296 # actual seek position within the full audio file
297 actual_seek = base_start + seek_position
298 remaining_duration = streamdetails.duration - seek_position
299 if remaining_duration <= 0:
300 return
301
302 # re-resolve to get a current absolute path (e.g. fresh auth for WebDAV)
303 audio_item = await self.provider.resolve(audio_relative_path)
304 async for chunk in get_ffmpeg_stream(
305 audio_input=audio_item.absolute_path,
306 input_format=original_format,
307 output_format=streamdetails.audio_format,
308 extra_input_args=["-ss", str(actual_seek), "-t", str(remaining_duration)],
309 ):
310 yield chunk
311
312 async def _parse_tracks_impl(self, cue_item: FileSystemItem) -> list[Track]:
313 """Parse a CUE sheet's tracks (implementation, see :meth:`parse_tracks`)."""
314 provider = self.provider
315 logger = provider.logger
316 cue_sheet = await self.load_cue_sheet(cue_item)
317
318 if not cue_sheet.tracks:
319 msg = f"CUE sheet has no tracks: {cue_item.relative_path}"
320 raise InvalidDataError(msg)
321
322 audio_relative_path = await self.find_audio_file(cue_item, cue_sheet)
323 if audio_relative_path is None:
324 msg = f"Audio file not found for CUE sheet: {cue_item.relative_path}"
325 raise MediaNotFoundError(msg)
326
327 audio_item = await provider.resolve(audio_relative_path)
328 tags = await async_parse_tags(audio_item.absolute_path, audio_item.file_size)
329 total_duration = tags.duration or 0.0
330 if total_duration <= 0:
331 msg = f"Could not determine duration for audio file of CUE sheet: {cue_item.relative_path}"
332 raise InvalidDataError(msg)
333
334 self._apply_cue_overrides(tags, cue_sheet)
335
336 album: Album | None = None
337 if tags.album:
338 album = await provider._parse_album(
339 track_path=audio_relative_path,
340 track_tags=tags,
341 track_created_at=cue_item.created_at,
342 # the companion audio file is absorbed into CUE tracks and is never itself a
343 # synced item (its own sync entry is dropped, see sync_library's CUE-companion
344 # filter), so it cannot be re-queued for reparsing; register the CUE sheet's own
345 # path instead, since re-processing that path re-runs this same parse
346 representative_track=cue_item.relative_path,
347 )
348 else:
349 logger.warning(
350 "CUE sheet %s has no TITLE and audio file has no album tag",
351 cue_item.relative_path,
352 )
353
354 # embedded cover art is shared across all CUE tracks from this audio file
355 embedded_image = (
356 MediaItemImage(
357 type=ImageType.THUMB,
358 path=audio_relative_path,
359 provider=provider.instance_id,
360 remotely_accessible=False,
361 )
362 if tags.has_cover_image
363 else None
364 )
365 # if the album lacks its own image, adopt the embedded one
366 if album and embedded_image and not album.image:
367 album.metadata.images = UniqueList([embedded_image])
368
369 ctx = _TrackBuildContext(
370 audio_format=self._audio_format_from_tags(audio_relative_path, tags),
371 # honor audio file's DISCNUMBER (CUE does not carry disc info); defaults to 1
372 disc_number=tags.disc or 1,
373 date_added=(
374 datetime.fromtimestamp(cue_item.created_at, tz=UTC) if cue_item.created_at else None
375 ),
376 embedded_image=embedded_image,
377 track_genres=set(tags.genres) if tags.genres else None,
378 album=album,
379 album_performers=tuple(cue_sheet.performers),
380 )
381
382 sorted_tracks = sorted(cue_sheet.tracks, key=lambda t: t.start_position)
383 tracks: list[Track] = []
384 for i, cue_track in enumerate(sorted_tracks):
385 if i + 1 < len(sorted_tracks):
386 duration = sorted_tracks[i + 1].start_position - cue_track.start_position
387 else:
388 duration = total_duration - cue_track.start_position
389
390 if duration <= 0:
391 logger.warning(
392 "CUE sheet %s track %d has non-positive duration (%.2fs); skipping",
393 cue_item.relative_path,
394 cue_track.number,
395 duration,
396 )
397 continue
398 if not cue_track.title:
399 logger.warning(
400 "CUE sheet %s track %d has no TITLE; skipping",
401 cue_item.relative_path,
402 cue_track.number,
403 )
404 continue
405
406 tracks.append(await self._build_track(cue_track, cue_item, duration, ctx))
407
408 return tracks
409
410 @staticmethod
411 def _audio_format_from_tags(audio_path: str, tags: AudioTags) -> AudioFormat:
412 """
413 Build an AudioFormat for an audio file from its tags.
414
415 :param audio_path: Path to the audio file (used for extension fallback).
416 :param tags: Parsed audio tags.
417 """
418 return AudioFormat(
419 content_type=ContentType.try_parse(audio_path.rsplit(".", 1)[-1] or tags.format),
420 sample_rate=tags.sample_rate,
421 bit_depth=tags.bits_per_sample,
422 channels=tags.channels,
423 bit_rate=tags.bit_rate,
424 )
425
426 @staticmethod
427 def _apply_cue_overrides(tags: AudioTags, cue_sheet: CueSheet) -> None:
428 """Overwrite album-level audio tags with values from the CUE sheet."""
429 if cue_sheet.title:
430 tags.tags["album"] = cue_sheet.title
431 if cue_sheet.sort_title:
432 tags.tags["albumsort"] = cue_sheet.sort_title
433 if cue_sheet.performers:
434 # plural form so AudioTags.album_artists sees every value
435 tags.tags.pop("albumartist", None)
436 tags.tags["albumartists"] = list(cue_sheet.performers)
437 if cue_sheet.album_artist_sort_names:
438 tags.tags["albumartistsort"] = ";".join(cue_sheet.album_artist_sort_names)
439 if cue_sheet.musicbrainz_albumartistids:
440 tags.tags["musicbrainzalbumartistid"] = ";".join(cue_sheet.musicbrainz_albumartistids)
441 if cue_sheet.date:
442 tags.tags["date"] = cue_sheet.date
443 if cue_sheet.genres:
444 # AudioTags.genres splits on ";" so we join multi-line values that way
445 tags.tags["genre"] = ";".join(cue_sheet.genres)
446 if cue_sheet.album_types:
447 # album_type reads "releasetype" as a single string and substring-matches
448 tags.tags["releasetype"] = " ".join(cue_sheet.album_types)
449 if cue_sheet.barcode:
450 tags.tags["barcode"] = cue_sheet.barcode
451 if cue_sheet.musicbrainz_albumid:
452 tags.tags["musicbrainzalbumid"] = cue_sheet.musicbrainz_albumid
453 if cue_sheet.musicbrainz_releasegroupid:
454 tags.tags["musicbrainzreleasegroupid"] = cue_sheet.musicbrainz_releasegroupid
455
456 async def _build_track(
457 self,
458 cue_track: CueTrack,
459 cue_item: FileSystemItem,
460 duration: float,
461 ctx: _TrackBuildContext,
462 ) -> Track:
463 """Construct a single Track from a CUE track entry."""
464 provider = self.provider
465 track_id = make_cue_track_id(cue_item.relative_path, cue_track.number)
466
467 # per-track PERFORMER wins over sheet-level. Multi-artist uses one PERFORMER
468 # line per artist (Vorbis convention); never delimiter-split, would mangle "AC/DC".
469 performer_names = cue_track.performers or list(ctx.album_performers)
470 track_artists: UniqueList[Artist | ItemMapping] = UniqueList()
471 for idx, artist_name in enumerate(performer_names):
472 artist = await provider._parse_artist(
473 name=artist_name,
474 sort_name=(
475 cue_track.artist_sort_names[idx]
476 if idx < len(cue_track.artist_sort_names)
477 else None
478 ),
479 mbid=(
480 cue_track.musicbrainz_artistids[idx]
481 if idx < len(cue_track.musicbrainz_artistids)
482 else None
483 ),
484 # same reasoning as the album parse above: this performer may have their own
485 # artist.nfo/images, and only the CUE sheet's own path can be re-queued later
486 representative_track=cue_item.relative_path,
487 )
488 if artist:
489 track_artists.append(artist)
490 if not track_artists:
491 # neither the track nor the sheet declared a PERFORMER; fall back to
492 # the [unknown] artist rather than leaving the track artist-less
493 unknown = await provider._parse_artist(name=UNKNOWN_ARTIST, mbid=UNKNOWN_ARTIST_ID_MBID)
494 if unknown:
495 track_artists.append(unknown)
496
497 track = Track(
498 item_id=track_id,
499 provider=provider.instance_id,
500 name=cue_track.title or f"Track {cue_track.number}",
501 sort_name=cue_track.sort_name,
502 provider_mappings={
503 ProviderMapping(
504 item_id=track_id,
505 provider_domain=provider.domain,
506 provider_instance=provider.instance_id,
507 audio_format=ctx.audio_format,
508 details=cue_metadata_checksum(cue_item.checksum),
509 in_library=True,
510 )
511 },
512 track_number=cue_track.number,
513 disc_number=ctx.disc_number,
514 duration=round(duration),
515 date_added=ctx.date_added,
516 )
517
518 if track_artists:
519 track.artists = track_artists
520 if ctx.album:
521 track.album = ctx.album
522 for isrc in cue_track.isrcs:
523 track.external_ids.add((ExternalID.ISRC, isrc))
524 if recording_mbid := clean_mbid(cue_track.musicbrainz_recordingid, cue_item.relative_path):
525 # the setter keeps external_ids in sync
526 track.mbid = recording_mbid
527 if cue_track.musicbrainz_releasetrackid:
528 track.external_ids.add((ExternalID.MB_TRACK, cue_track.musicbrainz_releasetrackid))
529 if ctx.embedded_image is not None:
530 track.metadata.images = UniqueList([ctx.embedded_image])
531 if cue_track.genres:
532 # per-track REM GENRE wins over the shared audio-file/album genres
533 track.metadata.genres = set(cue_track.genres)
534 elif ctx.track_genres is not None:
535 track.metadata.genres = ctx.track_genres
536 if cue_track.copyright:
537 track.metadata.copyright = cue_track.copyright
538 if cue_track.grouping:
539 track.metadata.grouping = cue_track.grouping
540 if cue_track.comment:
541 track.metadata.description = cue_track.comment
542 if cue_track.explicit is not None:
543 track.metadata.explicit = cue_track.explicit
544 return track
545