music-assistant-server

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        provider = self.provider
214        logger = provider.logger
215        cue_sheet = await self.load_cue_sheet(cue_item)
216
217        if not cue_sheet.tracks:
218            msg = f"CUE sheet has no tracks: {cue_item.relative_path}"
219            raise InvalidDataError(msg)
220
221        audio_relative_path = await self.find_audio_file(cue_item, cue_sheet)
222        if audio_relative_path is None:
223            msg = f"Audio file not found for CUE sheet: {cue_item.relative_path}"
224            raise MediaNotFoundError(msg)
225
226        audio_item = await provider.resolve(audio_relative_path)
227        tags = await async_parse_tags(audio_item.absolute_path, audio_item.file_size)
228        total_duration = tags.duration or 0.0
229        if total_duration <= 0:
230            msg = f"Could not determine duration for audio file of CUE sheet: {cue_item.relative_path}"
231            raise InvalidDataError(msg)
232
233        self._apply_cue_overrides(tags, cue_sheet)
234
235        album: Album | None = None
236        if tags.album:
237            album = await provider._parse_album(
238                track_path=audio_relative_path,
239                track_tags=tags,
240                track_created_at=cue_item.created_at,
241            )
242        else:
243            logger.warning(
244                "CUE sheet %s has no TITLE and audio file has no album tag",
245                cue_item.relative_path,
246            )
247
248        # embedded cover art is shared across all CUE tracks from this audio file
249        embedded_image = (
250            MediaItemImage(
251                type=ImageType.THUMB,
252                path=audio_relative_path,
253                provider=provider.instance_id,
254                remotely_accessible=False,
255            )
256            if tags.has_cover_image
257            else None
258        )
259        # if the album lacks its own image, adopt the embedded one
260        if album and embedded_image and not album.image:
261            album.metadata.images = UniqueList([embedded_image])
262
263        ctx = _TrackBuildContext(
264            audio_format=self._audio_format_from_tags(audio_relative_path, tags),
265            # honor audio file's DISCNUMBER (CUE does not carry disc info); defaults to 1
266            disc_number=tags.disc or 1,
267            date_added=(
268                datetime.fromtimestamp(cue_item.created_at, tz=UTC) if cue_item.created_at else None
269            ),
270            embedded_image=embedded_image,
271            track_genres=set(tags.genres) if tags.genres else None,
272            album=album,
273            album_performers=tuple(cue_sheet.performers),
274        )
275
276        sorted_tracks = sorted(cue_sheet.tracks, key=lambda t: t.start_position)
277        tracks: list[Track] = []
278        for i, cue_track in enumerate(sorted_tracks):
279            if i + 1 < len(sorted_tracks):
280                duration = sorted_tracks[i + 1].start_position - cue_track.start_position
281            else:
282                duration = total_duration - cue_track.start_position
283
284            if duration <= 0:
285                logger.warning(
286                    "CUE sheet %s track %d has non-positive duration (%.2fs); skipping",
287                    cue_item.relative_path,
288                    cue_track.number,
289                    duration,
290                )
291                continue
292            if not cue_track.title:
293                logger.warning(
294                    "CUE sheet %s track %d has no TITLE; skipping",
295                    cue_item.relative_path,
296                    cue_track.number,
297                )
298                continue
299
300            tracks.append(await self._build_track(cue_track, cue_item, duration, ctx))
301
302        return tracks
303
304    async def get_stream_details(self, item_id: str) -> StreamDetails:
305        """
306        Return the streamdetails for a CUE-sheet-derived track.
307
308        :param item_id: Track ID in format "path/to/file.cue::trackNN".
309        """
310        parsed = parse_cue_track_id(item_id)
311        if parsed is None:
312            msg = f"Invalid CUE track id: {item_id}"
313            raise InvalidDataError(msg)
314        cue_path, track_number = parsed
315
316        # audio format + duration were persisted at sync time; reuse them here
317        provider = self.provider
318        library_track = await provider.mass.music.tracks.get_library_item_by_prov_id(
319            item_id, provider.instance_id
320        )
321        if library_track is None:
322            msg = f"CUE track not in library: {item_id}"
323            raise MediaNotFoundError(msg)
324        prov_mapping = next(x for x in library_track.provider_mappings if x.item_id == item_id)
325        original_format = prov_mapping.audio_format
326
327        # re-parse to read the track's start offset
328        cue_item = await provider.resolve(cue_path)
329        cue_sheet = await self.load_cue_sheet(cue_item)
330        cue_track = next((t for t in cue_sheet.tracks if t.number == track_number), None)
331        if cue_track is None:
332            msg = f"Track {track_number} not found in CUE sheet: {cue_path}"
333            raise MediaNotFoundError(msg)
334
335        audio_relative_path = await self.find_audio_file(cue_item, cue_sheet)
336        if audio_relative_path is None:
337            msg = f"Audio file not found for CUE sheet: {cue_path}"
338            raise MediaNotFoundError(msg)
339
340        # CUE tracks need StreamType.CUSTOM: they are a segment of a larger file and
341        # require -ss/-t at the track offset. Core appends its own -ss for user seeks
342        # after extra_input_args, and a second input -ss overrides the first, so
343        # LOCAL_FILE with a base offset cannot coexist with user seeking.
344        output_format = AudioFormat(
345            content_type=ContentType.PCM_F32LE,
346            sample_rate=original_format.sample_rate,
347            bit_depth=32,
348            channels=original_format.channels,
349        )
350        # store the relative path so get_audio_stream re-resolves at stream time;
351        # keeps WebDAV auth fresh and keeps credentials out of persisted StreamDetails
352        return StreamDetails(
353            provider=provider.instance_id,
354            item_id=item_id,
355            audio_format=output_format,
356            media_type=MediaType.TRACK,
357            stream_type=StreamType.CUSTOM,
358            duration=library_track.duration,
359            can_seek=True,
360            allow_seek=True,
361            data={
362                "audio_relative_path": audio_relative_path,
363                "start_seconds": cue_track.start_position,
364                "original_format": original_format.to_dict(),
365            },
366        )
367
368    async def get_audio_stream(
369        self, streamdetails: StreamDetails, seek_position: int = 0
370    ) -> AsyncGenerator[bytes]:
371        """
372        Yield the segment of the underlying audio file for a CUE-derived track.
373
374        :param streamdetails: Streamdetails previously built by :meth:`get_stream_details`.
375        :param seek_position: Position (seconds) within the track to start from.
376        """
377        # streamdetails was built by get_stream_details; asserts narrow for mypy
378        assert streamdetails.data is not None
379        assert streamdetails.duration is not None
380        audio_relative_path: str = streamdetails.data["audio_relative_path"]
381        base_start: float = streamdetails.data["start_seconds"]
382        original_format = AudioFormat.from_dict(streamdetails.data["original_format"])
383
384        # actual seek position within the full audio file
385        actual_seek = base_start + seek_position
386        remaining_duration = streamdetails.duration - seek_position
387        if remaining_duration <= 0:
388            return
389
390        # re-resolve to get a current absolute path (e.g. fresh auth for WebDAV)
391        audio_item = await self.provider.resolve(audio_relative_path)
392        async for chunk in get_ffmpeg_stream(
393            audio_input=audio_item.absolute_path,
394            input_format=original_format,
395            output_format=streamdetails.audio_format,
396            extra_input_args=["-ss", str(actual_seek), "-t", str(remaining_duration)],
397        ):
398            yield chunk
399
400    @staticmethod
401    def _audio_format_from_tags(audio_path: str, tags: AudioTags) -> AudioFormat:
402        """
403        Build an AudioFormat for an audio file from its tags.
404
405        :param audio_path: Path to the audio file (used for extension fallback).
406        :param tags: Parsed audio tags.
407        """
408        return AudioFormat(
409            content_type=ContentType.try_parse(audio_path.rsplit(".", 1)[-1] or tags.format),
410            sample_rate=tags.sample_rate,
411            bit_depth=tags.bits_per_sample,
412            channels=tags.channels,
413            bit_rate=tags.bit_rate,
414        )
415
416    @staticmethod
417    def _apply_cue_overrides(tags: AudioTags, cue_sheet: CueSheet) -> None:
418        """Overwrite album-level audio tags with values from the CUE sheet."""
419        if cue_sheet.title:
420            tags.tags["album"] = cue_sheet.title
421        if cue_sheet.sort_title:
422            tags.tags["albumsort"] = cue_sheet.sort_title
423        if cue_sheet.performers:
424            # plural form so AudioTags.album_artists sees every value
425            tags.tags.pop("albumartist", None)
426            tags.tags["albumartists"] = list(cue_sheet.performers)
427        if cue_sheet.album_artist_sort_names:
428            tags.tags["albumartistsort"] = ";".join(cue_sheet.album_artist_sort_names)
429        if cue_sheet.musicbrainz_albumartistids:
430            tags.tags["musicbrainzalbumartistid"] = ";".join(cue_sheet.musicbrainz_albumartistids)
431        if cue_sheet.date:
432            tags.tags["date"] = cue_sheet.date
433        if cue_sheet.genres:
434            # AudioTags.genres splits on ";" so we join multi-line values that way
435            tags.tags["genre"] = ";".join(cue_sheet.genres)
436        if cue_sheet.album_types:
437            # album_type reads "releasetype" as a single string and substring-matches
438            tags.tags["releasetype"] = " ".join(cue_sheet.album_types)
439        if cue_sheet.barcode:
440            tags.tags["barcode"] = cue_sheet.barcode
441        if cue_sheet.musicbrainz_albumid:
442            tags.tags["musicbrainzalbumid"] = cue_sheet.musicbrainz_albumid
443        if cue_sheet.musicbrainz_releasegroupid:
444            tags.tags["musicbrainzreleasegroupid"] = cue_sheet.musicbrainz_releasegroupid
445
446    async def _build_track(
447        self,
448        cue_track: CueTrack,
449        cue_item: FileSystemItem,
450        duration: float,
451        ctx: _TrackBuildContext,
452    ) -> Track:
453        """Construct a single Track from a CUE track entry."""
454        provider = self.provider
455        track_id = make_cue_track_id(cue_item.relative_path, cue_track.number)
456
457        # per-track PERFORMER wins over sheet-level. Multi-artist uses one PERFORMER
458        # line per artist (Vorbis convention); never delimiter-split, would mangle "AC/DC".
459        performer_names = cue_track.performers or list(ctx.album_performers)
460        track_artists: UniqueList[Artist | ItemMapping] = UniqueList()
461        for idx, artist_name in enumerate(performer_names):
462            artist = await provider._parse_artist(
463                name=artist_name,
464                sort_name=(
465                    cue_track.artist_sort_names[idx]
466                    if idx < len(cue_track.artist_sort_names)
467                    else None
468                ),
469                mbid=(
470                    cue_track.musicbrainz_artistids[idx]
471                    if idx < len(cue_track.musicbrainz_artistids)
472                    else None
473                ),
474            )
475            if artist:
476                track_artists.append(artist)
477        if not track_artists:
478            # neither the track nor the sheet declared a PERFORMER; fall back to
479            # the [unknown] artist rather than leaving the track artist-less
480            unknown = await provider._parse_artist(name=UNKNOWN_ARTIST, mbid=UNKNOWN_ARTIST_ID_MBID)
481            if unknown:
482                track_artists.append(unknown)
483
484        track = Track(
485            item_id=track_id,
486            provider=provider.instance_id,
487            name=cue_track.title or f"Track {cue_track.number}",
488            sort_name=cue_track.sort_name,
489            provider_mappings={
490                ProviderMapping(
491                    item_id=track_id,
492                    provider_domain=provider.domain,
493                    provider_instance=provider.instance_id,
494                    audio_format=ctx.audio_format,
495                    details=cue_metadata_checksum(cue_item.checksum),
496                    in_library=True,
497                )
498            },
499            track_number=cue_track.number,
500            disc_number=ctx.disc_number,
501            duration=round(duration),
502            date_added=ctx.date_added,
503        )
504
505        if track_artists:
506            track.artists = track_artists
507        if ctx.album:
508            track.album = ctx.album
509        for isrc in cue_track.isrcs:
510            track.external_ids.add((ExternalID.ISRC, isrc))
511        if recording_mbid := clean_mbid(cue_track.musicbrainz_recordingid, cue_item.relative_path):
512            # the setter keeps external_ids in sync
513            track.mbid = recording_mbid
514        if cue_track.musicbrainz_releasetrackid:
515            track.external_ids.add((ExternalID.MB_TRACK, cue_track.musicbrainz_releasetrackid))
516        if ctx.embedded_image is not None:
517            track.metadata.images = UniqueList([ctx.embedded_image])
518        if cue_track.genres:
519            # per-track REM GENRE wins over the shared audio-file/album genres
520            track.metadata.genres = set(cue_track.genres)
521        elif ctx.track_genres is not None:
522            track.metadata.genres = ctx.track_genres
523        if cue_track.copyright:
524            track.metadata.copyright = cue_track.copyright
525        if cue_track.grouping:
526            track.metadata.grouping = cue_track.grouping
527        if cue_track.comment:
528            track.metadata.description = cue_track.comment
529        if cue_track.explicit is not None:
530            track.metadata.explicit = cue_track.explicit
531        return track
532