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