music-assistant-server

1"""Filesystem musicprovider support for MusicAssistant."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7import logging
8import os
9import os.path
10import posixpath
11import urllib.parse
12from collections.abc import AsyncGenerator, Sequence
13from datetime import UTC, datetime
14from pathlib import Path
15from typing import TYPE_CHECKING, Any, ClassVar, cast
16from xml.parsers.expat import ExpatError
17
18import aiofiles
19import shortuuid
20import xmltodict
21from aiofiles.os import wrap
22from music_assistant_models.enums import (
23    ContentType,
24    EventType,
25    ExternalID,
26    ImageType,
27    MediaType,
28    ProviderFeature,
29    StreamType,
30)
31from music_assistant_models.errors import (
32    InvalidDataError,
33    MediaNotFoundError,
34    MusicAssistantError,
35    SetupFailedError,
36)
37from music_assistant_models.media_items import (
38    Album,
39    Artist,
40    Audiobook,
41    AudioFormat,
42    BrowseFolder,
43    ItemMapping,
44    MediaItemChapter,
45    MediaItemImage,
46    MediaItemType,
47    Playlist,
48    Podcast,
49    PodcastEpisode,
50    ProviderMapping,
51    SearchResults,
52    SoundEffect,
53    Track,
54    UniqueList,
55    is_track,
56)
57from music_assistant_models.streamdetails import MultiPartPath, StreamDetails
58
59from music_assistant.constants import (
60    CONF_PATH,
61    DB_TABLE_ALBUM_ARTISTS,
62    DB_TABLE_ALBUM_TRACKS,
63    DB_TABLE_ALBUMS,
64    DB_TABLE_ARTISTS,
65    DB_TABLE_PROVIDER_MAPPINGS,
66    DB_TABLE_TRACK_ARTISTS,
67    VARIOUS_ARTISTS_MBID,
68    VARIOUS_ARTISTS_NAME,
69    VERBOSE_LOG_LEVEL,
70)
71from music_assistant.controllers.tasks.context import (
72    report_current_task_failure,
73    update_current_task_progress_from_index,
74    update_current_task_progress_text,
75)
76from music_assistant.helpers import lyrics
77from music_assistant.helpers.compare import compare_strings, create_safe_string
78from music_assistant.helpers.json import SerializableType, json_loads
79from music_assistant.helpers.playlists import parse_m3u, parse_pls
80from music_assistant.helpers.tags import AudioTags, async_parse_tags, clean_mbid, split_items
81from music_assistant.helpers.util import (
82    TaskManager,
83    detect_charset,
84    parse_title_and_version,
85    try_parse_int,
86)
87from music_assistant.models.music_provider import MusicProvider
88
89from .constants import (
90    AUDIOBOOK_EXTENSIONS,
91    CACHE_CATEGORY_ALBUM_INFO,
92    CACHE_CATEGORY_ARTIST_INFO,
93    CACHE_CATEGORY_AUDIOBOOK_CHAPTERS,
94    CACHE_CATEGORY_FOLDER_IMAGES,
95    CACHE_CATEGORY_PODCAST_EPISODES,
96    CACHE_CATEGORY_PODCAST_METADATA,
97    CACHE_CATEGORY_SOUND_EFFECTS,
98    CONF_CONTENT_TYPE,
99    CONF_ENTRY_CONTENT_TYPE,
100    CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
101    CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
102    CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
103    CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
104    CONF_ENTRY_LIBRARY_SYNC_TRACKS,
105    CONF_ENTRY_MISSING_ALBUM_ARTIST,
106    CONF_ENTRY_PROPAGATE_GENRES,
107    CUE_EXTENSIONS,
108    DEFAULT_AUDIOBOOK_PODCAST_GENRE,
109    IMAGE_EXTENSIONS,
110    PARTIAL_LISTING_CACHE_EXPIRATION,
111    PLAYLIST_EXTENSIONS,
112    PODCAST_EPISODE_EXTENSIONS,
113    SOUND_EFFECT_EXTENSIONS,
114    SUPPORTED_EXTENSIONS,
115    TRACK_EXTENSIONS,
116    IsChapterFile,
117    content_type_config_entry,
118)
119from .cue import (
120    CueSheetHandler,
121    cue_referenced_audio_stem,
122    make_cue_track_id,
123    parse_cue_track_id,
124)
125from .helpers import (
126    FileSystemItem,
127    ScanErrors,
128    get_absolute_path,
129    get_album_dir,
130    get_artist_dir,
131    get_folder_signature,
132    get_relative_path,
133    recursive_iter,
134    sorted_scandir,
135)
136from .parsers import parse_album_nfo
137
138if TYPE_CHECKING:
139    from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
140    from music_assistant_models.provider import ProviderManifest
141
142    from music_assistant.mass import MusicAssistant
143    from music_assistant.models import ProviderInstanceType
144    from music_assistant.providers.musicbrainz import MusicbrainzProvider
145
146
147isdir = wrap(os.path.isdir)
148isfile = wrap(os.path.isfile)
149ismount = wrap(os.path.ismount)
150exists = wrap(os.path.exists)
151makedirs = wrap(os.makedirs)
152scandir = wrap(os.scandir)
153
154SUPPORTED_FEATURES = {
155    ProviderFeature.BROWSE,
156    ProviderFeature.SEARCH,
157}
158
159
160async def setup(
161    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
162) -> ProviderInstanceType:
163    """Initialize provider(instance) with given configuration."""
164    return LocalFileSystemProvider(mass, manifest, config)
165
166
167class LocalFileSystemProvider(MusicProvider):
168    """
169    Implementation of a musicprovider for (local) files.
170
171    Reads ID3 tags from file and falls back to parsing filename.
172    Optionally reads metadata from nfo files and images in folder structure <artist>/<album>.
173    Supports m3u files for playlists.
174    """
175
176    # parallel workers per sync; subclasses lower this for slower transports
177    _SYNC_CONCURRENCY: ClassVar[int] = 16
178    _sync_tracks: bool = True
179    _sync_playlists: bool = True
180
181    def __init__(
182        self,
183        mass: MusicAssistant,
184        manifest: ProviderManifest,
185        config: ProviderConfig,
186        base_path: str | None = None,
187    ) -> None:
188        """Initialize MusicProvider."""
189        super().__init__(mass, manifest, config, SUPPORTED_FEATURES)
190        # subclasses (NFS/SMB/...) mount elsewhere and pass their own base_path;
191        # the plain local provider reads its scan directory from the setup data
192        self.base_path: str = (
193            base_path if base_path is not None else cast("str", self.get_setup_value(CONF_PATH))
194        )
195        self.write_access: bool = False
196        self.sync_running: bool = False
197        self.media_content_type = cast(
198            "str", self.get_setup_value(CONF_CONTENT_TYPE, CONF_ENTRY_CONTENT_TYPE.default_value)
199        )
200        self._cue = CueSheetHandler(self)
201
202    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
203        """Return Config entries to configure this provider."""
204        # content type and path are collected by the setup flow; surface the (immutable)
205        # content type read-only so the sync options' depends_on chains still resolve
206        content_type = str(
207            self.get_setup_value(CONF_CONTENT_TYPE, CONF_ENTRY_CONTENT_TYPE.default_value)
208        )
209        return (
210            content_type_config_entry(content_type),
211            CONF_ENTRY_MISSING_ALBUM_ARTIST,
212            CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
213            CONF_ENTRY_LIBRARY_SYNC_TRACKS,
214            CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
215            CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
216            CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
217            CONF_ENTRY_PROPAGATE_GENRES,
218        )
219
220    @property
221    def supported_features(self) -> set[ProviderFeature]:
222        """Return the features supported by this Provider."""
223        base_features = {*SUPPORTED_FEATURES}
224        if self.media_content_type == "audiobooks":
225            return {ProviderFeature.LIBRARY_AUDIOBOOKS, *base_features}
226        if self.media_content_type == "podcasts":
227            return {ProviderFeature.LIBRARY_PODCASTS, *base_features}
228        if self.media_content_type == "sound_effects":
229            # sound effects are live-fetched content, never synced into the library
230            return {ProviderFeature.SOUND_EFFECTS, *base_features}
231        music_features = {
232            ProviderFeature.LIBRARY_ALBUMS,
233            ProviderFeature.LIBRARY_ARTISTS,
234            ProviderFeature.LIBRARY_TRACKS,
235            ProviderFeature.LIBRARY_PLAYLISTS,
236            *base_features,
237        }
238        if self.write_access:
239            music_features.add(ProviderFeature.PLAYLIST_TRACKS_EDIT)
240            music_features.add(ProviderFeature.PLAYLIST_CREATE)
241        return music_features
242
243    @property
244    def is_streaming_provider(self) -> bool:
245        """Return True if the provider is a streaming provider."""
246        return False
247
248    @property
249    def instance_name_postfix(self) -> str | None:
250        """Return a (default) instance name postfix for this provider instance."""
251        return self.base_path.split(os.sep)[-1]
252
253    async def handle_async_init(self) -> None:
254        """Handle async initialization of the provider."""
255        if not await isdir(self.base_path):
256            msg = f"Music Directory {self.base_path} does not exist"
257            raise SetupFailedError(
258                msg,
259                translation_key="music_directory_not_found",
260                translation_owner=self.translation_owner,
261                translation_args=[self.base_path],
262            )
263        await self.check_write_access()
264
265    async def get_diagnostics(self) -> dict[str, SerializableType]:
266        """Return diagnostics info for this provider to include in diagnostics reports."""
267        return {
268            "sync_running": self.sync_running,
269            "write_access": self.write_access,
270            "content_type": self.media_content_type,
271        }
272
273    async def search(
274        self,
275        search_query: str,
276        media_types: list[MediaType] | None,
277        limit: int = 5,
278    ) -> SearchResults:
279        """Perform search on this file based musicprovider."""
280        result = SearchResults()
281        # searching the filesystem is slow and unreliable,
282        # so instead we just query the db...
283        if media_types is None or MediaType.TRACK in media_types:
284            result.tracks = await self.mass.music.tracks.get_library_items_by_query(
285                search=search_query, provider_filter=[self.instance_id], limit=limit
286            )
287
288        if media_types is None or MediaType.ALBUM in media_types:
289            result.albums = await self.mass.music.albums.get_library_items_by_query(
290                search=search_query,
291                provider_filter=[self.instance_id],
292                limit=limit,
293            )
294
295        if media_types is None or MediaType.ARTIST in media_types:
296            result.artists = await self.mass.music.artists.get_library_items_by_query(
297                search=search_query,
298                provider_filter=[self.instance_id],
299                limit=limit,
300            )
301        if media_types is None or MediaType.PLAYLIST in media_types:
302            result.playlists = await self.mass.music.playlists.get_library_items_by_query(
303                search=search_query,
304                provider_filter=[self.instance_id],
305                limit=limit,
306            )
307        if media_types is None or MediaType.AUDIOBOOK in media_types:
308            result.audiobooks = await self.mass.music.audiobooks.get_library_items_by_query(
309                search=search_query,
310                provider_filter=[self.instance_id],
311                limit=limit,
312            )
313        if media_types is None or MediaType.PODCAST in media_types:
314            result.podcasts = await self.mass.music.podcasts.get_library_items_by_query(
315                search=search_query,
316                provider_filter=[self.instance_id],
317                limit=limit,
318            )
319        return result
320
321    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
322        """
323        Browse this provider's items.
324
325        :param path: The path to browse, (e.g. provid://artists).
326        """
327        # for audiobooks and podcasts we just return all library items
328        if self.media_content_type == "podcasts":
329            return await self.mass.music.podcasts.library_items(
330                provider=self.instance_id, summary=False
331            )
332        if self.media_content_type == "audiobooks":
333            return await self.mass.music.audiobooks.library_items(
334                provider=self.instance_id, summary=False
335            )
336        items: list[MediaItemType | ItemMapping | BrowseFolder] = []
337        item_path = path.split("://", 1)[1]
338        if not item_path:
339            item_path = ""
340        scanned = await self._scandir(item_path)
341        # expand CUE sheets into per-track entries and hide the companion audio;
342        # synthetic ids match those minted during sync so get_track resolves them
343        cue_stems: set[str] = set()
344        if self.media_content_type == "music":
345            for item in scanned:
346                if item.ext not in CUE_EXTENSIONS:
347                    continue
348                cue_stems.add(item.absolute_path.rsplit(".", 1)[0])
349                try:
350                    cue_sheet = await self._cue.load_cue_sheet(item)
351                except InvalidDataError as err:
352                    self.logger.warning("Unable to parse CUE sheet %s: %s", item.relative_path, err)
353                    continue
354                # also hide the audio file named in the CUE (may differ from its stem)
355                if companion_stem := cue_referenced_audio_stem(item, cue_sheet):
356                    cue_stems.add(companion_stem)
357                for cue_track in cue_sheet.tracks:
358                    items.append(
359                        ItemMapping(
360                            media_type=MediaType.TRACK,
361                            item_id=make_cue_track_id(item.relative_path, cue_track.number),
362                            provider=self.instance_id,
363                            name=cue_track.title or f"Track {cue_track.number}",
364                        )
365                    )
366        for item in scanned:
367            if not item.is_dir and ("." not in item.filename or not item.ext):
368                # skip system files and files without extension
369                continue
370
371            if item.is_dir:
372                items.append(
373                    BrowseFolder(
374                        item_id=item.relative_path,
375                        provider=self.instance_id,
376                        path=f"{self.instance_id}://{item.relative_path}",
377                        name=item.filename,
378                        # mark folder as playable, assuming it contains tracks underneath
379                        is_playable=True,
380                    )
381                )
382            elif item.ext in TRACK_EXTENSIONS:
383                if item.absolute_path.rsplit(".", 1)[0] in cue_stems:
384                    continue
385                items.append(
386                    ItemMapping(
387                        media_type=(
388                            MediaType.SOUND_EFFECT
389                            if self.media_content_type == "sound_effects"
390                            else MediaType.TRACK
391                        ),
392                        item_id=item.relative_path,
393                        provider=self.instance_id,
394                        name=item.filename,
395                    )
396                )
397            elif item.ext in PLAYLIST_EXTENSIONS and self.media_content_type == "music":
398                items.append(
399                    ItemMapping(
400                        media_type=MediaType.PLAYLIST,
401                        item_id=item.relative_path,
402                        provider=self.instance_id,
403                        name=item.filename,
404                    )
405                )
406        if self.media_content_type == "music":
407            track_indexes = [
408                index
409                for index, item in enumerate(items)
410                if isinstance(item, ItemMapping) and item.media_type == MediaType.TRACK
411            ]
412            library_tracks = await asyncio.gather(
413                *(
414                    self.mass.music.tracks.get_library_item_by_prov_id(
415                        items[index].item_id, self.instance_id
416                    )
417                    for index in track_indexes
418                )
419            )
420            for index, library_track in zip(track_indexes, library_tracks, strict=True):
421                if library_track:
422                    items[index] = library_track
423        return items
424
425    async def sync_library(self, media_type: MediaType) -> None:
426        """Run library sync for this provider."""
427        if media_type in (MediaType.ARTIST, MediaType.ALBUM):
428            # artists and albums are synced as part of track sync
429            return
430        if self.media_content_type == "sound_effects":
431            # sound effects are live-fetched content, never synced into the library
432            return
433        # check if any sync options are enabled for this content type
434        # the filesystem provider processes all file types in one scan,
435        # so we can return early if nothing needs syncing
436        if self.media_content_type == "music":
437            self._sync_tracks = bool(self.config.get_value(CONF_ENTRY_LIBRARY_SYNC_TRACKS.key))
438            self._sync_playlists = bool(
439                self.config.get_value(CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS.key)
440            )
441            if not self._sync_tracks and not self._sync_playlists:
442                return
443        elif self.media_content_type == "audiobooks":
444            if not self.config.get_value(CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS.key):
445                return
446        elif self.media_content_type == "podcasts":
447            if not self.config.get_value(CONF_ENTRY_LIBRARY_SYNC_PODCASTS.key):
448                return
449        assert self.mass.music.database
450        if self.sync_running:
451            self.logger.warning("Library sync already running for %s", self.name)
452            return
453        file_checksums: dict[str, str] = {}
454        # NOTE: we always run a scan of the entire library, as we need to detect changes
455        # we ignore any given mediatype(s) and just scan all supported files
456        query = (
457            f"SELECT provider_item_id, details FROM {DB_TABLE_PROVIDER_MAPPINGS} "
458            f"WHERE provider_instance = '{self.instance_id}' "
459            f"AND media_type in ('track', 'playlist', 'audiobook', 'podcast_episode')"
460        )
461        for db_row in await self.mass.music.database.get_rows_from_query(query, limit=0):
462            file_checksums[db_row["provider_item_id"]] = str(db_row["details"])
463        # provider_mappings stores synthetic per-track ids for CUE sheets, not the
464        # CUE path, so derive a path-keyed checksum map for the scan classifier
465        cue_file_checksums: dict[str, str] = {}
466        for prov_item_id, checksum in file_checksums.items():
467            parsed = parse_cue_track_id(prov_item_id)
468            if parsed is not None:
469                cue_file_checksums[parsed[0]] = checksum
470        # find all supported files in the base directory and all subfolders
471        # we work bottom up, as-in we derive all info from the tracks
472        cur_filenames: set[str] = set()
473        prev_filenames = set(file_checksums.keys())
474
475        items_to_process: list[tuple[FileSystemItem, str | None]] = []
476        unchanged_cue_items: list[FileSystemItem] = []
477        # absolute paths of every CUE sheet in this scan with the ".cue" stripped,
478        # used for O(1) companion-CUE lookups per audio file
479        cue_stems: set[str] = set()
480        # collects the errors raised while walking the tree; any error means the
481        # scan is incomplete, a fatal one means the provider is unreachable
482        scan_errors = ScanErrors()
483
484        self.sync_running = True
485        try:
486            await self._enumerate_files_for_sync(
487                file_checksums=file_checksums,
488                cue_file_checksums=cue_file_checksums,
489                cur_filenames=cur_filenames,
490                items_to_process=items_to_process,
491                unchanged_cue_items=unchanged_cue_items,
492                cue_stems=cue_stems,
493                scan_errors=scan_errors,
494            )
495            if scan_errors.fatal:
496                # the storage is gone, so reading the files collected before it went
497                # away would only add a timeout each
498                self.logger.error("Aborting sync for %s: %s", self.name, scan_errors.fatal)
499                report_current_task_failure("Sync aborted: filesystem unavailable during scan")
500                self._set_available(False)
501                return
502            # a CUE may name an audio file other than its own; hide that companion too
503            if self.media_content_type == "music":
504                for cue_item in (
505                    *unchanged_cue_items,
506                    *(item for item, _ in items_to_process if item.ext in CUE_EXTENSIONS),
507                ):
508                    try:
509                        cue_sheet = await self._cue.load_cue_sheet(cue_item)
510                    except InvalidDataError:
511                        continue
512                    if companion_stem := cue_referenced_audio_stem(cue_item, cue_sheet):
513                        cue_stems.add(companion_stem)
514            # drop CUE companion audio: absorbed into CUE tracks and not tracked in
515            # provider_mappings, so they would otherwise flag as changed every sync
516            items_to_process = [
517                (item, prev)
518                for item, prev in items_to_process
519                if not (
520                    item.ext in TRACK_EXTENSIONS
521                    and item.absolute_path.rsplit(".", 1)[0] in cue_stems
522                )
523            ]
524            # register synthetic track IDs for unchanged CUE files so the
525            # deletion pass does not treat them as removed
526            for cue_item in unchanged_cue_items:
527                try:
528                    cue_sheet = await self._cue.load_cue_sheet(cue_item)
529                except InvalidDataError as err:
530                    self.logger.warning(
531                        "Unable to parse CUE sheet %s: %s", cue_item.relative_path, err
532                    )
533                    continue
534                for cue_track in cue_sheet.tracks:
535                    cur_filenames.add(make_cue_track_id(cue_item.relative_path, cue_track.number))
536            total_items = len(items_to_process)
537            self.logger.info(
538                "Found %d changed/new items to process for %s",
539                total_items,
540                self.name,
541            )
542
543            # _SYNC_CONCURRENCY caps parallelism per provider (NFS/SMB/WebDAV friendly)
544            processed_count = 0
545
546            async def _process(item: FileSystemItem, prev_checksum: str | None) -> None:
547                nonlocal processed_count
548                if await self._process_item_async(
549                    item, prev_checksum, cur_filenames, cue_stems, prev_filenames
550                ):
551                    cur_filenames.add(item.relative_path)
552                processed_count += 1
553                if processed_count % 50 == 0 or processed_count == total_items:
554                    update_current_task_progress_from_index(
555                        processed_count,
556                        total_items,
557                        f"Processed {processed_count}/{total_items} files",
558                    )
559
560            async with TaskManager(self.mass, self._SYNC_CONCURRENCY) as tm:
561                for item, prev_checksum in items_to_process:
562                    await tm.create_task_with_limit(_process(item, prev_checksum))
563        finally:
564            self.sync_running = False
565
566        # do not run deletions on a clean but empty scan of a previously non-empty library
567        # (wrong share mounted, empty backup mount, ...)
568        if prev_filenames and not cur_filenames:
569            self.logger.error(
570                "Aborting sync for %s: scan found no files but %d were previously indexed",
571                self.name,
572                len(prev_filenames),
573            )
574            report_current_task_failure(
575                f"Sync aborted: scan found no files but {len(prev_filenames)} "
576                "were previously indexed"
577            )
578            return
579
580        # a scan that skipped folders or files is incomplete: what it missed is still
581        # there, so deleting it from the library would throw away valid content
582        if scan_errors.incomplete:
583            summary = scan_errors.describe()
584            self.logger.warning("Skipping deletions for %s: %s", self.name, summary)
585            report_current_task_failure(f"Deletions skipped: {summary}")
586        else:
587            deleted_files = prev_filenames - cur_filenames
588            await self._process_deletions(deleted_files)
589            await self._process_orphaned_albums_and_artists()
590
591        # flag provider as available again if an earlier sync had marked it down
592        self._set_available(True)
593
594    async def get_artist(self, prov_artist_id: str) -> Artist:
595        """Get full artist details by id."""
596        db_artist = await self.mass.music.artists.get_library_item_by_prov_id(
597            prov_artist_id, self.instance_id
598        )
599        if not db_artist:
600            # this may happen if the artist is not in the db yet
601            # e.g. when browsing the filesystem
602            if await self.exists(prov_artist_id):
603                return await self._parse_artist(prov_artist_id, artist_path=prov_artist_id)
604            return await self._parse_artist(prov_artist_id)
605
606        # prov_artist_id is either an actual (relative) path or a name (as fallback)
607        safe_artist_name = create_safe_string(prov_artist_id, lowercase=False, replace_space=False)
608        if await self.exists(prov_artist_id):
609            artist_path = prov_artist_id
610        elif await self.exists(safe_artist_name):
611            artist_path = safe_artist_name
612        else:
613            for prov_mapping in db_artist.provider_mappings:
614                if prov_mapping.provider_instance != self.instance_id:
615                    continue
616                if prov_mapping.url:
617                    artist_path = prov_mapping.url
618                    break
619            else:
620                # this is an artist without an actual path on disk
621                # return the info we already have in the db
622                return db_artist
623        return await self._parse_artist(
624            db_artist.name,
625            sort_name=db_artist.sort_name,
626            mbid=db_artist.mbid,
627            artist_path=artist_path,
628        )
629
630    async def get_album(self, prov_album_id: str) -> Album:
631        """Get full album details by id."""
632        parsed_cue_paths: set[str] = set()
633        for track in await self.get_album_tracks(prov_album_id):
634            for prov_mapping in track.provider_mappings:
635                if prov_mapping.provider_instance != self.instance_id:
636                    continue
637                if parsed := parse_cue_track_id(prov_mapping.item_id):
638                    # every track from the same CUE shares the same album; only parse once
639                    if parsed[0] in parsed_cue_paths:
640                        continue
641                    parsed_cue_paths.add(parsed[0])
642                    cue_item = await self.resolve(parsed[0])
643                    for cue_track in await self._cue.parse_tracks(cue_item):
644                        if isinstance(cue_track.album, Album):
645                            return cue_track.album
646                    continue
647                file_item = await self.resolve(prov_mapping.item_id)
648                tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
649                full_track = await self._parse_track(file_item, tags)
650                assert isinstance(full_track.album, Album)
651                return full_track.album
652        msg = f"Album not found: {prov_album_id}"
653        raise MediaNotFoundError(msg)
654
655    async def get_track(self, prov_track_id: str) -> Track:
656        """Get full track details by id."""
657        # ruff: noqa: PLR0915
658        if parsed := parse_cue_track_id(prov_track_id):
659            cue_item = await self.resolve(parsed[0])
660            for cue_track in await self._cue.parse_tracks(cue_item):
661                if cue_track.item_id == prov_track_id:
662                    return cue_track
663            msg = f"CUE track not found: {prov_track_id}"
664            raise MediaNotFoundError(msg)
665
666        if not await self.exists(prov_track_id):
667            msg = f"Track path does not exist: {prov_track_id}"
668            raise MediaNotFoundError(msg)
669
670        file_item = await self.resolve(prov_track_id)
671        tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
672        return await self._parse_track(file_item, tags=tags, full_album_metadata=True)
673
674    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
675        """Get (full) podcast episode details by id."""
676        if not await self.exists(prov_episode_id):
677            msg = f"Episode path does not exist: {prov_episode_id}"
678            raise MediaNotFoundError(msg)
679        file_item = await self.resolve(prov_episode_id)
680        tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
681        return await self._parse_podcast_episode(file_item, tags=tags)
682
683    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
684        """Get full playlist details by id."""
685        if not await self.exists(prov_playlist_id):
686            msg = f"Playlist path does not exist: {prov_playlist_id}"
687            raise MediaNotFoundError(msg)
688
689        file_item = await self.resolve(prov_playlist_id)
690        playlist = Playlist(
691            item_id=file_item.relative_path,
692            provider=self.instance_id,
693            name=file_item.name,
694            provider_mappings={
695                ProviderMapping(
696                    item_id=file_item.relative_path,
697                    provider_domain=self.domain,
698                    provider_instance=self.instance_id,
699                    details=file_item.checksum,
700                    in_library=True,
701                )
702            },
703        )
704        playlist.is_editable = ProviderFeature.PLAYLIST_TRACKS_EDIT in self.supported_features
705        # only playlists in the root are editable - all other are read only
706        if "/" in prov_playlist_id or "\\" in prov_playlist_id:
707            playlist.is_editable = False
708        # we do not (yet) have support to edit/create pls playlists, only m3u files can be edited
709        if file_item.ext == "pls":
710            playlist.is_editable = False
711        playlist.owner = self.name
712        # Check for local image with the same basename
713        if local_image := await self._get_playlist_local_image(file_item):
714            playlist.metadata.images = UniqueList([local_image])
715        return playlist
716
717    async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
718        """Get full audiobook details by id."""
719        # ruff: noqa: PLR0915
720        if not await self.exists(prov_audiobook_id):
721            msg = f"Audiobook path does not exist: {prov_audiobook_id}"
722            raise MediaNotFoundError(msg)
723
724        file_item = await self.resolve(prov_audiobook_id)
725        tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
726        return await self._parse_audiobook(file_item, tags=tags)
727
728    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
729        """Get full podcast details by id."""
730        async for episode in self.get_podcast_episodes(prov_podcast_id):
731            assert isinstance(episode.podcast, Podcast)
732            return episode.podcast
733        msg = f"Podcast not found: {prov_podcast_id}"
734        raise MediaNotFoundError(msg)
735
736    async def get_sound_effect(self, prov_sound_effect_id: str) -> SoundEffect:
737        """Get full sound effect details by id."""
738        if not await self.exists(prov_sound_effect_id):
739            msg = f"Sound effect path does not exist: {prov_sound_effect_id}"
740            raise MediaNotFoundError(msg)
741        file_item = await self.resolve(prov_sound_effect_id)
742        return await self._get_or_parse_sound_effect(file_item)
743
744    async def get_sound_effects(self) -> AsyncGenerator[SoundEffect]:
745        """Get all sound effect items this provider offers."""
746
747        def _walk() -> list[FileSystemItem]:
748            return sorted(
749                recursive_iter(
750                    self.base_path, self.base_path, SOUND_EFFECT_EXTENSIONS, self.logger
751                ),
752                key=lambda x: x.relative_path,
753            )
754
755        for file_item in await asyncio.to_thread(_walk):
756            yield await self._get_or_parse_sound_effect(file_item)
757
758    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
759        """Get album tracks for given album id."""
760        # filesystem items are always stored in db so we can query the database
761        db_album = await self.mass.music.albums.get_library_item_by_prov_id(
762            prov_album_id, self.instance_id
763        )
764        if db_album is None:
765            msg = f"Album not found: {prov_album_id}"
766            raise MediaNotFoundError(msg)
767        album_tracks = await self.mass.music.albums.get_library_album_tracks(db_album.item_id)
768        return [
769            track
770            for track in album_tracks
771            if any(x.provider_instance == self.instance_id for x in track.provider_mappings)
772        ]
773
774    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
775        """Get playlist tracks."""
776        result: list[Track] = []
777        if page > 0:
778            # paging not (yet) supported
779            return result
780        if not await self.exists(prov_playlist_id):
781            msg = f"Playlist path does not exist: {prov_playlist_id}"
782            raise MediaNotFoundError(msg)
783
784        file_item = await self.resolve(prov_playlist_id)
785        # We are using the checksum of the playlist file here to invalidate the cache
786        # when a change has been made to the playlist file (ie track addition/deletion)
787        cache_checksum = file_item.checksum
788
789        cache_key = f"get_playlist_tracks.{prov_playlist_id}"
790        cached_data = await self.mass.cache.get(
791            cache_key,
792            provider=self.instance_id,
793            checksum=cache_checksum,
794            category=0,
795            base_class=Track,
796        )
797        if cached_data is not None:
798            return cached_data  # type: ignore[no-any-return]
799
800        _, ext = prov_playlist_id.rsplit(".", 1)
801        try:
802            # get playlist file contents
803            playlist_data_raw = await self._read_file(prov_playlist_id)
804            encoding = await detect_charset(playlist_data_raw)
805            playlist_data = playlist_data_raw.decode(encoding, errors="replace")
806
807            if ext in ("m3u", "m3u8"):
808                playlist_lines = parse_m3u(playlist_data)
809            else:
810                playlist_lines = parse_pls(playlist_data)
811
812            for idx, playlist_line in enumerate(playlist_lines, 1):
813                if "#EXT" in playlist_line.path:
814                    continue
815                if track := await self._parse_playlist_line(
816                    playlist_line.path, os.path.dirname(prov_playlist_id)
817                ):
818                    track.position = idx
819                    result.append(track)
820
821        except Exception as err:
822            self.logger.warning(
823                "Error while parsing playlist %s: %s",
824                prov_playlist_id,
825                str(err),
826                exc_info=err if self.logger.isEnabledFor(10) else None,
827            )
828
829        await self.mass.cache.set(
830            key=cache_key,
831            data=[track.to_dict() for track in result],
832            expiration=3600 * 24 * 365,  # File timestamp checksum handles invalidation
833            provider=self.instance_id,
834            checksum=cache_checksum,
835            category=0,
836        )
837
838        return result
839
840    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
841        """Get podcast episodes for given podcast id."""
842        folder_items = [item for item in await self._scandir(prov_podcast_id) if not item.is_dir]
843        episode_files = [x for x in folder_items if x.ext in PODCAST_EPISODE_EXTENSIONS]
844        # artwork and metadata.json count towards the signature too, because the parse embeds
845        # them into every episode. Case-insensitive, matching _get_podcast_metadata's exists()
846        signature_files = [
847            x
848            for x in folder_items
849            if x.ext in PODCAST_EPISODE_EXTENSIONS
850            or x.ext in IMAGE_EXTENSIONS
851            or x.filename.lower() == "metadata.json"
852        ]
853        cache_key = f"podcast_episodes.{prov_podcast_id}"
854        cache_checksum = get_folder_signature(signature_files)
855        if (
856            cached_episodes := await self.mass.cache.get(
857                cache_key,
858                provider=self.instance_id,
859                category=CACHE_CATEGORY_PODCAST_EPISODES,
860                checksum=cache_checksum,
861                base_class=PodcastEpisode,
862            )
863        ) is not None:
864            for episode in cached_episodes:
865                yield episode
866            return
867
868        # these caches have no checksum of their own, so drop them before parsing or the new
869        # entry gets the values the signature just invalidated. Refill once, or every parse
870        # task below misses at the same time and repeats the same scandir and file read
871        for stale_category in (CACHE_CATEGORY_FOLDER_IMAGES, CACHE_CATEGORY_PODCAST_METADATA):
872            await self.mass.cache.delete(
873                prov_podcast_id, category=stale_category, provider=self.instance_id
874            )
875        await self._get_local_images(prov_podcast_id)
876        await self._get_podcast_metadata(prov_podcast_id)
877
878        # collected by index so the listing keeps scandir order, not parse completion order
879        parsed: list[PodcastEpisode | None] = [None] * len(episode_files)
880
881        async def _process_podcast_episode(index: int, item: FileSystemItem) -> None:
882            try:
883                tags = await async_parse_tags(item.absolute_path, item.file_size)
884                parsed[index] = await self._parse_podcast_episode(item, tags)
885            except MusicAssistantError as err:
886                self.logger.warning(
887                    "Could not parse uri/file %s to podcast episode: %s",
888                    item.relative_path,
889                    str(err),
890                )
891
892        # reuse the per-sync worker limit: the slowest filesystems to parse are exactly the
893        # ones that lower it
894        async with TaskManager(self.mass, self._SYNC_CONCURRENCY) as tm:
895            for index, item in enumerate(episode_files):
896                await tm.create_task_with_limit(_process_podcast_episode(index, item))
897
898        episodes = [episode for episode in parsed if episode is not None]
899        # cache an incomplete listing briefly rather than not at all, so one unreadable file
900        # cannot make every request re-parse the whole folder
901        complete = len(episodes) == len(episode_files)
902        await self.mass.cache.set(
903            key=cache_key,
904            data=[episode.to_dict() for episode in episodes],
905            # a complete listing is invalidated by the folder signature instead
906            expiration=3600 * 24 * 365 if complete else PARTIAL_LISTING_CACHE_EXPIRATION,
907            provider=self.instance_id,
908            category=CACHE_CATEGORY_PODCAST_EPISODES,
909            checksum=cache_checksum,
910        )
911
912        for episode in episodes:
913            yield episode
914
915    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
916        """Add track(s) to playlist."""
917        if not await self.exists(prov_playlist_id):
918            msg = f"Playlist path does not exist: {prov_playlist_id}"
919            raise MediaNotFoundError(msg)
920        playlist_filename = self.get_absolute_path(prov_playlist_id)
921        async with aiofiles.open(playlist_filename, encoding="utf-8") as _file:
922            playlist_data = await _file.read()
923        for file_path in prov_track_ids:
924            track = await self.get_track(file_path)
925            playlist_data += f"\n#EXTINF:{track.duration or 0},{track.name}\n{file_path}\n"
926
927        # write playlist file (always in utf-8)
928        async with aiofiles.open(playlist_filename, "w", encoding="utf-8") as _file:
929            await _file.write(playlist_data)
930
931    async def remove_playlist_tracks(
932        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
933    ) -> None:
934        """Remove track(s) from playlist."""
935        if not await self.exists(prov_playlist_id):
936            msg = f"Playlist path does not exist: {prov_playlist_id}"
937            raise MediaNotFoundError(msg)
938        _, ext = prov_playlist_id.rsplit(".", 1)
939        # get playlist file contents
940        playlist_filename = self.get_absolute_path(prov_playlist_id)
941        async with aiofiles.open(playlist_filename, encoding="utf-8") as _file:
942            playlist_data = await _file.read()
943        # get current contents first
944        if ext in ("m3u", "m3u8"):
945            playlist_items = parse_m3u(playlist_data)
946        else:
947            playlist_items = parse_pls(playlist_data)
948        # remove items by index
949        for i in sorted(positions_to_remove, reverse=True):
950            # position = index + 1
951            del playlist_items[i - 1]
952        # build new playlist data
953        new_playlist_data = "#EXTM3U\n"
954        for item in playlist_items:
955            new_playlist_data += f"\n#EXTINF:{item.length or 0},{item.title}\n{item.path}\n"
956        async with aiofiles.open(playlist_filename, "w", encoding="utf-8") as _file:
957            await _file.write(new_playlist_data)
958
959    async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
960        """Create a new playlist on provider with given name."""
961        # creating a new playlist on the filesystem is as easy
962        # as creating a new (empty) file with the m3u extension...
963        # filename = await self.resolve(f"{name}.m3u")
964        filename = f"{name}.m3u"
965        playlist_filename = self.get_absolute_path(filename)
966        async with aiofiles.open(playlist_filename, "w", encoding="utf-8") as _file:
967            await _file.write("#EXTM3U\n")
968        return await self.get_playlist(filename)
969
970    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
971        """Return the content details for the given track when it will be streamed."""
972        try:
973            if media_type == MediaType.AUDIOBOOK:
974                return await self._get_stream_details_for_audiobook(item_id)
975            if media_type == MediaType.PODCAST_EPISODE:
976                return await self._get_stream_details_for_podcast_episode(item_id)
977            if media_type == MediaType.SOUND_EFFECT:
978                return await self._get_stream_details_for_sound_effect(item_id)
979            return await self._get_stream_details_for_track(item_id)
980        except FileNotFoundError:
981            self.logger.warning(
982                "File not found for media item %s",
983                item_id,
984            )
985            msg = f"Media file not found: {item_id}"
986            raise MediaNotFoundError(msg)
987
988    async def get_audio_stream(
989        self, streamdetails: StreamDetails, seek_position: int = 0
990    ) -> AsyncGenerator[bytes]:
991        """Return the custom audio stream for the provider item."""
992        # only CUE-derived tracks use StreamType.CUSTOM in this provider
993        async for chunk in self._cue.get_audio_stream(streamdetails, seek_position):
994            yield chunk
995
996    async def resolve_image(self, path: str) -> str | bytes:
997        """
998        Resolve an image from an image path.
999
1000        This either returns (a generator to get) raw bytes of the image or
1001        a string with an http(s) URL or local path that is accessible from the server.
1002        """
1003        # drop the cache-busting suffix appended by _versioned_image_path
1004        try:
1005            file_item = await self.resolve(path.split("?cs=", 1)[0])
1006        except FileNotFoundError as err:
1007            # the referenced image file was removed from disk; surface a typed
1008            # not-found so the image layer treats it as a missing image
1009            raise MediaNotFoundError(f"Image not found: {path}") from err
1010        return file_item.absolute_path
1011
1012    async def check_write_access(self) -> None:
1013        """Perform check if we have write access."""
1014        # verify write access to determine we have playlist create/edit support
1015        # overwrite with provider specific implementation if needed
1016        temp_file_name = self.get_absolute_path(f"{shortuuid.random(8)}.txt")
1017        try:
1018            async with aiofiles.open(temp_file_name, "w") as _file:
1019                await _file.write("test")
1020            await asyncio.to_thread(os.remove, temp_file_name)
1021            self.write_access = True
1022        except Exception as err:
1023            self.logger.debug("Write access disabled: %s", str(err))
1024
1025    async def resolve(self, file_path: str) -> FileSystemItem:
1026        """Resolve (absolute or relative) path to FileSystemItem."""
1027        absolute_path = self.get_absolute_path(file_path)
1028
1029        def _create_item() -> FileSystemItem:
1030            if os.path.isdir(absolute_path):
1031                return FileSystemItem(
1032                    filename=Path(file_path).name,
1033                    relative_path=get_relative_path(self.base_path, file_path),
1034                    absolute_path=absolute_path,
1035                    is_dir=True,
1036                )
1037            stat_info = Path(absolute_path).stat(follow_symlinks=False)
1038            return FileSystemItem(
1039                filename=Path(file_path).name,
1040                relative_path=get_relative_path(self.base_path, file_path),
1041                absolute_path=absolute_path,
1042                is_dir=False,
1043                checksum=str(int(stat_info.st_mtime)),
1044                file_size=stat_info.st_size,
1045            )
1046
1047        return await asyncio.to_thread(_create_item)
1048
1049    async def exists(self, file_path: str) -> bool:
1050        """Return bool is this FileSystem musicprovider has given file/dir."""
1051        if not file_path:
1052            return False
1053        try:
1054            abs_path = self.get_absolute_path(file_path)
1055        except MediaNotFoundError:
1056            # a path that escapes the base directory simply does not exist here
1057            return False
1058        return bool(await exists(abs_path))
1059
1060    def get_absolute_path(self, file_path: str) -> str:
1061        """Return absolute path for given file path."""
1062        return get_absolute_path(self.base_path, file_path)
1063
1064    async def _enumerate_files_for_sync(
1065        self,
1066        *,
1067        file_checksums: dict[str, str],
1068        cue_file_checksums: dict[str, str],
1069        cur_filenames: set[str],
1070        items_to_process: list[tuple[FileSystemItem, str | None]],
1071        unchanged_cue_items: list[FileSystemItem],
1072        cue_stems: set[str],
1073        scan_errors: ScanErrors,
1074    ) -> None:
1075        """
1076        Walk every supported file under the provider root and populate the sync buckets.
1077
1078        Override in subclasses that cannot use a local ``os.scandir`` walk.
1079        Implementations must route each discovered file through
1080        :meth:`_classify_scan_item`, report every unreadable directory to
1081        ``scan_errors`` and stop the walk once it reports ``aborted``.
1082
1083        :param file_checksums: Previously stored checksum per provider item id.
1084        :param cue_file_checksums: Previously stored checksum keyed by CUE relative_path.
1085        :param cur_filenames: Receives the ids/paths present in this scan.
1086        :param items_to_process: Receives changed or new items to process.
1087        :param unchanged_cue_items: Receives CUE sheets whose checksum matches.
1088        :param cue_stems: Receives absolute paths (minus extension) of CUE sheets.
1089        :param scan_errors: Receives the errors raised while walking the tree.
1090        """
1091        ignore_album_playlists = self.media_content_type == "music" and bool(
1092            self.config.get_value(CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS.key)
1093        )
1094
1095        def _walk() -> None:
1096            for scanned, item in enumerate(
1097                recursive_iter(
1098                    self.base_path,
1099                    self.base_path,
1100                    SUPPORTED_EXTENSIONS,
1101                    self.logger,
1102                    scan_errors=scan_errors,
1103                ),
1104                start=1,
1105            ):
1106                if scanned % 500 == 0:
1107                    update_current_task_progress_text(f"Scanning files: {scanned} found")
1108                self._classify_scan_item(
1109                    item,
1110                    file_checksums=file_checksums,
1111                    cue_file_checksums=cue_file_checksums,
1112                    cur_filenames=cur_filenames,
1113                    items_to_process=items_to_process,
1114                    unchanged_cue_items=unchanged_cue_items,
1115                    cue_stems=cue_stems,
1116                    ignore_album_playlists=ignore_album_playlists,
1117                )
1118
1119        await asyncio.to_thread(_walk)
1120
1121    def _classify_scan_item(
1122        self,
1123        item: FileSystemItem,
1124        *,
1125        file_checksums: dict[str, str],
1126        cue_file_checksums: dict[str, str],
1127        cur_filenames: set[str],
1128        items_to_process: list[tuple[FileSystemItem, str | None]],
1129        unchanged_cue_items: list[FileSystemItem],
1130        cue_stems: set[str],
1131        ignore_album_playlists: bool,
1132    ) -> None:
1133        """
1134        Route a single scanned file into the correct sync bucket.
1135
1136        :param item: The file to classify.
1137        :param file_checksums: Previously stored checksum per provider item id.
1138        :param cue_file_checksums: Previously stored checksum keyed by CUE relative_path.
1139        :param cur_filenames: Receives the ids/paths present in this scan.
1140        :param items_to_process: Receives changed or new items to process.
1141        :param unchanged_cue_items: Receives CUE sheets whose checksum matches.
1142        :param cue_stems: Receives absolute paths (minus extension) of CUE sheets.
1143        :param ignore_album_playlists: When True, skip playlists nested inside
1144            album directories.
1145        """
1146        # a file this provider never imports gets no mapping, so it would flag as
1147        # changed on every sync; it is still on disk, so record it as present
1148        if not self._is_imported_file(item):
1149            cur_filenames.add(item.relative_path)
1150            return
1151        # skip playlists in album directories if configured
1152        if (
1153            item.ext in PLAYLIST_EXTENSIONS
1154            and ignore_album_playlists
1155            and len(item.relative_path.split("/")) > 2
1156        ):
1157            return
1158        is_cue = item.ext in CUE_EXTENSIONS and self.media_content_type == "music"
1159        if is_cue:
1160            cue_stems.add(item.absolute_path.rsplit(".", 1)[0])
1161            prev_checksum = cue_file_checksums.get(item.relative_path)
1162        else:
1163            prev_checksum = file_checksums.get(item.relative_path)
1164        if item.checksum == prev_checksum:
1165            # unchanged, just record it as still present
1166            cur_filenames.add(item.relative_path)
1167            if is_cue:
1168                unchanged_cue_items.append(item)
1169        else:
1170            items_to_process.append((item, prev_checksum))
1171
1172    def _is_imported_file(self, item: FileSystemItem) -> bool:
1173        """Return True when this provider imports the given file into the library."""
1174        if self.media_content_type == "music":
1175            if item.ext in CUE_EXTENSIONS:
1176                return True
1177            if item.ext in TRACK_EXTENSIONS:
1178                return self._sync_tracks
1179            if item.ext in PLAYLIST_EXTENSIONS:
1180                return self._sync_playlists
1181            return False
1182        if self.media_content_type == "audiobooks":
1183            return item.ext in AUDIOBOOK_EXTENSIONS
1184        if self.media_content_type == "podcasts":
1185            return item.ext in PODCAST_EPISODE_EXTENSIONS
1186        return False
1187
1188    def _set_available(self, available: bool) -> None:
1189        """Update the provider availability and notify listeners on change."""
1190        if self.available == available:
1191            return
1192        self.available = available
1193        self.mass.signal_event(EventType.PROVIDERS_UPDATED, data=self.mass.get_providers())
1194
1195    async def _process_item_async(
1196        self,
1197        item: FileSystemItem,
1198        prev_checksum: str | None,
1199        cur_filenames: set[str] | None = None,
1200        cue_stems: set[str] | None = None,
1201        prev_filenames: set[str] | None = None,
1202    ) -> bool:
1203        """
1204        Process a single item asynchronously.
1205
1206        :param item: The filesystem item to process.
1207        :param prev_checksum: Previous checksum from the database, or None for new items.
1208        :param cur_filenames: Set of current filenames being tracked (for CUE track IDs).
1209        :param cue_stems: Absolute paths (without extension) of CUE sheets in this scan,
1210            used to detect companion-CUE audio files without a filesystem stat.
1211        :param prev_filenames: The ids/paths the previous scan found, used to keep the
1212            ids of a CUE sheet that fails to parse.
1213        """
1214        try:
1215            self.logger.log(VERBOSE_LOG_LEVEL, "Processing: %s", item.relative_path)
1216
1217            if prev_checksum is not None:
1218                # the file changed on disk: drop cached artwork derived from it
1219                # (thumbnails, source bytes, palette) so re-read embedded art is
1220                # served fresh, for both reference forms of the image path
1221                await self.mass.metadata.invalidate_image_cache(
1222                    self.instance_id, item.relative_path
1223                )
1224                await self.mass.metadata.invalidate_image_cache(
1225                    self.instance_id, self._versioned_image_path(item.relative_path, prev_checksum)
1226                )
1227
1228            if item.ext in CUE_EXTENSIONS and self.media_content_type == "music":
1229                tracks = await self._cue.parse_tracks(item)
1230                for track in tracks:
1231                    track.favorite = False
1232                    await self.mass.music.tracks.add_item_to_library(
1233                        track, overwrite_existing=prev_checksum is not None
1234                    )
1235                    if cur_filenames is not None:
1236                        cur_filenames.add(track.item_id)
1237                return True
1238
1239            if item.ext in TRACK_EXTENSIONS and self.media_content_type == "music":
1240                if not self._sync_tracks:
1241                    return False
1242                # skip audio files that have a companion CUE sheet
1243                if cue_stems is not None and item.absolute_path.rsplit(".", 1)[0] in cue_stems:
1244                    return False
1245                tags = await async_parse_tags(item.absolute_path, item.file_size)
1246                track = await self._parse_track(item, tags)
1247                track.favorite = False  # TODO: implement favorite status based on rating ?
1248                await self.mass.music.tracks.add_item_to_library(
1249                    track, overwrite_existing=prev_checksum is not None
1250                )
1251                return True
1252
1253            if item.ext in AUDIOBOOK_EXTENSIONS and self.media_content_type == "audiobooks":
1254                tags = await async_parse_tags(item.absolute_path, item.file_size)
1255                try:
1256                    audiobook = await self._parse_audiobook(item, tags)
1257                except IsChapterFile:
1258                    return True
1259                await self.mass.music.audiobooks.add_item_to_library(
1260                    audiobook, overwrite_existing=prev_checksum is not None
1261                )
1262                return True
1263
1264            if item.ext in PODCAST_EPISODE_EXTENSIONS and self.media_content_type == "podcasts":
1265                tags = await async_parse_tags(item.absolute_path, item.file_size)
1266                episode = await self._parse_podcast_episode(item, tags)
1267                assert isinstance(episode.podcast, Podcast)
1268                await self.mass.music.podcasts.add_item_to_library(
1269                    episode.podcast, overwrite_existing=prev_checksum is not None
1270                )
1271                return True
1272
1273            if item.ext in PLAYLIST_EXTENSIONS and self.media_content_type == "music":
1274                if not self._sync_playlists:
1275                    return False
1276                playlist = await self.get_playlist(item.relative_path)
1277                await self.mass.music.playlists.add_item_to_library(
1278                    playlist, overwrite_existing=prev_checksum is not None
1279                )
1280                return True
1281
1282        except Exception as err:
1283            # we don't want the whole sync to crash on one file so we catch all exceptions here
1284            self.logger.error(
1285                "Error processing %s - %s",
1286                item.relative_path,
1287                str(err),
1288                exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
1289            )
1290            report_current_task_failure(f"Failed to process {item.relative_path}: {err}")
1291            # the file is still on the storage, so keep it in the scan result:
1292            # leaving it out makes the deletion step treat it as removed
1293            self._keep_failed_item(item, cur_filenames, prev_filenames)
1294        return False
1295
1296    def _keep_failed_item(
1297        self,
1298        item: FileSystemItem,
1299        cur_filenames: set[str] | None,
1300        prev_filenames: set[str] | None,
1301    ) -> None:
1302        """
1303        Keep an item that could not be processed in the scan result.
1304
1305        :param item: The item that failed to process.
1306        :param cur_filenames: Receives the ids/paths present in this scan.
1307        :param prev_filenames: The ids/paths the previous scan found.
1308        """
1309        if cur_filenames is None:
1310            return
1311        cur_filenames.add(item.relative_path)
1312        if (
1313            item.ext not in CUE_EXTENSIONS
1314            or self.media_content_type != "music"
1315            or not prev_filenames
1316        ):
1317            return
1318        # a CUE sheet stands in for one id per track it describes and those cannot be
1319        # rebuilt without parsing it, so carry over the ids of the previous scan
1320        cur_filenames.update(
1321            item_id
1322            for item_id in prev_filenames
1323            if (parsed := parse_cue_track_id(item_id)) and parsed[0] == item.relative_path
1324        )
1325
1326    async def _process_orphaned_albums_and_artists(self) -> None:
1327        """Process deletion of orphaned albums and artists."""
1328        assert self.mass.music.database
1329        # Remove albums without any tracks
1330        query = (
1331            f"SELECT item_id FROM {DB_TABLE_ALBUMS} "
1332            f"WHERE item_id not in ( SELECT album_id from {DB_TABLE_ALBUM_TRACKS}) "
1333            f"AND item_id in ( SELECT item_id from {DB_TABLE_PROVIDER_MAPPINGS} "
1334            f"WHERE provider_instance = '{self.instance_id}' and media_type = 'album' )"
1335        )
1336        for db_row in await self.mass.music.database.get_rows_from_query(
1337            query,
1338            limit=100000,
1339        ):
1340            await self.mass.music.albums.remove_item_from_library(db_row["item_id"])
1341
1342        # Remove artists without any tracks or albums
1343        query = (
1344            f"SELECT item_id FROM {DB_TABLE_ARTISTS} "
1345            f"WHERE item_id not in "
1346            f"( select artist_id from {DB_TABLE_TRACK_ARTISTS} "
1347            f"UNION SELECT artist_id from {DB_TABLE_ALBUM_ARTISTS} ) "
1348            f"AND item_id in ( SELECT item_id from {DB_TABLE_PROVIDER_MAPPINGS} "
1349            f"WHERE provider_instance = '{self.instance_id}' and media_type = 'artist' )"
1350        )
1351        for db_row in await self.mass.music.database.get_rows_from_query(
1352            query,
1353            limit=100000,
1354        ):
1355            await self.mass.music.artists.remove_item_from_library(db_row["item_id"])
1356
1357    async def _process_deletions(self, deleted_files: set[str]) -> None:
1358        """Process all deletions."""
1359        # process deleted tracks/playlists
1360        album_ids = set()
1361        artist_ids = set()
1362        for file_path in deleted_files:
1363            if parse_cue_track_id(file_path) is not None and self.media_content_type == "music":
1364                controller = self.mass.music.get_controller(MediaType.TRACK)
1365            elif "." not in file_path:
1366                continue
1367            else:
1368                _, ext = file_path.rsplit(".", 1)
1369                if ext in PODCAST_EPISODE_EXTENSIONS and self.media_content_type == "podcasts":
1370                    controller = self.mass.music.get_controller(MediaType.PODCAST_EPISODE)
1371                elif ext in AUDIOBOOK_EXTENSIONS and self.media_content_type == "audiobooks":
1372                    controller = self.mass.music.get_controller(MediaType.AUDIOBOOK)
1373                elif ext in PLAYLIST_EXTENSIONS and self.media_content_type == "music":
1374                    controller = self.mass.music.get_controller(MediaType.PLAYLIST)
1375                elif ext in TRACK_EXTENSIONS and self.media_content_type == "music":
1376                    controller = self.mass.music.get_controller(MediaType.TRACK)
1377                else:
1378                    # unsupported file extension?
1379                    continue
1380
1381            if library_item := await controller.get_library_item_by_prov_id(
1382                file_path, self.instance_id
1383            ):
1384                if is_track(library_item):
1385                    if library_item.album:
1386                        album_ids.add(library_item.album.item_id)
1387                        # need to fetch the library album to resolve the itemmapping
1388                        db_album = await self.mass.music.albums.get_library_item(
1389                            library_item.album.item_id
1390                        )
1391                        for artist in db_album.artists:
1392                            artist_ids.add(artist.item_id)
1393                    for artist in library_item.artists:
1394                        artist_ids.add(artist.item_id)
1395                await controller.remove_item_from_library(library_item.item_id)
1396        # check if any albums need to be cleaned up
1397        for album_id in album_ids:
1398            if not await self.mass.music.albums.tracks(album_id, "library"):
1399                await self.mass.music.albums.remove_item_from_library(album_id)
1400        # check if any artists need to be cleaned up
1401        for artist_id in artist_ids:
1402            artist_albums = await self.mass.music.artists.albums(artist_id, "library")
1403            artist_tracks = await self.mass.music.artists.tracks(artist_id, "library")
1404            if not (artist_albums or artist_tracks):
1405                await self.mass.music.artists.remove_item_from_library(artist_id)
1406
1407    async def _get_playlist_local_image(self, file_item: FileSystemItem) -> MediaItemImage | None:
1408        """Return a local image alongside the playlist file (matching basename) if any."""
1409        cache_key = f"playlist_image.{file_item.relative_path}"
1410        cached = await self.cache.get(
1411            key=cache_key,
1412            provider=self.instance_id,
1413            category=CACHE_CATEGORY_FOLDER_IMAGES,
1414            base_class=MediaItemImage,
1415        )
1416        if cached is not None:
1417            return cached[0] if cached else None
1418        try:
1419            folder_files = await self._scandir(file_item.relative_parent_path)
1420        except OSError, MusicAssistantError:
1421            return None
1422        target = file_item.name.lower()
1423        result: MediaItemImage | None = None
1424        for item in folder_files:
1425            if item.is_dir or not item.ext:
1426                continue
1427            if item.ext.lower() not in IMAGE_EXTENSIONS:
1428                continue
1429            if item.name.lower() != target:
1430                continue
1431            result = MediaItemImage(
1432                type=ImageType.THUMB,
1433                path=item.relative_path,
1434                provider=self.instance_id,
1435                remotely_accessible=False,
1436            )
1437            break
1438        await self.cache.set(
1439            key=cache_key,
1440            data=[result.to_dict()] if result is not None else [],
1441            provider=self.instance_id,
1442            category=CACHE_CATEGORY_FOLDER_IMAGES,
1443            expiration=120,
1444        )
1445        return result
1446
1447    async def _parse_playlist_line(self, line: str, playlist_path: str) -> Track | None:
1448        """Try to parse a track from a playlist line."""
1449        try:
1450            line = line.replace("file://", "").strip()
1451            # try to resolve the filename (both normal and url decoded):
1452            # - relative to the playlist folder (normpath resolves parent .. references)
1453            # - as-is: an absolute path, or relative to our base path
1454            # candidates stay relative so subclasses with virtual paths (cloud,
1455            # webdav) resolve them too, instead of leaking the server CWD
1456            for _line in (line, urllib.parse.unquote(line)):
1457                if playlist_path:
1458                    normalized = posixpath.normpath(f"{playlist_path}/{_line}")
1459                    with contextlib.suppress(FileNotFoundError, MediaNotFoundError):
1460                        file_item = await self.resolve(normalized)
1461                        tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
1462                        return await self._parse_track(file_item, tags)
1463                with contextlib.suppress(FileNotFoundError, MediaNotFoundError):
1464                    file_item = await self.resolve(_line)
1465                    tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
1466                    return await self._parse_track(file_item, tags)
1467            # all attempts failed
1468            raise MediaNotFoundError("Invalid path/uri")
1469
1470        except MusicAssistantError as err:
1471            self.logger.warning("Could not parse %s to track: %s", line, str(err))
1472
1473        return None
1474
1475    @staticmethod
1476    def _versioned_image_path(relative_path: str, checksum: str | None) -> str:
1477        """Append the file checksum so the image cache busts when the file is replaced."""
1478        if checksum:
1479            return f"{relative_path}?cs={checksum}"
1480        return relative_path
1481
1482    async def _parse_track(
1483        self, file_item: FileSystemItem, tags: AudioTags, full_album_metadata: bool = False
1484    ) -> Track:
1485        """Parse full track details from file tags."""
1486        # ruff: noqa: PLR0915
1487        name, version = parse_title_and_version(tags.title, tags.version)
1488        track = Track(
1489            item_id=file_item.relative_path,
1490            provider=self.instance_id,
1491            name=name,
1492            sort_name=tags.title_sort,
1493            version=version,
1494            provider_mappings={
1495                ProviderMapping(
1496                    item_id=file_item.relative_path,
1497                    provider_domain=self.domain,
1498                    provider_instance=self.instance_id,
1499                    audio_format=AudioFormat(
1500                        content_type=ContentType.try_parse(file_item.ext or tags.format),
1501                        sample_rate=tags.sample_rate,
1502                        bit_depth=tags.bits_per_sample,
1503                        channels=tags.channels,
1504                        bit_rate=tags.bit_rate,
1505                    ),
1506                    details=file_item.checksum,
1507                    in_library=True,
1508                )
1509            },
1510            disc_number=tags.disc or 0,
1511            track_number=tags.track or 0,
1512            date_added=(
1513                datetime.fromtimestamp(file_item.created_at, tz=UTC)
1514                if file_item.created_at
1515                else None
1516            ),
1517        )
1518
1519        if isrc_tags := tags.isrc:
1520            for isrsc in isrc_tags:
1521                track.external_ids.add((ExternalID.ISRC, isrsc))
1522
1523        if acoustid := tags.get("acoustid"):
1524            track.external_ids.add((ExternalID.ACOUSTID, acoustid))
1525
1526        # album
1527        album = track.album = (
1528            await self._parse_album(
1529                track_path=file_item.relative_path,
1530                track_tags=tags,
1531                track_created_at=file_item.created_at,
1532            )
1533            if tags.album
1534            else None
1535        )
1536
1537        # track artist(s)
1538        resolved_track_artists = await self._resolve_artists_with_mbids(
1539            tags.artists,
1540            tags.musicbrainz_artistids,
1541            tags.artist_sort_names,
1542            log_label="ARTISTS tag",
1543        )
1544        for name, mbid, sort_name in resolved_track_artists:
1545            # prefer the existing album artist object when it's the same artist
1546            if album_artist_match := self._match_album_artist(album, name, mbid):
1547                track.artists.append(album_artist_match)
1548                continue
1549            artist = await self._parse_artist(name, sort_name=sort_name, mbid=mbid)
1550            track.artists.append(artist)
1551
1552        # handle embedded cover image
1553        if tags.has_cover_image:
1554            # we do not actually embed the image in the metadata because that would consume too
1555            # much space and bandwidth. Instead we set the filename as value so the image can
1556            # be retrieved later in realtime.
1557            track.metadata.images = UniqueList(
1558                [
1559                    MediaItemImage(
1560                        type=ImageType.THUMB,
1561                        path=file_item.relative_path,
1562                        provider=self.instance_id,
1563                        remotely_accessible=False,
1564                    )
1565                ]
1566            )
1567
1568        # copy (embedded) album image from track (if the album itself doesn't have an image)
1569        if album and not album.image and track.image:
1570            album.metadata.images = UniqueList([track.image])
1571
1572        # parse other info
1573        track.duration = int(tags.duration or 0)
1574        track.metadata.genres = set(tags.genres)
1575        if tags.disc:
1576            track.disc_number = tags.disc
1577        if tags.track:
1578            track.track_number = tags.track
1579        track.metadata.copyright = tags.get("copyright")
1580        track.metadata.lyrics = tags.lyrics
1581        track.metadata.grouping = tags.get("grouping")
1582        track.metadata.description = tags.get("comment")
1583        explicit_tag = tags.get("itunesadvisory")
1584        if explicit_tag is not None:
1585            track.metadata.explicit = explicit_tag == "1"
1586        if recording_mbid := clean_mbid(tags.musicbrainz_recordingid, tags.filename):
1587            track.mbid = recording_mbid
1588
1589        # handle (optional) loudness measurement tag(s)
1590        if tags.track_loudness is not None:
1591            self.mass.create_task(
1592                self.mass.streams.audio_analysis.set_track_loudness(
1593                    track.item_id,
1594                    self.instance_id,
1595                    tags.track_loudness,
1596                    tags.track_album_loudness,
1597                )
1598            )
1599
1600        # possible lrclib metadata
1601        # synced lyrics are saved as "filename.lrc" by lrcget alongside
1602        # the actual file location - just change the file extension
1603        assert file_item.ext is not None  # for type checking
1604        lrc_path = f"{file_item.relative_path.removesuffix(file_item.ext)}lrc"
1605        if await self.exists(lrc_path):
1606            try:
1607                raw = await self._read_file(lrc_path)
1608                track.metadata.lrc_lyrics = raw.decode("utf-8")
1609            except Exception as err:
1610                self.logger.warning(
1611                    "Failed to read lyrics file %s: %s",
1612                    lrc_path,
1613                    str(err),
1614                )
1615        elif syn_lyrics := tags.synchronized_lyrics:
1616            track.metadata.lrc_lyrics = lyrics.convert_to_lrc_lyrics(syn_lyrics)
1617
1618        return track
1619
1620    async def _resolve_artists_with_mbids(
1621        self,
1622        parsed_names: tuple[str, ...],
1623        mbids: tuple[str, ...],
1624        sort_names: tuple[str, ...],
1625        log_label: str,
1626    ) -> list[tuple[str, str | None, str | None]]:
1627        """
1628        Return ``(name, mbid, sort_name)`` triples for a track's or album's artists.
1629
1630        When the parsed name count and the MBID count disagree, canonical names
1631        are looked up from MusicBrainz; otherwise the tag-parsed names are used.
1632
1633        :param parsed_names: Tag-parsed artist names.
1634        :param mbids: MusicBrainz artist IDs from the tag.
1635        :param sort_names: Sort names from the corresponding *sort tag.
1636        :param log_label: Tag name used in warning messages (e.g. "ARTISTS tag").
1637        """
1638
1639        def _sort_name(index: int) -> str | None:
1640            return sort_names[index] if index < len(sort_names) else None
1641
1642        def _from_tags() -> list[tuple[str, str | None, str | None]]:
1643            return [
1644                (
1645                    name,
1646                    mbids[i] if i < len(mbids) else None,
1647                    _sort_name(i),
1648                )
1649                for i, name in enumerate(parsed_names)
1650            ]
1651
1652        if not mbids or len(parsed_names) == len(mbids):
1653            return _from_tags()
1654
1655        mb_provider = cast("MusicbrainzProvider | None", self.mass.get_provider("musicbrainz"))
1656        if mb_provider is None:
1657            self.logger.warning(
1658                "%s count (%d) doesn't match MBID count (%d) and MusicBrainz "
1659                "provider is not loaded; using tag-parsed names: %s",
1660                log_label,
1661                len(parsed_names),
1662                len(mbids),
1663                parsed_names,
1664            )
1665            return _from_tags()
1666
1667        mb_results = await mb_provider.resolve_artists_from_mbids(mbids)
1668        # counts disagree, so positional fallback to a tag name is unreliable;
1669        # drop any MBID whose lookup failed (already logged per-MBID)
1670        resolved: list[tuple[str, str | None, str | None]] = [
1671            mb_result for mb_result in mb_results if mb_result is not None
1672        ]
1673        if not resolved:
1674            self.logger.warning(
1675                "%s count (%d) didn't match MBID count (%d) and every MusicBrainz "
1676                "lookup failed; falling back to tag-parsed names: %s",
1677                log_label,
1678                len(parsed_names),
1679                len(mbids),
1680                parsed_names,
1681            )
1682            return _from_tags()
1683        self.logger.info(
1684            "%s count (%d) didn't match MBID count (%d); resolved canonical names "
1685            "via MusicBrainz: %s",
1686            log_label,
1687            len(parsed_names),
1688            len(mbids),
1689            [r[0] for r in resolved],
1690        )
1691        return resolved
1692
1693    def _match_album_artist(
1694        self, album: Album | None, name: str, mbid: str | None
1695    ) -> Artist | ItemMapping | None:
1696        """
1697        Return an existing album artist representing the same artist, if any.
1698
1699        Matches on MusicBrainz ID when available (names may differ when only one
1700        side was resolved against MusicBrainz), otherwise on exact name.
1701
1702        :param album: The track's album, if known.
1703        :param name: Resolved track-artist name.
1704        :param mbid: Resolved track-artist MusicBrainz ID, if any.
1705        """
1706        if not album:
1707            return None
1708        return next(
1709            (x for x in album.artists if (mbid and x.mbid == mbid) or x.name == name),
1710            None,
1711        )
1712
1713    async def _parse_artist(
1714        self,
1715        name: str,
1716        album_dir: str | None = None,
1717        sort_name: str | None = None,
1718        mbid: str | None = None,
1719        artist_path: str | None = None,
1720    ) -> Artist:
1721        """Parse full (album) Artist."""
1722        if not artist_path:
1723            # we need to hunt for the artist (metadata) path on disk
1724            # this can either be relative to the album path or at root level
1725            # check if we have an artist folder for this artist at root level
1726            safe_artist_name = create_safe_string(name, lowercase=False, replace_space=False)
1727            if await self.exists(name):
1728                artist_path = name
1729            elif await self.exists(safe_artist_name):
1730                artist_path = safe_artist_name
1731            elif album_dir and (foldermatch := get_artist_dir(name, album_dir=album_dir)):
1732                # try to find (album)artist folder based on album path
1733                artist_path = foldermatch
1734            else:
1735                # check if we have an existing item to retrieve the artist path
1736                async for item in self.mass.music.artists.iter_library_items(
1737                    search=name, provider=self.instance_id
1738                ):
1739                    if not compare_strings(name, item.name):
1740                        continue
1741                    for prov_mapping in item.provider_mappings:
1742                        if prov_mapping.provider_instance != self.instance_id:
1743                            continue
1744                        if prov_mapping.url:
1745                            artist_path = prov_mapping.url
1746                            break
1747                    if artist_path:
1748                        break
1749
1750        # prefer (short lived) cache for a bit more speed
1751        if artist_path and (
1752            cache := await self.cache.get(
1753                key=artist_path,
1754                provider=self.instance_id,
1755                category=CACHE_CATEGORY_ARTIST_INFO,
1756                base_class=Artist,
1757            )
1758        ):
1759            return cache  # type: ignore[no-any-return]
1760
1761        prov_artist_id = artist_path or name
1762        artist = Artist(
1763            item_id=prov_artist_id,
1764            provider=self.instance_id,
1765            name=name,
1766            sort_name=sort_name,
1767            provider_mappings={
1768                ProviderMapping(
1769                    item_id=prov_artist_id,
1770                    provider_domain=self.domain,
1771                    provider_instance=self.instance_id,
1772                    url=artist_path,
1773                    in_library=True,
1774                )
1775            },
1776        )
1777        if mbid := clean_mbid(mbid, f"tags of artist {name}"):
1778            artist.mbid = mbid
1779        if not artist_path or not await self.exists(artist_path):
1780            return artist
1781
1782        # grab additional metadata within the Artist's folder
1783        nfo_file = os.path.join(artist_path, "artist.nfo")
1784        if await self.exists(nfo_file):
1785            # found NFO file with metadata
1786            # https://kodi.wiki/view/NFO_files/Artists
1787            try:
1788                data = (await self._read_file(nfo_file)).decode("utf-8")
1789                info = await asyncio.to_thread(xmltodict.parse, data)
1790                info = info["artist"]
1791                artist.name = info.get("title", info.get("name", name))
1792                if sort_name := info.get("sortname"):
1793                    artist.sort_name = sort_name
1794                if mbid := clean_mbid(info.get("musicbrainzartistid"), nfo_file):
1795                    artist.mbid = mbid
1796                if description := info.get("biography"):
1797                    artist.metadata.description = description
1798                if genre := info.get("genre"):
1799                    artist.metadata.genres = set(split_items(genre))
1800            except (ExpatError, KeyError) as err:
1801                self.logger.warning(
1802                    "Failed to parse artist NFO file %s: %s",
1803                    nfo_file,
1804                    str(err),
1805                )
1806        # find local images
1807        if images := await self._get_local_images(artist_path, extra_thumb_names=("artist",)):
1808            artist.metadata.images = UniqueList(images)
1809
1810        await self.cache.set(
1811            key=artist_path,
1812            data=artist.to_dict(),
1813            provider=self.instance_id,
1814            category=CACHE_CATEGORY_ARTIST_INFO,
1815            expiration=120,
1816        )
1817
1818        return artist
1819
1820    async def _parse_audiobook(self, file_item: FileSystemItem, tags: AudioTags) -> Audiobook:
1821        """
1822        Parse Audiobook details from file tags.
1823
1824        Audiobooks can be single files with embedded chapters or multiple files per folder.
1825        Only the first file (by track number or alphabetically) is processed as the audiobook.
1826        """
1827        # Skip files that aren't the first chapter.
1828        # A file carrying its own embedded chapter markers is a standalone audiobook,
1829        # so it should never be treated as a chapter file of another book.
1830        track_tag = tags.tags.get("track")
1831        if track_tag:
1832            track_num = try_parse_int(str(track_tag).split("/")[0], None)
1833            if track_num and track_num > 1 and not tags.chapters:
1834                raise IsChapterFile
1835        elif not tags.chapters:
1836            # No track tag and no embedded chapters -
1837            # assume part of a multi-file audiobook, only process the first file alphabetically
1838            items = await self._scandir(file_item.relative_parent_path)
1839            # Sort by filename for alphabetical ordering
1840            items.sort(key=lambda x: x.filename.lower())
1841            for item in items:
1842                if item.is_dir or item.ext not in AUDIOBOOK_EXTENSIONS:
1843                    continue
1844                if item.absolute_path != file_item.absolute_path:
1845                    raise IsChapterFile
1846                break
1847
1848        # For multi-file audiobooks, album tag is the book name, title is the chapter name
1849        if tags.album:
1850            book_name = tags.album
1851            sort_name = tags.album_sort
1852        elif (title := tags.tags.get("title")) and tags.track is None:
1853            book_name = title
1854            sort_name = tags.title_sort
1855        else:
1856            # file(s) without tags, use foldername
1857            book_name = file_item.parent_name
1858            sort_name = None
1859
1860        # collect all chapters
1861        total_duration, chapters = await self._get_chapters_for_audiobook(file_item, tags)
1862
1863        audio_book = Audiobook(
1864            item_id=file_item.relative_path,
1865            provider=self.instance_id,
1866            name=book_name,
1867            sort_name=sort_name,
1868            version=tags.version,
1869            duration=total_duration or int(tags.duration or 0),
1870            provider_mappings={
1871                ProviderMapping(
1872                    item_id=file_item.relative_path,
1873                    provider_domain=self.domain,
1874                    provider_instance=self.instance_id,
1875                    audio_format=AudioFormat(
1876                        content_type=ContentType.try_parse(file_item.ext or tags.format),
1877                        sample_rate=tags.sample_rate,
1878                        bit_depth=tags.bits_per_sample,
1879                        channels=tags.channels,
1880                        bit_rate=tags.bit_rate,
1881                    ),
1882                    details=file_item.checksum,
1883                    in_library=True,
1884                )
1885            },
1886        )
1887        audio_book.metadata.chapters = chapters
1888
1889        # handle embedded cover image
1890        if tags.has_cover_image:
1891            # we do not actually embed the image in the metadata because that would consume too
1892            # much space and bandwidth. Instead we set the filename as value so the image can
1893            # be retrieved later in realtime.
1894            audio_book.metadata.add_image(
1895                MediaItemImage(
1896                    type=ImageType.THUMB,
1897                    path=self._versioned_image_path(file_item.relative_path, file_item.checksum),
1898                    provider=self.instance_id,
1899                    remotely_accessible=False,
1900                )
1901            )
1902
1903        # parse other info
1904        audio_book.authors.set(tags.writers or tags.album_artists or tags.artists)
1905        audio_book.metadata.genres = (
1906            set(tags.genres) if tags.genres else {DEFAULT_AUDIOBOOK_PODCAST_GENRE}
1907        )
1908        audio_book.metadata.copyright = tags.get("copyright")
1909        audio_book.metadata.lyrics = tags.lyrics
1910        audio_book.metadata.description = tags.get("comment")
1911        explicit_tag = tags.get("itunesadvisory")
1912        if explicit_tag is not None:
1913            audio_book.metadata.explicit = explicit_tag == "1"
1914        if recording_mbid := clean_mbid(tags.musicbrainz_recordingid, tags.filename):
1915            audio_book.mbid = recording_mbid
1916
1917        # try to fetch additional metadata from the folder
1918        if not audio_book.image or not audio_book.metadata.description:
1919            # try to get an image by traversing files in the same folder
1920            for _item in await self._scandir(file_item.relative_parent_path):
1921                if "." not in _item.relative_path or _item.is_dir:
1922                    continue
1923                if _item.ext in IMAGE_EXTENSIONS and not audio_book.image:
1924                    audio_book.metadata.add_image(
1925                        MediaItemImage(
1926                            type=ImageType.THUMB,
1927                            path=self._versioned_image_path(_item.relative_path, _item.checksum),
1928                            provider=self.instance_id,
1929                            remotely_accessible=False,
1930                        )
1931                    )
1932                if _item.ext == "txt" and not audio_book.metadata.description:
1933                    # try to parse a description from a text file
1934                    try:
1935                        raw = await self._read_file(_item.relative_path)
1936                        audio_book.metadata.description = raw.decode("utf-8")
1937                    except Exception as err:
1938                        self.logger.warning(
1939                            "Could not read description from file %s: %s",
1940                            _item.relative_path,
1941                            str(err),
1942                        )
1943
1944        # handle (optional) loudness measurement tag(s)
1945        if tags.track_loudness is not None:
1946            self.mass.create_task(
1947                self.mass.streams.audio_analysis.set_track_loudness(
1948                    audio_book.item_id,
1949                    self.instance_id,
1950                    tags.track_loudness,
1951                    tags.track_album_loudness,
1952                    media_type=MediaType.AUDIOBOOK,
1953                )
1954            )
1955        return audio_book
1956
1957    async def _parse_podcast_episode(
1958        self, file_item: FileSystemItem, tags: AudioTags
1959    ) -> PodcastEpisode:
1960        """Parse full PodcastEpisode details from file tags."""
1961        # ruff: noqa: PLR0915
1962        podcast_name = tags.album or file_item.parent_name
1963        podcast_path = file_item.relative_parent_path
1964        episode = PodcastEpisode(
1965            item_id=file_item.relative_path,
1966            provider=self.instance_id,
1967            name=tags.title,
1968            sort_name=tags.title_sort,
1969            provider_mappings={
1970                ProviderMapping(
1971                    item_id=file_item.relative_path,
1972                    provider_domain=self.domain,
1973                    provider_instance=self.instance_id,
1974                    audio_format=AudioFormat(
1975                        content_type=ContentType.try_parse(file_item.ext or tags.format),
1976                        sample_rate=tags.sample_rate,
1977                        bit_depth=tags.bits_per_sample,
1978                        channels=tags.channels,
1979                        bit_rate=tags.bit_rate,
1980                    ),
1981                    details=file_item.checksum,
1982                    in_library=True,
1983                )
1984            },
1985            position=tags.track or 0,
1986            duration=try_parse_int(tags.duration) or 0,
1987            podcast=Podcast(
1988                item_id=podcast_path,
1989                provider=self.instance_id,
1990                name=podcast_name,
1991                sort_name=tags.album_sort,
1992                publisher=tags.tags.get("publisher"),
1993                provider_mappings={
1994                    ProviderMapping(
1995                        item_id=podcast_path,
1996                        provider_domain=self.domain,
1997                        provider_instance=self.instance_id,
1998                        in_library=True,
1999                    )
2000                },
2001            ),
2002        )
2003        # handle embedded cover image
2004        if tags.has_cover_image:
2005            # we do not actually embed the image in the metadata because that would consume too
2006            # much space and bandwidth. Instead we set the filename as value so the image can
2007            # be retrieved later in realtime.
2008            episode.metadata.add_image(
2009                MediaItemImage(
2010                    type=ImageType.THUMB,
2011                    path=file_item.relative_path,
2012                    provider=self.instance_id,
2013                    remotely_accessible=False,
2014                )
2015            )
2016        # parse other info
2017        episode.metadata.genres = (
2018            set(tags.genres) if tags.genres else {DEFAULT_AUDIOBOOK_PODCAST_GENRE}
2019        )
2020        episode.metadata.copyright = tags.get("copyright")
2021        episode.metadata.lyrics = tags.lyrics
2022        episode.metadata.description = tags.get("comment")
2023        explicit_tag = tags.get("itunesadvisory")
2024        if explicit_tag is not None:
2025            episode.metadata.explicit = explicit_tag == "1"
2026
2027        # handle (optional) chapters
2028        if tags.chapters:
2029            episode.metadata.chapters = [
2030                MediaItemChapter(
2031                    position=chapter.chapter_id,
2032                    name=chapter.title or f"Chapter {chapter.chapter_id}",
2033                    start=chapter.position_start,
2034                    end=chapter.position_end,
2035                )
2036                for chapter in tags.chapters
2037            ]
2038
2039        # try to fetch additional Podcast metadata from the folder
2040        assert isinstance(episode.podcast, Podcast)
2041        if images := await self._get_local_images(file_item.relative_parent_path):
2042            episode.podcast.metadata.images = images
2043        if metadata := await self._get_podcast_metadata(file_item.relative_parent_path):
2044            if title := metadata.get("title"):
2045                episode.podcast.name = title
2046            if sort_name := metadata.get("sorttitle"):
2047                episode.podcast.sort_name = sort_name
2048            if description := metadata.get("description"):
2049                episode.podcast.metadata.description = description
2050            if genres := metadata.get("genres"):
2051                episode.podcast.metadata.genres = set(genres)
2052            if publisher := metadata.get("publisher"):
2053                episode.podcast.publisher = publisher
2054            if image := metadata.get("imageURL"):
2055                episode.podcast.metadata.add_image(
2056                    MediaItemImage(
2057                        type=ImageType.THUMB,
2058                        path=image,
2059                        provider=self.instance_id,
2060                        remotely_accessible=True,
2061                    )
2062                )
2063        # copy (embedded) image from episode (or vice versa)
2064        if not episode.podcast.image and episode.image:
2065            episode.podcast.metadata.add_image(episode.image)
2066        elif not episode.image and episode.podcast.image:
2067            episode.metadata.add_image(episode.podcast.image)
2068        # ensure podcast has a default genre if none set
2069        if not episode.podcast.metadata.genres:
2070            episode.podcast.metadata.genres = {DEFAULT_AUDIOBOOK_PODCAST_GENRE}
2071
2072        # handle (optional) loudness measurement tag(s)
2073        if tags.track_loudness is not None:
2074            self.mass.create_task(
2075                self.mass.streams.audio_analysis.set_track_loudness(
2076                    episode.item_id,
2077                    self.instance_id,
2078                    tags.track_loudness,
2079                    tags.track_album_loudness,
2080                    media_type=MediaType.PODCAST_EPISODE,
2081                )
2082            )
2083        return episode
2084
2085    async def _parse_sound_effect(self, file_item: FileSystemItem, tags: AudioTags) -> SoundEffect:
2086        """Parse full sound effect details from file tags."""
2087        sound_effect = SoundEffect(
2088            item_id=file_item.relative_path,
2089            provider=self.instance_id,
2090            name=tags.title,
2091            sort_name=tags.title_sort,
2092            duration=int(tags.duration or 0),
2093            provider_mappings={
2094                ProviderMapping(
2095                    item_id=file_item.relative_path,
2096                    provider_domain=self.domain,
2097                    provider_instance=self.instance_id,
2098                    audio_format=AudioFormat(
2099                        content_type=ContentType.try_parse(file_item.ext or tags.format),
2100                        sample_rate=tags.sample_rate,
2101                        bit_depth=tags.bits_per_sample,
2102                        channels=tags.channels,
2103                        bit_rate=tags.bit_rate,
2104                    ),
2105                    details=file_item.checksum,
2106                    in_library=True,
2107                )
2108            },
2109        )
2110        sound_effect.metadata.description = tags.get("comment")
2111        # handle embedded cover image
2112        if tags.has_cover_image:
2113            # we do not actually embed the image in the metadata because that would consume too
2114            # much space and bandwidth. Instead we set the filename as value so the image can
2115            # be retrieved later in realtime.
2116            sound_effect.metadata.add_image(
2117                MediaItemImage(
2118                    type=ImageType.THUMB,
2119                    path=file_item.relative_path,
2120                    provider=self.instance_id,
2121                    remotely_accessible=False,
2122                )
2123            )
2124        return sound_effect
2125
2126    async def _get_or_parse_sound_effect(self, file_item: FileSystemItem) -> SoundEffect:
2127        """Return the (cached) SoundEffect for the given file, parsing tags when needed."""
2128        cache_key = f"sound_effect.{file_item.relative_path}"
2129        cached_data: SoundEffect | None = await self.cache.get(
2130            cache_key,
2131            provider=self.instance_id,
2132            checksum=file_item.checksum,
2133            category=CACHE_CATEGORY_SOUND_EFFECTS,
2134            base_class=SoundEffect,
2135        )
2136        if cached_data is not None:
2137            return cached_data
2138        tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
2139        sound_effect = await self._parse_sound_effect(file_item, tags)
2140        await self.cache.set(
2141            cache_key,
2142            sound_effect.to_dict(),
2143            expiration=3600 * 24 * 365,  # File timestamp checksum handles invalidation
2144            provider=self.instance_id,
2145            checksum=file_item.checksum,
2146            category=CACHE_CATEGORY_SOUND_EFFECTS,
2147        )
2148        return sound_effect
2149
2150    async def _parse_album(
2151        self, track_path: str, track_tags: AudioTags, track_created_at: int | None = None
2152    ) -> Album:
2153        """
2154        Parse Album metadata from Track tags.
2155
2156        :param track_path: Path to the track file.
2157        :param track_tags: Audio tags from the track.
2158        :param track_created_at: Creation timestamp of the track file (Unix epoch).
2159        """
2160        assert track_tags.album
2161        # work out if we have an album and/or disc folder
2162        # track_dir is the folder level where the tracks are located
2163        # this may be a separate disc folder (Disc 1, Disc 2 etc) underneath the album folder
2164        # or this is an album folder with the disc attached
2165        track_dir = os.path.dirname(track_path)
2166        album_dir = get_album_dir(track_dir, track_tags.album)
2167
2168        if album_dir and (
2169            cache := await self.cache.get(
2170                key=album_dir,
2171                provider=self.instance_id,
2172                category=CACHE_CATEGORY_ALBUM_INFO,
2173                base_class=Album,
2174            )
2175        ):
2176            return cache  # type: ignore[no-any-return]
2177
2178        # album artist(s)
2179        album_artists: UniqueList[Artist | ItemMapping] = UniqueList()
2180        if track_tags.album_artists:
2181            resolved_album_artists = await self._resolve_artists_with_mbids(
2182                track_tags.album_artists,
2183                track_tags.musicbrainz_albumartistids,
2184                track_tags.album_artist_sort_names,
2185                log_label="ALBUMARTIST tag",
2186            )
2187            for name, mbid, sort_name in resolved_album_artists:
2188                artist = await self._parse_artist(
2189                    name, album_dir=album_dir, sort_name=sort_name, mbid=mbid
2190                )
2191                album_artists.append(artist)
2192        else:
2193            # album artist tag is missing, determine fallback
2194            fallback_action = self.config.get_value(CONF_ENTRY_MISSING_ALBUM_ARTIST.key)
2195            if fallback_action == "folder_name" and album_dir:
2196                possible_artist_folder = os.path.dirname(album_dir)
2197                self.logger.warning(
2198                    "%s is missing ID3 tag [albumartist], using foldername %s as fallback",
2199                    track_path,
2200                    possible_artist_folder,
2201                )
2202                album_artist_str = possible_artist_folder.rsplit(os.sep)[-1]
2203                album_artists = UniqueList(
2204                    [await self._parse_artist(name=album_artist_str, album_dir=album_dir)]
2205                )
2206            # fallback to track artists (if defined by user)
2207            elif fallback_action == "track_artist":
2208                self.logger.warning(
2209                    "%s is missing ID3 tag [albumartist], using track artist(s) as fallback",
2210                    track_path,
2211                )
2212                album_artists = UniqueList(
2213                    [
2214                        await self._parse_artist(name=track_artist_str, album_dir=album_dir)
2215                        for track_artist_str in track_tags.artists
2216                    ]
2217                )
2218            # all other: fallback to various artists
2219            else:
2220                self.logger.warning(
2221                    "%s is missing ID3 tag [albumartist], using %s as fallback",
2222                    track_path,
2223                    VARIOUS_ARTISTS_NAME,
2224                )
2225                album_artists = UniqueList(
2226                    [await self._parse_artist(name=VARIOUS_ARTISTS_NAME, mbid=VARIOUS_ARTISTS_MBID)]
2227                )
2228
2229        if album_dir:  # noqa: SIM108
2230            # prefer the path as id
2231            item_id = album_dir
2232        else:
2233            # create fake item_id based on artist + album
2234            item_id = album_artists[0].name + os.sep + track_tags.album
2235
2236        name, version = parse_title_and_version(track_tags.album)
2237        album = Album(
2238            item_id=item_id,
2239            provider=self.instance_id,
2240            name=name,
2241            version=version,
2242            sort_name=track_tags.album_sort,
2243            artists=album_artists,
2244            provider_mappings={
2245                ProviderMapping(
2246                    item_id=item_id,
2247                    provider_domain=self.domain,
2248                    provider_instance=self.instance_id,
2249                    url=album_dir,
2250                    in_library=True,
2251                )
2252            },
2253            date_added=(
2254                datetime.fromtimestamp(track_created_at, tz=UTC) if track_created_at else None
2255            ),
2256        )
2257        if track_tags.barcode:
2258            album.external_ids.add((ExternalID.BARCODE, track_tags.barcode))
2259
2260        if album_mbid := clean_mbid(track_tags.musicbrainz_albumid, track_tags.filename):
2261            album.mbid = album_mbid
2262        if releasegroup_mbid := clean_mbid(
2263            track_tags.musicbrainz_releasegroupid, track_tags.filename
2264        ):
2265            album.add_external_id(ExternalID.MB_RELEASEGROUP, releasegroup_mbid)
2266        if track_tags.year:
2267            album.year = track_tags.year
2268        album.album_type = track_tags.album_type
2269
2270        # hunt for additional metadata and images in the folder structure
2271        if not album_dir:
2272            return album
2273
2274        for folder_path in (track_dir, album_dir):
2275            if not folder_path or not await self.exists(folder_path):
2276                continue
2277            nfo_file = os.path.join(folder_path, "album.nfo")
2278            if await self.exists(nfo_file):
2279                # found NFO file with metadata
2280                # https://kodi.wiki/view/NFO_files/Artists
2281                try:
2282                    data = (await self._read_file(nfo_file)).decode("utf-8")
2283                    info = await asyncio.to_thread(xmltodict.parse, data)
2284                    parse_album_nfo(album, info["album"], nfo_file)
2285                except (ExpatError, KeyError) as err:
2286                    self.logger.warning(
2287                        "Failed to parse album NFO file %s: %s",
2288                        nfo_file,
2289                        str(err),
2290                    )
2291
2292            # find local images
2293            if images := await self._get_local_images(folder_path, extra_thumb_names=("album",)):
2294                if album.metadata.images is None:
2295                    album.metadata.images = UniqueList(images)
2296                else:
2297                    album.metadata.images += images
2298
2299        await self.cache.set(
2300            key=album_dir,
2301            data=album.to_dict(),
2302            provider=self.instance_id,
2303            category=CACHE_CATEGORY_ALBUM_INFO,
2304            expiration=120,
2305        )
2306        return album
2307
2308    async def _get_local_images(
2309        self, folder: str, extra_thumb_names: tuple[str, ...] | None = None
2310    ) -> UniqueList[MediaItemImage]:
2311        """Return local images found in a given folderpath."""
2312        if (
2313            cache := await self.cache.get(
2314                key=folder,
2315                provider=self.instance_id,
2316                category=CACHE_CATEGORY_FOLDER_IMAGES,
2317                base_class=MediaItemImage,
2318            )
2319        ) is not None:
2320            return UniqueList(cache)
2321        if extra_thumb_names is None:
2322            extra_thumb_names = ()
2323        images: UniqueList[MediaItemImage] = UniqueList()
2324        folder_files = await self._scandir(folder)
2325        for item in folder_files:
2326            if "." not in item.relative_path or item.is_dir or not item.ext:
2327                continue
2328            if item.ext.lower() not in IMAGE_EXTENSIONS:
2329                continue
2330            # try match on filename = one of our imagetypes
2331            if item.name.lower() in ImageType:
2332                images.append(
2333                    MediaItemImage(
2334                        type=ImageType(item.name),
2335                        path=item.relative_path,
2336                        provider=self.instance_id,
2337                        remotely_accessible=False,
2338                    )
2339                )
2340
2341        # try alternative names for thumbs
2342        extra_thumb_names = ("folder", "cover", *extra_thumb_names)
2343        for item in folder_files:
2344            if "." not in item.relative_path or item.is_dir or not item.ext:
2345                continue
2346            if item.ext.lower() not in IMAGE_EXTENSIONS:
2347                continue
2348            if item.name.lower() not in extra_thumb_names:
2349                continue
2350            images.append(
2351                MediaItemImage(
2352                    type=ImageType.THUMB,
2353                    path=item.relative_path,
2354                    provider=self.instance_id,
2355                    remotely_accessible=False,
2356                )
2357            )
2358
2359        await self.cache.set(
2360            key=folder,
2361            data=[img.to_dict() for img in images],
2362            provider=self.instance_id,
2363            category=CACHE_CATEGORY_FOLDER_IMAGES,
2364            expiration=120,
2365        )
2366        return images
2367
2368    async def _get_stream_details_for_track(self, item_id: str) -> StreamDetails:
2369        """Return the streamdetails for a track/song."""
2370        if parse_cue_track_id(item_id) is not None:
2371            return await self._cue.get_stream_details(item_id)
2372
2373        library_item = await self.mass.music.tracks.get_library_item_by_prov_id(
2374            item_id, self.instance_id
2375        )
2376        if library_item is None:
2377            # this could be a file that has just been added, try parsing it
2378            file_item = await self.resolve(item_id)
2379            tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
2380            if not (library_item := await self._parse_track(file_item, tags)):
2381                msg = f"Item not found: {item_id}"
2382                raise MediaNotFoundError(msg)
2383
2384        prov_mapping = next(x for x in library_item.provider_mappings if x.item_id == item_id)
2385        file_item = await self.resolve(item_id)
2386
2387        return StreamDetails(
2388            provider=self.instance_id,
2389            item_id=item_id,
2390            audio_format=prov_mapping.audio_format,
2391            media_type=MediaType.TRACK,
2392            stream_type=StreamType.LOCAL_FILE,
2393            duration=library_item.duration,
2394            size=file_item.file_size,
2395            data=file_item,
2396            path=file_item.absolute_path,
2397            can_seek=True,
2398            allow_seek=True,
2399        )
2400
2401    async def _get_stream_details_for_podcast_episode(self, item_id: str) -> StreamDetails:
2402        """Return the streamdetails for a podcast episode."""
2403        # podcasts episodes are never stored in the library so we need to parse the file
2404        file_item = await self.resolve(item_id)
2405        tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
2406        return StreamDetails(
2407            provider=self.instance_id,
2408            item_id=item_id,
2409            audio_format=AudioFormat(
2410                content_type=ContentType.try_parse(file_item.ext or tags.format),
2411                sample_rate=tags.sample_rate,
2412                bit_depth=tags.bits_per_sample,
2413                channels=tags.channels,
2414                bit_rate=tags.bit_rate,
2415            ),
2416            media_type=MediaType.PODCAST_EPISODE,
2417            stream_type=StreamType.LOCAL_FILE,
2418            duration=try_parse_int(tags.duration or 0),
2419            size=file_item.file_size,
2420            data=file_item,
2421            path=file_item.absolute_path,
2422            allow_seek=True,
2423            can_seek=True,
2424        )
2425
2426    async def _get_stream_details_for_sound_effect(self, item_id: str) -> StreamDetails:
2427        """Return the streamdetails for a sound effect."""
2428        # sound effects are never stored in the library so we parse the file,
2429        # served from cache unless the file changed on disk
2430        file_item = await self.resolve(item_id)
2431        sound_effect = await self._get_or_parse_sound_effect(file_item)
2432        prov_mapping = next(x for x in sound_effect.provider_mappings if x.item_id == item_id)
2433        return StreamDetails(
2434            provider=self.instance_id,
2435            item_id=item_id,
2436            audio_format=prov_mapping.audio_format,
2437            media_type=MediaType.SOUND_EFFECT,
2438            stream_type=StreamType.LOCAL_FILE,
2439            duration=sound_effect.duration,
2440            size=file_item.file_size,
2441            data=file_item,
2442            path=file_item.absolute_path,
2443            allow_seek=True,
2444            can_seek=True,
2445        )
2446
2447    async def _get_stream_details_for_audiobook(self, item_id: str) -> StreamDetails:
2448        """Return the streamdetails for an audiobook."""
2449        library_item = await self.mass.music.audiobooks.get_library_item_by_prov_id(
2450            item_id, self.instance_id
2451        )
2452        if library_item is None:
2453            # this could be a file that has just been added, try parsing it
2454            file_item = await self.resolve(item_id)
2455            tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
2456            if not (library_item := await self._parse_audiobook(file_item, tags)):
2457                msg = f"Item not found: {item_id}"
2458                raise MediaNotFoundError(msg)
2459
2460        prov_mapping = next(x for x in library_item.provider_mappings if x.item_id == item_id)
2461        file_item = await self.resolve(item_id)
2462        duration = library_item.duration
2463        file_based_chapters: list[tuple[str, float]] | None = await self.cache.get(
2464            key=file_item.relative_path,
2465            provider=self.instance_id,
2466            category=CACHE_CATEGORY_AUDIOBOOK_CHAPTERS,
2467        )
2468        if file_based_chapters is None:
2469            # no cache available for this audiobook, we need to parse the chapters
2470            tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
2471            await self._parse_audiobook(file_item, tags)
2472            file_based_chapters = await self.cache.get(
2473                key=file_item.relative_path,
2474                provider=self.instance_id,
2475                category=CACHE_CATEGORY_AUDIOBOOK_CHAPTERS,
2476            )
2477
2478        if file_based_chapters:
2479            # this is a multi-file audiobook
2480            return StreamDetails(
2481                provider=self.instance_id,
2482                item_id=item_id,
2483                audio_format=prov_mapping.audio_format,
2484                media_type=MediaType.AUDIOBOOK,
2485                stream_type=StreamType.LOCAL_FILE,
2486                duration=duration,
2487                path=[
2488                    MultiPartPath(
2489                        path=self._get_chapter_path(chapter_path),
2490                        duration=chapter_duration,
2491                    )
2492                    for chapter_path, chapter_duration in file_based_chapters
2493                ],
2494                allow_seek=True,
2495            )
2496
2497        # regular single-file streaming, simply let ffmpeg deal with the file directly
2498        return StreamDetails(
2499            provider=self.instance_id,
2500            item_id=item_id,
2501            audio_format=prov_mapping.audio_format,
2502            media_type=MediaType.AUDIOBOOK,
2503            stream_type=StreamType.LOCAL_FILE,
2504            duration=library_item.duration,
2505            size=file_item.file_size,
2506            data=file_item,
2507            path=file_item.absolute_path,
2508            allow_seek=True,
2509            can_seek=True,
2510        )
2511
2512    def _get_chapter_path(self, relative_path: str) -> str:
2513        """Return absolute path for a chapter file. Override for network storage."""
2514        return self.get_absolute_path(relative_path)
2515
2516    async def _get_chapters_for_audiobook(
2517        self, audiobook_file_item: FileSystemItem, tags: AudioTags
2518    ) -> tuple[int, list[MediaItemChapter]]:
2519        """
2520        Return chapters for an audiobook.
2521
2522        Chapter sources in order of preference:
2523        1. Multiple files with track tags - sorted by track number
2524        2. Single file with embedded chapters - use embedded chapter markers
2525        3. Multiple files without track tags - sorted alphabetically (fallback)
2526        """
2527        chapters: list[MediaItemChapter] = []
2528        all_chapter_files: list[tuple[str, float]] = []
2529        total_duration = 0.0
2530
2531        # Scan folder for chapter files, separating tagged from untagged
2532        chapter_file_items: list[tuple[FileSystemItem, AudioTags]] = []
2533        untagged_file_items: list[tuple[FileSystemItem, AudioTags]] = []
2534
2535        items = await self._scandir(audiobook_file_item.relative_parent_path)
2536        # Sort by filename for consistent alphabetical ordering
2537        items.sort(key=lambda x: x.filename.lower())
2538
2539        for item in items:
2540            if "." not in item.relative_path or item.is_dir:
2541                continue
2542            if item.ext not in AUDIOBOOK_EXTENSIONS:
2543                continue
2544            item_tags = await async_parse_tags(item.absolute_path, item.file_size)
2545            if not (tags.album == item_tags.album or (item_tags.tags.get("title") is None)):
2546                continue
2547            if item_tags.tags.get("track") is None:
2548                untagged_file_items.append((item, item_tags))
2549            else:
2550                chapter_file_items.append((item, item_tags))
2551
2552        # Determine chapter source
2553        use_embedded = False
2554        use_alphabetical = False
2555
2556        if len(chapter_file_items) > 1:
2557            chapter_file_items.sort(key=lambda x: (x[1].disc or 0, x[1].track or 0))
2558        elif len(chapter_file_items) <= 1 and tags.chapters:
2559            use_embedded = True
2560        elif len(untagged_file_items) > 1:
2561            use_alphabetical = True
2562            chapter_file_items = untagged_file_items
2563            self.logger.info(
2564                "Audiobook files have no track tags, using alphabetical order: %s",
2565                tags.album,
2566            )
2567
2568        if use_embedded:
2569            chapters = [
2570                MediaItemChapter(
2571                    position=chapter.chapter_id,
2572                    name=chapter.title or f"Chapter {chapter.chapter_id}",
2573                    start=chapter.position_start,
2574                    end=chapter.position_end,
2575                )
2576                for chapter in tags.chapters
2577            ]
2578            total_duration = try_parse_int(tags.duration) or 0
2579            self.logger.log(
2580                VERBOSE_LOG_LEVEL,
2581                "Audiobook '%s': %d embedded chapters, duration=%d",
2582                tags.album,
2583                len(chapters),
2584                int(total_duration),
2585            )
2586        else:
2587            for position, (chapter_item, chapter_tags) in enumerate(chapter_file_items, start=1):
2588                if chapter_tags.duration is None:
2589                    self.logger.warning(
2590                        "Chapter file has no duration, skipping: %s",
2591                        chapter_item.relative_path,
2592                    )
2593                    continue
2594                self.logger.debug("Chapter filename: %s", chapter_item.relative_path)
2595                chapters.append(
2596                    MediaItemChapter(
2597                        position=position,
2598                        name=chapter_tags.title,
2599                        start=total_duration,
2600                        end=total_duration + chapter_tags.duration,
2601                    )
2602                )
2603                all_chapter_files.append(
2604                    (
2605                        chapter_item.relative_path,
2606                        chapter_tags.duration,
2607                    )
2608                )
2609                total_duration += chapter_tags.duration
2610            sort_method = "alphabetical" if use_alphabetical else "track"
2611            self.logger.log(
2612                VERBOSE_LOG_LEVEL,
2613                "Audiobook '%s': %d files (%s order), duration=%d",
2614                tags.album,
2615                len(chapters),
2616                sort_method,
2617                int(total_duration),
2618            )
2619        # Cache chapter files for streaming
2620        await self.cache.set(
2621            key=audiobook_file_item.relative_path,
2622            data=all_chapter_files,
2623            provider=self.instance_id,
2624            category=CACHE_CATEGORY_AUDIOBOOK_CHAPTERS,
2625        )
2626        return int(total_duration), chapters
2627
2628    async def _get_podcast_metadata(self, podcast_folder: str) -> dict[str, Any]:
2629        """Return metadata for a podcast."""
2630        if (
2631            cache := await self.cache.get(
2632                key=podcast_folder,
2633                provider=self.instance_id,
2634                category=CACHE_CATEGORY_PODCAST_METADATA,
2635            )
2636        ) is not None:
2637            return cast("dict[str, Any]", cache)
2638        data: dict[str, Any] = {}
2639        metadata_file = os.path.join(podcast_folder, "metadata.json")
2640        if await self.exists(metadata_file):
2641            # found json file with metadata
2642            raw = await self._read_file(metadata_file)
2643            data.update(json_loads(raw.decode("utf-8")))
2644        await self.cache.set(
2645            key=podcast_folder,
2646            data=data,
2647            provider=self.instance_id,
2648            category=CACHE_CATEGORY_PODCAST_METADATA,
2649        )
2650        return data
2651
2652    async def _scandir(self, path: str) -> list[FileSystemItem]:
2653        """List directory contents in natural sort order."""
2654        # raw scandir order depends on the underlying filesystem (e.g. hash order
2655        # on ext4) so sort to make browse and folder playback order deterministic
2656        abs_path = self.get_absolute_path(path)
2657        return await asyncio.to_thread(sorted_scandir, self.base_path, abs_path, sort=True)
2658
2659    async def _read_file(self, path: str) -> bytes:
2660        """Read file contents. Override for network storage."""
2661        async with aiofiles.open(self.get_absolute_path(path), mode="rb") as f:
2662            return cast("bytes", await f.read())
2663