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