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, AsyncIterator, Iterator, Sequence
13from contextvars import ContextVar
14from datetime import UTC, datetime
15from pathlib import Path
16from typing import TYPE_CHECKING, Any, ClassVar, cast
17
18import aiofiles
19import shortuuid
20from aiofiles.os import wrap
21from music_assistant_models.enums import (
22    ContentType,
23    EventType,
24    ExternalID,
25    ImageType,
26    MediaType,
27    ProviderFeature,
28    StreamType,
29)
30from music_assistant_models.errors import (
31    InvalidDataError,
32    MediaNotFoundError,
33    MusicAssistantError,
34    SetupFailedError,
35)
36from music_assistant_models.helpers import create_safe_string
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.cache import BYPASS_CACHE
72from music_assistant.controllers.tasks.context import (
73    report_current_task_failure,
74    update_current_task_progress_from_index,
75    update_current_task_progress_text,
76)
77from music_assistant.helpers import lyrics
78from music_assistant.helpers.compare import compare_strings
79from music_assistant.helpers.cue_sheet import CueSheet
80from music_assistant.helpers.json import SerializableType, json_loads
81from music_assistant.helpers.playlists import parse_m3u, parse_pls
82from music_assistant.helpers.tags import AudioTags, async_parse_tags, clean_mbid
83from music_assistant.helpers.uri import create_uri
84from music_assistant.helpers.util import (
85    TaskManager,
86    detect_charset,
87    parse_title_and_version,
88    try_parse_int,
89)
90from music_assistant.models.music_provider import MusicProvider
91
92from .constants import (
93    AUDIOBOOK_EXTENSIONS,
94    AVAILABILITY_PROBE_INTERVAL,
95    CACHE_CATEGORY_ALBUM_INFO,
96    CACHE_CATEGORY_ARTIST_INFO,
97    CACHE_CATEGORY_AUDIOBOOK_CHAPTERS,
98    CACHE_CATEGORY_FOLDER_IMAGES,
99    CACHE_CATEGORY_METADATA_FILE,
100    CACHE_CATEGORY_PODCAST_EPISODES,
101    CACHE_CATEGORY_PODCAST_METADATA,
102    CACHE_CATEGORY_SOUND_EFFECTS,
103    CONF_CONTENT_TYPE,
104    CONF_ENTRY_CONTENT_TYPE,
105    CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
106    CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
107    CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
108    CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
109    CONF_ENTRY_LIBRARY_SYNC_TRACKS,
110    CONF_ENTRY_MISSING_ALBUM_ARTIST,
111    CONF_ENTRY_PROPAGATE_GENRES,
112    CUE_EXTENSIONS,
113    DEFAULT_AUDIOBOOK_PODCAST_GENRE,
114    IMAGE_EXTENSIONS,
115    METADATA_FILE_CACHE_EXPIRATION,
116    METADATA_FILE_EXTENSIONS,
117    NFO_FILENAMES,
118    PARTIAL_LISTING_CACHE_EXPIRATION,
119    PLAYLIST_EXTENSIONS,
120    PODCAST_EPISODE_EXTENSIONS,
121    SOUND_EFFECT_EXTENSIONS,
122    TRACK_EXTENSIONS,
123    WALK_EXTENSIONS,
124    IsChapterFile,
125    content_type_config_entry,
126)
127from .cue import (
128    CueSheetHandler,
129    cue_metadata_checksum,
130    cue_referenced_audio_stem,
131    make_cue_track_id,
132    parse_cue_track_id,
133)
134from .helpers import (
135    FileSystemItem,
136    ScanErrors,
137    get_absolute_path,
138    get_album_dir,
139    get_artist_dir,
140    get_folder_signature,
141    get_relative_path,
142    is_disc_dir,
143    is_image_file,
144    is_metadata_file,
145    parse_nfo_root,
146    recursive_iter,
147    sorted_scandir,
148)
149from .parsers import parse_album_nfo, parse_artist_nfo
150
151if TYPE_CHECKING:
152    from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
153    from music_assistant_models.provider import ProviderManifest
154
155    from music_assistant.mass import MusicAssistant
156    from music_assistant.models import ProviderInstanceType
157    from music_assistant.providers.musicbrainz import MusicbrainzProvider
158
159
160isdir = wrap(os.path.isdir)
161isfile = wrap(os.path.isfile)
162ismount = wrap(os.path.ismount)
163exists = wrap(os.path.exists)
164makedirs = wrap(os.makedirs)
165
166SUPPORTED_FEATURES = {
167    ProviderFeature.BROWSE,
168    ProviderFeature.SEARCH,
169}
170
171# task-local memo of on-demand folder listings (NFO files only) for one outermost parse, so
172# overlapping lookups for the same candidate folder list it only once; unset outside such a
173# scope, and unused entirely once the sync's own NFO index is ready
174_ONDEMAND_NFO_ITEMS: ContextVar[dict[str, dict[str, FileSystemItem]] | None] = ContextVar(
175    "ondemand_nfo_items", default=None
176)
177
178# every field parse_album_nfo/parse_artist_nfo reads; each, if present, must be a plain scalar
179# (not a repeated/nested XML element) or the NFO is not trusted as folder identity
180_ALBUM_NFO_FIELDS = (
181    "title",
182    "name",
183    "sortname",
184    "review",
185    "year",
186    "genre",
187    "musicbrainzalbumid",
188    "musicbrainzreleasegroupid",
189    "musicbrainzalbumartistid",
190)
191_ARTIST_NFO_FIELDS = ("title", "name", "sortname", "biography", "genre", "musicbrainzartistid")
192# fields whose consumer (split_items) explicitly also accepts a list/tuple of scalars, since
193# xmltodict yields a list for a repeated element (e.g. multiple <genre> tags)
194_LIST_ALLOWED_NFO_FIELDS = ("genre",)
195# the MusicBrainz id fields among the above: if present at all, must also be a valid UUID
196_ALBUM_MBID_FIELDS = ("musicbrainzalbumid", "musicbrainzreleasegroupid", "musicbrainzalbumartistid")
197_ARTIST_MBID_FIELDS = ("musicbrainzartistid",)
198
199
200async def setup(
201    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
202) -> ProviderInstanceType:
203    """Initialize provider(instance) with given configuration."""
204    return LocalFileSystemProvider(mass, manifest, config)
205
206
207class LocalFileSystemProvider(MusicProvider):
208    """
209    Implementation of a musicprovider for (local) files.
210
211    Reads ID3 tags from file and falls back to parsing filename.
212    Optionally reads metadata from nfo files and images in folder structure <artist>/<album>.
213    Supports m3u files for playlists.
214    """
215
216    # parallel workers per sync; subclasses lower this for slower transports
217    _SYNC_CONCURRENCY: ClassVar[int] = 16
218    _sync_tracks: bool = True
219    _sync_playlists: bool = True
220
221    def __init__(
222        self,
223        mass: MusicAssistant,
224        manifest: ProviderManifest,
225        config: ProviderConfig,
226        base_path: str | None = None,
227    ) -> None:
228        """Initialize MusicProvider."""
229        super().__init__(mass, manifest, config, SUPPORTED_FEATURES)
230        # subclasses (NFS/SMB/...) mount elsewhere and pass their own base_path;
231        # the plain local provider reads its scan directory from the setup data
232        self.base_path: str = (
233            base_path if base_path is not None else cast("str", self.get_setup_value(CONF_PATH))
234        )
235        self.write_access: bool = False
236        self.sync_running: bool = False
237        self.media_content_type = cast(
238            "str", self.get_setup_value(CONF_CONTENT_TYPE, CONF_ENTRY_CONTENT_TYPE.default_value)
239        )
240        self._cue = CueSheetHandler(self)
241        # sync-scoped index of this sync's walked album.nfo/artist.nfo, keyed by parent
242        # directory; an O(1) lookup for the NFO-resolution fallback instead of a filesystem
243        # probe per candidate folder. Ephemeral: built after the walk, cleared in _run_sync.
244        self._sync_nfo_by_dir: dict[str, dict[str, FileSystemItem]] = {}
245        # True only once _sync_nfo_by_dir reflects a completed walk; `sync_running` alone
246        # is not enough, since a concurrent on-demand parse could otherwise start consulting
247        # the index for the entire (potentially long) walk before it is actually populated
248        self._sync_nfo_index_ready: bool = False
249
250    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
251        """Return Config entries to configure this provider."""
252        # content type and path are collected by the setup flow; surface the (immutable)
253        # content type read-only so the sync options' depends_on chains still resolve
254        content_type = str(
255            self.get_setup_value(CONF_CONTENT_TYPE, CONF_ENTRY_CONTENT_TYPE.default_value)
256        )
257        return (
258            content_type_config_entry(content_type),
259            CONF_ENTRY_MISSING_ALBUM_ARTIST,
260            CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
261            CONF_ENTRY_LIBRARY_SYNC_TRACKS,
262            CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
263            CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
264            CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
265            CONF_ENTRY_PROPAGATE_GENRES,
266        )
267
268    @property
269    def supported_features(self) -> set[ProviderFeature]:
270        """Return the features supported by this Provider."""
271        base_features = {*SUPPORTED_FEATURES}
272        if self.media_content_type == "audiobooks":
273            return {ProviderFeature.LIBRARY_AUDIOBOOKS, *base_features}
274        if self.media_content_type == "podcasts":
275            return {ProviderFeature.LIBRARY_PODCASTS, *base_features}
276        if self.media_content_type == "sound_effects":
277            # sound effects are live-fetched content, never synced into the library
278            return {ProviderFeature.SOUND_EFFECTS, *base_features}
279        music_features = {
280            ProviderFeature.LIBRARY_ALBUMS,
281            ProviderFeature.LIBRARY_ARTISTS,
282            ProviderFeature.LIBRARY_TRACKS,
283            ProviderFeature.LIBRARY_PLAYLISTS,
284            *base_features,
285        }
286        if self.write_access:
287            music_features.add(ProviderFeature.PLAYLIST_TRACKS_EDIT)
288            music_features.add(ProviderFeature.PLAYLIST_CREATE)
289        return music_features
290
291    @property
292    def is_streaming_provider(self) -> bool:
293        """Return True if the provider is a streaming provider."""
294        return False
295
296    @property
297    def instance_name_postfix(self) -> str | None:
298        """Return a (default) instance name postfix for this provider instance."""
299        return Path(self.base_path).name
300
301    async def handle_async_init(self) -> None:
302        """Handle async initialization of the provider."""
303        if not await isdir(self.base_path):
304            msg = f"Music Directory {self.base_path} does not exist"
305            raise SetupFailedError(
306                msg,
307                translation_key="music_directory_not_found",
308                translation_owner=self.translation_owner,
309                translation_args=[self.base_path],
310            )
311        await self.check_write_access()
312
313    async def unload(self, is_removed: bool = False) -> None:
314        """Handle unload/close of the provider."""
315        self._cancel_availability_probe()
316        # a check that already started runs as a task under the same id, and it would
317        # otherwise keep talking to storage this unload is in the middle of tearing down
318        self.mass.cancel_task(self._availability_probe_id)
319
320    async def get_diagnostics(self) -> dict[str, SerializableType]:
321        """Return diagnostics info for this provider to include in diagnostics reports."""
322        return {
323            "sync_running": self.sync_running,
324            "write_access": self.write_access,
325            "content_type": self.media_content_type,
326        }
327
328    async def search(
329        self,
330        search_query: str,
331        media_types: list[MediaType] | None,
332        limit: int = 5,
333    ) -> SearchResults:
334        """Perform search on this file based musicprovider."""
335        result = SearchResults()
336        # searching the filesystem is slow and unreliable,
337        # so instead we just query the db...
338        if media_types is None or MediaType.TRACK in media_types:
339            result.tracks = await self.mass.music.tracks.get_library_items_by_query(
340                search=search_query, provider_filter=[self.instance_id], limit=limit
341            )
342
343        if media_types is None or MediaType.ALBUM in media_types:
344            result.albums = await self.mass.music.albums.get_library_items_by_query(
345                search=search_query,
346                provider_filter=[self.instance_id],
347                limit=limit,
348            )
349
350        if media_types is None or MediaType.ARTIST in media_types:
351            result.artists = await self.mass.music.artists.get_library_items_by_query(
352                search=search_query,
353                provider_filter=[self.instance_id],
354                limit=limit,
355            )
356        if media_types is None or MediaType.PLAYLIST in media_types:
357            result.playlists = await self.mass.music.playlists.get_library_items_by_query(
358                search=search_query,
359                provider_filter=[self.instance_id],
360                limit=limit,
361            )
362        if media_types is None or MediaType.AUDIOBOOK in media_types:
363            result.audiobooks = await self.mass.music.audiobooks.get_library_items_by_query(
364                search=search_query,
365                provider_filter=[self.instance_id],
366                limit=limit,
367            )
368        if media_types is None or MediaType.PODCAST in media_types:
369            result.podcasts = await self.mass.music.podcasts.get_library_items_by_query(
370                search=search_query,
371                provider_filter=[self.instance_id],
372                limit=limit,
373            )
374        return result
375
376    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
377        """
378        Browse this provider's items.
379
380        :param path: The path to browse, (e.g. provid://artists).
381        """
382        # for audiobooks and podcasts we just return all library items
383        if self.media_content_type == "podcasts":
384            return await self.mass.music.podcasts.library_items(
385                provider=self.instance_id, summary=False
386            )
387        if self.media_content_type == "audiobooks":
388            return await self.mass.music.audiobooks.library_items(
389                provider=self.instance_id, summary=False
390            )
391        items: list[MediaItemType | ItemMapping | BrowseFolder] = []
392        item_path = path.split("://", 1)[1]
393        if not item_path:
394            item_path = ""
395        scanned = await self._scandir(item_path)
396        # expand CUE sheets into per-track entries and hide the companion audio;
397        # synthetic ids match those minted during sync so get_track resolves them
398        cue_stems: set[str] = set()
399        if self.media_content_type == "music":
400            for item in scanned:
401                if item.ext not in CUE_EXTENSIONS:
402                    continue
403                cue_stems.add(item.absolute_path.rsplit(".", 1)[0])
404                try:
405                    cue_sheet = await self._cue.load_cue_sheet(item)
406                except InvalidDataError as err:
407                    self.logger.warning("Unable to parse CUE sheet %s: %s", item.relative_path, err)
408                    continue
409                # also hide the audio file named in the CUE (may differ from its stem)
410                if companion_stem := cue_referenced_audio_stem(item, cue_sheet):
411                    cue_stems.add(companion_stem)
412                for cue_track in cue_sheet.tracks:
413                    items.append(
414                        ItemMapping(
415                            media_type=MediaType.TRACK,
416                            item_id=make_cue_track_id(item.relative_path, cue_track.number),
417                            provider=self.instance_id,
418                            name=cue_track.title or f"Track {cue_track.number}",
419                        )
420                    )
421        for item in scanned:
422            if not item.is_dir and ("." not in item.filename or not item.ext):
423                # skip system files and files without extension
424                continue
425
426            if item.is_dir:
427                items.append(
428                    BrowseFolder(
429                        item_id=item.relative_path,
430                        provider=self.instance_id,
431                        path=f"{self.instance_id}://{item.relative_path}",
432                        name=item.filename,
433                        # mark folder as playable, assuming it contains tracks underneath
434                        is_playable=True,
435                    )
436                )
437            elif item.ext in TRACK_EXTENSIONS:
438                if item.absolute_path.rsplit(".", 1)[0] in cue_stems:
439                    continue
440                items.append(
441                    ItemMapping(
442                        media_type=(
443                            MediaType.SOUND_EFFECT
444                            if self.media_content_type == "sound_effects"
445                            else MediaType.TRACK
446                        ),
447                        item_id=item.relative_path,
448                        provider=self.instance_id,
449                        name=item.filename,
450                    )
451                )
452            elif item.ext in PLAYLIST_EXTENSIONS and self.media_content_type == "music":
453                items.append(
454                    ItemMapping(
455                        media_type=MediaType.PLAYLIST,
456                        item_id=item.relative_path,
457                        provider=self.instance_id,
458                        name=item.filename,
459                    )
460                )
461        if self.media_content_type == "music":
462            track_indexes = [
463                index
464                for index, item in enumerate(items)
465                if isinstance(item, ItemMapping) and item.media_type == MediaType.TRACK
466            ]
467            library_tracks = await asyncio.gather(
468                *(
469                    self.mass.music.tracks.get_library_item_by_prov_id(
470                        items[index].item_id, self.instance_id
471                    )
472                    for index in track_indexes
473                )
474            )
475            for index, library_track in zip(track_indexes, library_tracks, strict=True):
476                if library_track:
477                    items[index] = library_track
478        return items
479
480    async def sync_library(self, media_type: MediaType) -> None:
481        """Run library sync for this provider."""
482        if media_type in (MediaType.ARTIST, MediaType.ALBUM):
483            # artists and albums are synced as part of track sync
484            return
485        if self.media_content_type == "sound_effects":
486            # sound effects are live-fetched content, never synced into the library
487            return
488        # check if any sync options are enabled for this content type
489        # the filesystem provider processes all file types in one scan,
490        # so we can return early if nothing needs syncing
491        if self.media_content_type == "music":
492            self._sync_tracks = bool(self.config.get_value(CONF_ENTRY_LIBRARY_SYNC_TRACKS.key))
493            self._sync_playlists = bool(
494                self.config.get_value(CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS.key)
495            )
496            if not self._sync_tracks and not self._sync_playlists:
497                return
498        elif self.media_content_type == "audiobooks":
499            if not self.config.get_value(CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS.key):
500                return
501        elif self.media_content_type == "podcasts":
502            if not self.config.get_value(CONF_ENTRY_LIBRARY_SYNC_PODCASTS.key):
503                return
504        assert self.mass.music.database
505        if self.sync_running:
506            self.logger.warning("Library sync already running for %s", self.name)
507            return
508        file_checksums: dict[str, str] = {}
509        # NOTE: we always run a scan of the entire library, as we need to detect changes
510        # we ignore any given mediatype(s) and just scan all supported files
511        query = (
512            f"SELECT provider_item_id, details FROM {DB_TABLE_PROVIDER_MAPPINGS} "
513            f"WHERE provider_instance = '{self.instance_id}' "
514            f"AND media_type in ('track', 'playlist', 'audiobook', 'podcast_episode')"
515        )
516        for db_row in await self.mass.music.database.get_rows_from_query(query, limit=0):
517            file_checksums[db_row["provider_item_id"]] = str(db_row["details"])
518        # provider_mappings stores synthetic per-track ids for CUE sheets, not the
519        # CUE path, so collect every track checksum per path for the scan classifier
520        cue_file_checksums: dict[str, set[str]] = {}
521        for prov_item_id, checksum in file_checksums.items():
522            parsed = parse_cue_track_id(prov_item_id)
523            if parsed is not None:
524                cue_file_checksums.setdefault(parsed[0], set()).add(checksum)
525        # find all supported files in the base directory and all subfolders
526        # we work bottom up, as-in we derive all info from the tracks
527        cur_filenames: set[str] = set()
528        prev_filenames = set(file_checksums.keys())
529
530        items_to_process: list[tuple[FileSystemItem, str | None]] = []
531        unchanged_cue_items: list[FileSystemItem] = []
532        # absolute paths of every CUE sheet in this scan with the ".cue" stripped,
533        # used for O(1) companion-CUE lookups per audio file
534        cue_stems: set[str] = set()
535        # local metadata files (NFO/images) collected by the walk alongside normal media;
536        # never imported themselves, only used to detect a change worth reparsing their
537        # registered representative track
538        metadata_files: list[FileSystemItem] = []
539        # relative_path of every representative queued because of a metadata-file change
540        force_refresh_tracks: set[str] = set()
541        # collects the errors raised while walking the tree; any error means the
542        # scan is incomplete, a fatal one means the provider is unreachable
543        scan_errors = ScanErrors()
544
545        self.sync_running = True
546        self._sync_nfo_by_dir = {}
547        self._sync_nfo_index_ready = False
548        try:
549            await self._enumerate_files_for_sync(
550                file_checksums=file_checksums,
551                cue_file_checksums=cue_file_checksums,
552                cur_filenames=cur_filenames,
553                items_to_process=items_to_process,
554                unchanged_cue_items=unchanged_cue_items,
555                cue_stems=cue_stems,
556                scan_errors=scan_errors,
557                metadata_files=metadata_files,
558            )
559            if scan_errors.fatal:
560                # the storage is gone, so reading the files collected before it went
561                # away would only add a timeout each
562                self.logger.error("Aborting sync for %s: %s", self.name, scan_errors.fatal)
563                report_current_task_failure("Sync aborted: filesystem unavailable during scan")
564                self._set_available(False)
565                return
566            self._sync_nfo_by_dir = self._build_nfo_index(metadata_files)
567            # an incomplete scan (some folders/files failed to read) may be missing NFOs
568            # that do exist on disk; treating this partial index as authoritative could
569            # make a changed track wrongly resolve to a synthetic identity. Leave it
570            # unready so lookups fall back to listing each folder directly instead
571            self._sync_nfo_index_ready = not scan_errors.incomplete
572            if metadata_files:
573                await self._queue_changed_metadata_files(
574                    metadata_files,
575                    file_checksums,
576                    cue_file_checksums,
577                    items_to_process,
578                    force_refresh_tracks,
579                )
580            if force_refresh_tracks:
581                await self._drop_stale_album_artist_caches()
582            # a CUE may name an audio file other than its own; hide that companion too
583            if self.media_content_type == "music":
584                for cue_item in (
585                    *unchanged_cue_items,
586                    *(item for item, _ in items_to_process if item.ext in CUE_EXTENSIONS),
587                ):
588                    try:
589                        cue_sheet = await self._cue.load_cue_sheet(cue_item)
590                    except InvalidDataError:
591                        continue
592                    if companion_stem := cue_referenced_audio_stem(cue_item, cue_sheet):
593                        cue_stems.add(companion_stem)
594            # drop CUE companion audio: absorbed into CUE tracks and not tracked in
595            # provider_mappings, so they would otherwise flag as changed every sync
596            items_to_process = [
597                (item, prev)
598                for item, prev in items_to_process
599                if not (
600                    item.ext in TRACK_EXTENSIONS
601                    and item.absolute_path.rsplit(".", 1)[0] in cue_stems
602                )
603            ]
604            # register synthetic track IDs for unchanged CUE files so the
605            # deletion pass does not treat them as removed
606            for cue_item in unchanged_cue_items:
607                try:
608                    cue_sheet = await self._cue.load_cue_sheet(cue_item)
609                except InvalidDataError as err:
610                    self.logger.warning(
611                        "Unable to parse CUE sheet %s: %s", cue_item.relative_path, err
612                    )
613                    continue
614                for cue_track in cue_sheet.tracks:
615                    cur_filenames.add(make_cue_track_id(cue_item.relative_path, cue_track.number))
616            total_items = len(items_to_process)
617            self.logger.info(
618                "Found %d changed/new items to process for %s",
619                total_items,
620                self.name,
621            )
622
623            # _SYNC_CONCURRENCY caps parallelism per provider (NFS/SMB/WebDAV friendly)
624            processed_count = 0
625
626            async def _process(item: FileSystemItem, prev_checksum: str | None) -> None:
627                nonlocal processed_count
628                if await self._process_item_async(
629                    item, prev_checksum, cur_filenames, cue_stems, prev_filenames
630                ):
631                    cur_filenames.add(item.relative_path)
632                processed_count += 1
633                if processed_count % 50 == 0 or processed_count == total_items:
634                    update_current_task_progress_from_index(
635                        processed_count,
636                        total_items,
637                        f"Processed {processed_count}/{total_items} files",
638                    )
639
640            with self._ondemand_listing_scope():
641                async with TaskManager(self.mass, self._SYNC_CONCURRENCY) as tm:
642                    for item, prev_checksum in items_to_process:
643                        await tm.create_task_with_limit(_process(item, prev_checksum))
644        finally:
645            self.sync_running = False
646            self._sync_nfo_by_dir = {}
647            self._sync_nfo_index_ready = False
648
649        # do not run deletions on a clean but empty scan of a previously non-empty library
650        # (wrong share mounted, empty backup mount, ...)
651        if prev_filenames and not cur_filenames:
652            self.logger.error(
653                "Aborting sync for %s: scan found no files but %d were previously indexed",
654                self.name,
655                len(prev_filenames),
656            )
657            report_current_task_failure(
658                f"Sync aborted: scan found no files but {len(prev_filenames)} "
659                "were previously indexed"
660            )
661            return
662
663        # a scan that skipped folders or files is incomplete: what it missed is still
664        # there, so deleting it from the library would throw away valid content
665        if scan_errors.incomplete:
666            summary = scan_errors.describe()
667            self.logger.warning("Skipping deletions for %s: %s", self.name, summary)
668            report_current_task_failure(f"Deletions skipped: {summary}")
669        else:
670            deleted_files = prev_filenames - cur_filenames
671            await self._process_deletions(deleted_files)
672            await self._process_orphaned_albums_and_artists()
673
674        # flag provider as available again if an earlier sync had marked it down
675        self._set_available(True)
676
677    async def get_artist(self, prov_artist_id: str) -> Artist:
678        """Get full artist details by id."""
679        db_artist = await self.mass.music.artists.get_library_item_by_prov_id(
680            prov_artist_id, self.instance_id
681        )
682        if not db_artist:
683            # no db item yet (e.g. browsing, or a manual refresh's second fetch after a
684            # normal/NFO match resolved a new path before its mapping was persisted).
685            # Recover identity from that path instead of falling back to its basename
686            if await self.exists(prov_artist_id):
687                with self._ondemand_listing_scope():
688                    name = Path(prov_artist_id).name
689                    sort_name: str | None = None
690                    mbid: str | None = None
691                    if nfo_item := await self._nfo_item_for(prov_artist_id, "artist.nfo"):
692                        nfo_root = await self._load_nfo_root(nfo_item, "artist")
693                        if nfo_root and (
694                            nfo_mbid := clean_mbid(
695                                nfo_root.get("musicbrainzartistid"), nfo_item.relative_path
696                            )
697                        ):
698                            mbid = nfo_mbid
699                            if library_artist := (
700                                await self.mass.music.artists.get_library_item_by_external_id(
701                                    mbid, ExternalID.MB_ARTIST
702                                )
703                            ):
704                                name = library_artist.name
705                                sort_name = library_artist.sort_name
706                    if not mbid and (
707                        library_artist := await self._find_artist_by_folder_name(name)
708                    ):
709                        # no MBID to resolve by (this path was matched via a normal
710                        # folder/sort-name-alias match, not an artist.nfo): recover the
711                        # one already-known library artist this path belongs to instead
712                        # of renaming it to the folder's own basename
713                        name = library_artist.name
714                        sort_name = library_artist.sort_name
715                        mbid = library_artist.mbid
716                    return await self._parse_artist(
717                        name, sort_name=sort_name, mbid=mbid, artist_path=prov_artist_id
718                    )
719            return await self._parse_artist(prov_artist_id)
720
721        # prov_artist_id is either an actual (relative) path or a name (as fallback)
722        safe_artist_name = create_safe_string(prov_artist_id, lowercase=False, replace_space=False)
723        if await self.exists(prov_artist_id):
724            artist_path = prov_artist_id
725        elif await self.exists(safe_artist_name):
726            artist_path = safe_artist_name
727        else:
728            for prov_mapping in db_artist.provider_mappings:
729                if prov_mapping.provider_instance != self.instance_id:
730                    continue
731                if prov_mapping.url:
732                    artist_path = prov_mapping.url
733                    break
734            else:
735                # no path of its own: anchor a bounded artist.nfo attempt on one of its own
736                # tracks instead of giving up, so adding an artist.nfo and refreshing can
737                # still resolve it
738                representative_track = await self._resolve_artist_representative_track(db_artist)
739                if representative_track:
740                    with self._ondemand_listing_scope():
741                        return await self._parse_artist(
742                            db_artist.name,
743                            sort_name=db_artist.sort_name,
744                            mbid=db_artist.mbid,
745                            album_dir=os.path.dirname(representative_track),
746                            representative_track=representative_track,
747                        )
748                return db_artist
749        return await self._parse_artist(
750            db_artist.name,
751            sort_name=db_artist.sort_name,
752            mbid=db_artist.mbid,
753            artist_path=artist_path,
754            representative_track=await self._resolve_artist_representative_track(db_artist),
755        )
756
757    async def get_album(self, prov_album_id: str) -> Album:
758        """Get full album details by id."""
759        parsed_cue_paths: set[str] = set()
760        # early returns below stop iterating this generator before it's exhausted; without an
761        # explicit aclose() that leaves its _ondemand_listing_scope() cleanup (a ContextVar
762        # reset) to whenever the event loop's async-generator finalizer happens to run, instead
763        # of deterministically, right here
764        async with contextlib.aclosing(self._iter_album_tracks(prov_album_id)) as tracks:
765            async for track in tracks:
766                if isinstance(track.album, Album):
767                    # already a fully parsed album: the folder-scan fallback (used when this id
768                    # has no library mapping yet) yields these directly, so re-resolving and
769                    # re-parsing the same file below would only repeat the same tag/NFO work
770                    return track.album
771                for prov_mapping in track.provider_mappings:
772                    if prov_mapping.provider_instance != self.instance_id:
773                        continue
774                    if parsed := parse_cue_track_id(prov_mapping.item_id):
775                        # every track from the same CUE shares the same album; only parse once
776                        if parsed[0] in parsed_cue_paths:
777                            continue
778                        parsed_cue_paths.add(parsed[0])
779                        cue_item = await self.resolve(parsed[0])
780                        for cue_track in await self._cue.parse_tracks(cue_item):
781                            if isinstance(cue_track.album, Album):
782                                return cue_track.album
783                        continue
784                    file_item = await self.resolve(prov_mapping.item_id)
785                    tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
786                    full_track = await self._parse_track(file_item, tags)
787                    assert isinstance(full_track.album, Album)
788                    return full_track.album
789        msg = f"Album not found: {prov_album_id}"
790        raise MediaNotFoundError(msg)
791
792    async def get_track(self, prov_track_id: str) -> Track:
793        """Get full track details by id."""
794        # ruff: noqa: PLR0915
795        if parsed := parse_cue_track_id(prov_track_id):
796            cue_item = await self.resolve(parsed[0])
797            for cue_track in await self._cue.parse_tracks(cue_item):
798                if cue_track.item_id == prov_track_id:
799                    return cue_track
800            msg = f"CUE track not found: {prov_track_id}"
801            raise MediaNotFoundError(msg)
802
803        if not await self.exists(prov_track_id):
804            msg = f"Track path does not exist: {prov_track_id}"
805            raise MediaNotFoundError(msg)
806
807        file_item = await self.resolve(prov_track_id)
808        tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
809        return await self._parse_track(file_item, tags=tags, full_album_metadata=True)
810
811    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
812        """Get (full) podcast episode details by id."""
813        if not await self.exists(prov_episode_id):
814            msg = f"Episode path does not exist: {prov_episode_id}"
815            raise MediaNotFoundError(msg)
816        file_item = await self.resolve(prov_episode_id)
817        tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
818        return await self._parse_podcast_episode(file_item, tags=tags)
819
820    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
821        """Get full playlist details by id."""
822        if not await self.exists(prov_playlist_id):
823            msg = f"Playlist path does not exist: {prov_playlist_id}"
824            raise MediaNotFoundError(msg)
825
826        file_item = await self.resolve(prov_playlist_id)
827        playlist = Playlist(
828            item_id=file_item.relative_path,
829            provider=self.instance_id,
830            name=file_item.name,
831            provider_mappings={
832                ProviderMapping(
833                    item_id=file_item.relative_path,
834                    provider_domain=self.domain,
835                    provider_instance=self.instance_id,
836                    details=file_item.checksum,
837                    in_library=True,
838                )
839            },
840        )
841        playlist.is_editable = ProviderFeature.PLAYLIST_TRACKS_EDIT in self.supported_features
842        # only playlists in the root are editable - all other are read only
843        if "/" in prov_playlist_id or "\\" in prov_playlist_id:
844            playlist.is_editable = False
845        # we do not (yet) have support to edit/create pls playlists, only m3u files can be edited
846        if file_item.ext == "pls":
847            playlist.is_editable = False
848        playlist.owner = self.name
849        # Check for local image with the same basename
850        if local_image := await self._get_playlist_local_image(file_item):
851            playlist.metadata.images = UniqueList([local_image])
852        return playlist
853
854    async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
855        """Get full audiobook details by id."""
856        # ruff: noqa: PLR0915
857        if not await self.exists(prov_audiobook_id):
858            msg = f"Audiobook path does not exist: {prov_audiobook_id}"
859            raise MediaNotFoundError(msg)
860
861        file_item = await self.resolve(prov_audiobook_id)
862        tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
863        return await self._parse_audiobook(file_item, tags=tags)
864
865    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
866        """Get full podcast details by id."""
867        async for episode in self.get_podcast_episodes(prov_podcast_id):
868            assert isinstance(episode.podcast, Podcast)
869            return episode.podcast
870        msg = f"Podcast not found: {prov_podcast_id}"
871        raise MediaNotFoundError(msg)
872
873    async def get_sound_effect(self, prov_sound_effect_id: str) -> SoundEffect:
874        """Get full sound effect details by id."""
875        if not await self.exists(prov_sound_effect_id):
876            msg = f"Sound effect path does not exist: {prov_sound_effect_id}"
877            raise MediaNotFoundError(msg)
878        file_item = await self.resolve(prov_sound_effect_id)
879        return await self._get_or_parse_sound_effect(file_item)
880
881    async def get_sound_effects(self) -> AsyncGenerator[SoundEffect]:
882        """Get all sound effect items this provider offers."""
883
884        def _walk() -> list[FileSystemItem]:
885            return sorted(
886                recursive_iter(
887                    self.base_path, self.base_path, SOUND_EFFECT_EXTENSIONS, self.logger
888                ),
889                key=lambda x: x.relative_path,
890            )
891
892        for file_item in await asyncio.to_thread(_walk):
893            yield await self._get_or_parse_sound_effect(file_item)
894
895    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
896        """Get album tracks for given album id."""
897        tracks = [track async for track in self._iter_album_tracks(prov_album_id)]
898        db_album = await self.mass.music.albums.get_library_item_by_prov_id(
899            prov_album_id, self.instance_id
900        )
901        if db_album is None:
902            # mappingless result: folder listing order (WebDAV/cloud listings are not
903            # guaranteed ordered) would otherwise be returned to the caller as-is
904            tracks.sort(key=lambda track: (track.disc_number, track.track_number))
905        return tracks
906
907    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
908        """Get playlist tracks."""
909        result: list[Track] = []
910        if page > 0:
911            # paging not (yet) supported
912            return result
913        if not await self.exists(prov_playlist_id):
914            msg = f"Playlist path does not exist: {prov_playlist_id}"
915            raise MediaNotFoundError(msg)
916
917        file_item = await self.resolve(prov_playlist_id)
918        # We are using the checksum of the playlist file here to invalidate the cache
919        # when a change has been made to the playlist file (ie track addition/deletion)
920        cache_checksum = file_item.checksum
921
922        cache_key = f"get_playlist_tracks.{prov_playlist_id}"
923        cached_data = await self.mass.cache.get(
924            cache_key,
925            provider=self.instance_id,
926            checksum=cache_checksum,
927            category=0,
928            base_class=Track,
929        )
930        if cached_data is not None:
931            return cached_data  # type: ignore[no-any-return]
932
933        _, ext = prov_playlist_id.rsplit(".", 1)
934        try:
935            # get playlist file contents
936            playlist_data_raw = await self._read_file(prov_playlist_id)
937            encoding = await detect_charset(playlist_data_raw)
938            playlist_data = playlist_data_raw.decode(encoding, errors="replace")
939
940            if ext in ("m3u", "m3u8"):
941                playlist_lines = parse_m3u(playlist_data)
942            else:
943                playlist_lines = parse_pls(playlist_data)
944
945            for idx, playlist_line in enumerate(playlist_lines, 1):
946                if "#EXT" in playlist_line.path:
947                    continue
948                if track := await self._parse_playlist_line(
949                    playlist_line.path, os.path.dirname(prov_playlist_id)
950                ):
951                    track.position = idx
952                    result.append(track)
953
954        except Exception as err:
955            self.logger.warning(
956                "Error while parsing playlist %s: %s",
957                prov_playlist_id,
958                str(err),
959                exc_info=err if self.logger.isEnabledFor(10) else None,
960            )
961
962        await self.mass.cache.set(
963            key=cache_key,
964            data=[track.to_dict() for track in result],
965            expiration=3600 * 24 * 365,  # File timestamp checksum handles invalidation
966            provider=self.instance_id,
967            checksum=cache_checksum,
968            category=0,
969        )
970
971        return result
972
973    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
974        """Get podcast episodes for given podcast id."""
975        folder_items = [item for item in await self._scandir(prov_podcast_id) if not item.is_dir]
976        episode_files = [x for x in folder_items if x.ext in PODCAST_EPISODE_EXTENSIONS]
977        # artwork and metadata.json count towards the signature too, because the parse embeds
978        # them into every episode. Case-insensitive, matching _get_podcast_metadata's exists()
979        signature_files = [
980            x
981            for x in folder_items
982            if x.ext in PODCAST_EPISODE_EXTENSIONS
983            or x.ext in IMAGE_EXTENSIONS
984            or x.filename.lower() == "metadata.json"
985        ]
986        cache_key = f"podcast_episodes.{prov_podcast_id}"
987        cache_checksum = get_folder_signature(signature_files)
988        if (
989            cached_episodes := await self.mass.cache.get(
990                cache_key,
991                provider=self.instance_id,
992                category=CACHE_CATEGORY_PODCAST_EPISODES,
993                checksum=cache_checksum,
994                base_class=PodcastEpisode,
995            )
996        ) is not None:
997            for episode in cached_episodes:
998                yield episode
999            return
1000
1001        # these caches have no checksum of their own, so drop them before parsing or the new
1002        # entry gets the values the signature just invalidated. Refill once, or every parse
1003        # task below misses at the same time and repeats the same scandir and file read
1004        for stale_category in (CACHE_CATEGORY_FOLDER_IMAGES, CACHE_CATEGORY_PODCAST_METADATA):
1005            await self.mass.cache.delete(
1006                prov_podcast_id, category=stale_category, provider=self.instance_id
1007            )
1008        await self._get_local_images(prov_podcast_id)
1009        await self._get_podcast_metadata(prov_podcast_id)
1010
1011        # collected by index so the listing keeps scandir order, not parse completion order
1012        parsed: list[PodcastEpisode | None] = [None] * len(episode_files)
1013
1014        async def _process_podcast_episode(index: int, item: FileSystemItem) -> None:
1015            try:
1016                tags = await async_parse_tags(item.absolute_path, item.file_size)
1017                parsed[index] = await self._parse_podcast_episode(item, tags)
1018            except MusicAssistantError as err:
1019                self.logger.warning(
1020                    "Could not parse uri/file %s to podcast episode: %s",
1021                    item.relative_path,
1022                    str(err),
1023                )
1024
1025        # reuse the per-sync worker limit: the slowest filesystems to parse are exactly the
1026        # ones that lower it
1027        async with TaskManager(self.mass, self._SYNC_CONCURRENCY) as tm:
1028            for index, item in enumerate(episode_files):
1029                await tm.create_task_with_limit(_process_podcast_episode(index, item))
1030
1031        episodes = [episode for episode in parsed if episode is not None]
1032        # cache an incomplete listing briefly rather than not at all, so one unreadable file
1033        # cannot make every request re-parse the whole folder
1034        complete = len(episodes) == len(episode_files)
1035        await self.mass.cache.set(
1036            key=cache_key,
1037            data=[episode.to_dict() for episode in episodes],
1038            # a complete listing is invalidated by the folder signature instead
1039            expiration=3600 * 24 * 365 if complete else PARTIAL_LISTING_CACHE_EXPIRATION,
1040            provider=self.instance_id,
1041            category=CACHE_CATEGORY_PODCAST_EPISODES,
1042            checksum=cache_checksum,
1043        )
1044
1045        for episode in episodes:
1046            yield episode
1047
1048    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
1049        """Add track(s) to playlist."""
1050        if not await self.exists(prov_playlist_id):
1051            msg = f"Playlist path does not exist: {prov_playlist_id}"
1052            raise MediaNotFoundError(msg)
1053        playlist_filename = self.get_absolute_path(prov_playlist_id)
1054        async with aiofiles.open(playlist_filename, encoding="utf-8") as _file:
1055            playlist_data = await _file.read()
1056        for file_path in prov_track_ids:
1057            track = await self.get_track(file_path)
1058            playlist_data += f"\n#EXTINF:{track.duration or 0},{track.name}\n{file_path}\n"
1059
1060        # write playlist file (always in utf-8)
1061        async with aiofiles.open(playlist_filename, "w", encoding="utf-8") as _file:
1062            await _file.write(playlist_data)
1063
1064    async def remove_playlist_tracks(
1065        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
1066    ) -> None:
1067        """Remove track(s) from playlist."""
1068        if not await self.exists(prov_playlist_id):
1069            msg = f"Playlist path does not exist: {prov_playlist_id}"
1070            raise MediaNotFoundError(msg)
1071        _, ext = prov_playlist_id.rsplit(".", 1)
1072        # get playlist file contents
1073        playlist_filename = self.get_absolute_path(prov_playlist_id)
1074        async with aiofiles.open(playlist_filename, encoding="utf-8") as _file:
1075            playlist_data = await _file.read()
1076        # get current contents first
1077        if ext in ("m3u", "m3u8"):
1078            playlist_items = parse_m3u(playlist_data)
1079        else:
1080            playlist_items = parse_pls(playlist_data)
1081        # remove items by index
1082        for i in sorted(positions_to_remove, reverse=True):
1083            # position = index + 1
1084            del playlist_items[i - 1]
1085        # build new playlist data
1086        new_playlist_data = "#EXTM3U\n"
1087        for item in playlist_items:
1088            new_playlist_data += f"\n#EXTINF:{item.length or 0},{item.title}\n{item.path}\n"
1089        async with aiofiles.open(playlist_filename, "w", encoding="utf-8") as _file:
1090            await _file.write(new_playlist_data)
1091
1092    async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
1093        """Create a new playlist on provider with given name."""
1094        # creating a new playlist on the filesystem is as easy
1095        # as creating a new (empty) file with the m3u extension...
1096        # filename = await self.resolve(f"{name}.m3u")
1097        filename = f"{name}.m3u"
1098        playlist_filename = self.get_absolute_path(filename)
1099        async with aiofiles.open(playlist_filename, "w", encoding="utf-8") as _file:
1100            await _file.write("#EXTM3U\n")
1101        return await self.get_playlist(filename)
1102
1103    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
1104        """Return the content details for the given track when it will be streamed."""
1105        try:
1106            if media_type == MediaType.AUDIOBOOK:
1107                return await self._get_stream_details_for_audiobook(item_id)
1108            if media_type == MediaType.PODCAST_EPISODE:
1109                return await self._get_stream_details_for_podcast_episode(item_id)
1110            if media_type == MediaType.SOUND_EFFECT:
1111                return await self._get_stream_details_for_sound_effect(item_id)
1112            return await self._get_stream_details_for_track(item_id)
1113        except FileNotFoundError:
1114            self.logger.warning(
1115                "File not found for media item %s",
1116                item_id,
1117            )
1118            msg = f"Media file not found: {item_id}"
1119            raise MediaNotFoundError(msg)
1120
1121    async def get_audio_stream(
1122        self, streamdetails: StreamDetails, seek_position: int = 0
1123    ) -> AsyncGenerator[bytes]:
1124        """Return the custom audio stream for the provider item."""
1125        # only CUE-derived tracks use StreamType.CUSTOM in this provider
1126        async for chunk in self._cue.get_audio_stream(streamdetails, seek_position):
1127            yield chunk
1128
1129    async def resolve_image(self, path: str) -> str | bytes:
1130        """
1131        Resolve an image from an image path.
1132
1133        This either returns (a generator to get) raw bytes of the image or
1134        a string with an http(s) URL or local path that is accessible from the server.
1135        """
1136        # drop the cache-busting suffix appended by _versioned_image_path
1137        try:
1138            file_item = await self.resolve(path.split("?cs=", 1)[0])
1139        except FileNotFoundError as err:
1140            # the referenced image file was removed from disk; surface a typed
1141            # not-found so the image layer treats it as a missing image
1142            raise MediaNotFoundError(f"Image not found: {path}") from err
1143        if file_item.is_dir:
1144            # handing the path back would have the image layer run an ffmpeg
1145            # embedded-artwork extraction on the directory before giving up
1146            raise MediaNotFoundError(f"Image path is a directory: {path}")
1147        return file_item.absolute_path
1148
1149    async def check_write_access(self) -> None:
1150        """Perform check if we have write access."""
1151        # verify write access to determine we have playlist create/edit support
1152        # overwrite with provider specific implementation if needed
1153        temp_file_name = self.get_absolute_path(f"{shortuuid.random(8)}.txt")
1154        try:
1155            async with aiofiles.open(temp_file_name, "w") as _file:
1156                await _file.write("test")
1157            await asyncio.to_thread(os.remove, temp_file_name)
1158            self.write_access = True
1159        except Exception as err:
1160            self.logger.debug("Write access disabled: %s", str(err))
1161
1162    async def resolve(self, file_path: str) -> FileSystemItem:
1163        """Resolve (absolute or relative) path to FileSystemItem."""
1164        absolute_path = self.get_absolute_path(file_path)
1165
1166        def _create_item() -> FileSystemItem:
1167            if Path(absolute_path).is_dir():
1168                return FileSystemItem(
1169                    filename=Path(file_path).name,
1170                    relative_path=get_relative_path(self.base_path, file_path),
1171                    absolute_path=absolute_path,
1172                    is_dir=True,
1173                )
1174            stat_info = Path(absolute_path).stat(follow_symlinks=False)
1175            return FileSystemItem(
1176                filename=Path(file_path).name,
1177                relative_path=get_relative_path(self.base_path, file_path),
1178                absolute_path=absolute_path,
1179                is_dir=False,
1180                checksum=str(int(stat_info.st_mtime)),
1181                file_size=stat_info.st_size,
1182                metadata_token=str(stat_info.st_mtime_ns),
1183            )
1184
1185        return await asyncio.to_thread(_create_item)
1186
1187    async def exists(self, file_path: str) -> bool:
1188        """Return bool is this FileSystem musicprovider has given file/dir."""
1189        if not file_path:
1190            return False
1191        try:
1192            abs_path = self.get_absolute_path(file_path)
1193        except MediaNotFoundError:
1194            # a path that escapes the base directory simply does not exist here
1195            return False
1196        return bool(await exists(abs_path))
1197
1198    def get_absolute_path(self, file_path: str) -> str:
1199        """Return absolute path for given file path."""
1200        return get_absolute_path(self.base_path, file_path)
1201
1202    async def _enumerate_files_for_sync(
1203        self,
1204        *,
1205        file_checksums: dict[str, str],
1206        cue_file_checksums: dict[str, set[str]],
1207        cur_filenames: set[str],
1208        items_to_process: list[tuple[FileSystemItem, str | None]],
1209        unchanged_cue_items: list[FileSystemItem],
1210        cue_stems: set[str],
1211        scan_errors: ScanErrors,
1212        metadata_files: list[FileSystemItem],
1213    ) -> None:
1214        """
1215        Walk every supported file under the provider root and populate the sync buckets.
1216
1217        Override in subclasses that cannot use a local ``os.scandir`` walk.
1218        Implementations must route each discovered file through
1219        :meth:`_classify_scan_item`, report every unreadable directory to
1220        ``scan_errors`` and stop the walk once it reports ``aborted``.
1221
1222        :param file_checksums: Previously stored checksum per provider item id.
1223        :param cue_file_checksums: Previously stored track checksums keyed by CUE relative_path.
1224        :param cur_filenames: Receives the ids/paths present in this scan.
1225        :param items_to_process: Receives changed or new items to process.
1226        :param unchanged_cue_items: Receives CUE sheets whose checksum matches.
1227        :param cue_stems: Receives absolute paths (minus extension) of CUE sheets.
1228        :param scan_errors: Receives the errors raised while walking the tree.
1229        :param metadata_files: Receives local metadata files (NFO/images) found in the walk.
1230        """
1231        ignore_album_playlists = self.media_content_type == "music" and bool(
1232            self.config.get_value(CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS.key)
1233        )
1234
1235        def _walk() -> None:
1236            for scanned, item in enumerate(
1237                recursive_iter(
1238                    self.base_path,
1239                    self.base_path,
1240                    WALK_EXTENSIONS,
1241                    self.logger,
1242                    scan_errors=scan_errors,
1243                ),
1244                start=1,
1245            ):
1246                if scanned % 500 == 0:
1247                    update_current_task_progress_text(f"Scanning files: {scanned} found")
1248                self._classify_scan_item(
1249                    item,
1250                    file_checksums=file_checksums,
1251                    cue_file_checksums=cue_file_checksums,
1252                    cur_filenames=cur_filenames,
1253                    items_to_process=items_to_process,
1254                    unchanged_cue_items=unchanged_cue_items,
1255                    cue_stems=cue_stems,
1256                    ignore_album_playlists=ignore_album_playlists,
1257                    metadata_files=metadata_files,
1258                )
1259
1260        await asyncio.to_thread(_walk)
1261
1262    def _classify_scan_item(
1263        self,
1264        item: FileSystemItem,
1265        *,
1266        file_checksums: dict[str, str],
1267        cue_file_checksums: dict[str, set[str]],
1268        cur_filenames: set[str],
1269        items_to_process: list[tuple[FileSystemItem, str | None]],
1270        unchanged_cue_items: list[FileSystemItem],
1271        cue_stems: set[str],
1272        ignore_album_playlists: bool,
1273        metadata_files: list[FileSystemItem],
1274    ) -> None:
1275        """
1276        Route a single scanned file into the correct sync bucket.
1277
1278        :param item: The file to classify.
1279        :param file_checksums: Previously stored checksum per provider item id.
1280        :param cue_file_checksums: Previously stored track checksums keyed by CUE relative_path.
1281        :param cur_filenames: Receives the ids/paths present in this scan.
1282        :param items_to_process: Receives changed or new items to process.
1283        :param unchanged_cue_items: Receives CUE sheets whose checksum matches.
1284        :param cue_stems: Receives absolute paths (minus extension) of CUE sheets.
1285        :param ignore_album_playlists: When True, skip playlists nested inside
1286            album directories.
1287        :param metadata_files: Receives local metadata files (NFO/images) found in the walk.
1288        """
1289        if is_metadata_file(item):
1290            # a local metadata file is never imported media: it carries no provider mapping
1291            # and is never deleted, only checked (after the walk) for a change worth
1292            # reparsing its registered representative track
1293            metadata_files.append(item)
1294            return
1295        if not item.is_dir and item.ext and item.ext.lower() in METADATA_FILE_EXTENSIONS:
1296            # an nfo/image extension is walked only to catch a recognized metadata file above;
1297            # an unrecognized one (wrong filename) must stay as invisible to the scan as before
1298            # this feature widened the walk beyond SUPPORTED_EXTENSIONS, or a stray image/nfo
1299            # on a wrong/empty mount could satisfy the "not empty" check below and silently
1300            # bypass the safeguard against deleting an entire library
1301            return
1302        # a file this provider never imports gets no mapping, so it would flag as
1303        # changed on every sync; it is still on disk, so record it as present
1304        if not self._is_imported_file(item):
1305            cur_filenames.add(item.relative_path)
1306            return
1307        # skip playlists in album directories if configured
1308        if (
1309            item.ext in PLAYLIST_EXTENSIONS
1310            and ignore_album_playlists
1311            and len(item.relative_path.split("/")) > 2
1312        ):
1313            return
1314        is_cue = item.ext in CUE_EXTENSIONS and self.media_content_type == "music"
1315        item_checksum = item.checksum
1316        if is_cue:
1317            cue_stems.add(item.absolute_path.rsplit(".", 1)[0])
1318            item_checksum = cue_metadata_checksum(item.checksum)
1319            prev_checksums = cue_file_checksums.get(item.relative_path, set())
1320            prev_checksum = min(prev_checksums, default=None)
1321            checksum_matches = prev_checksums == {item_checksum}
1322        else:
1323            prev_checksum = file_checksums.get(item.relative_path)
1324            checksum_matches = item_checksum == prev_checksum
1325        if checksum_matches:
1326            # unchanged, just record it as still present
1327            cur_filenames.add(item.relative_path)
1328            if is_cue:
1329                unchanged_cue_items.append(item)
1330        else:
1331            items_to_process.append((item, prev_checksum))
1332
1333    def _is_imported_file(self, item: FileSystemItem) -> bool:
1334        """Return True when this provider imports the given file into the library."""
1335        if self.media_content_type == "music":
1336            if item.ext in CUE_EXTENSIONS:
1337                return True
1338            if item.ext in TRACK_EXTENSIONS:
1339                return self._sync_tracks
1340            if item.ext in PLAYLIST_EXTENSIONS:
1341                return self._sync_playlists
1342            return False
1343        if self.media_content_type == "audiobooks":
1344            return item.ext in AUDIOBOOK_EXTENSIONS
1345        if self.media_content_type == "podcasts":
1346            return item.ext in PODCAST_EPISODE_EXTENSIONS
1347        return False
1348
1349    async def _root_artist_path(self, name: str) -> str | None:
1350        """
1351        Return a root-level artist folder matching this exact name, if any.
1352
1353        Tries the plain name and its filesystem-safe variant, so a root-level folder that
1354        differs only in punctuation (e.g. "AC/DC" stored as "ACDC") still resolves.
1355
1356        :param name: The artist name (or a sort-name alias) to match against a root folder.
1357        """
1358        if await self.exists(name):
1359            return name
1360        safe_name = create_safe_string(name, lowercase=False, replace_space=False)
1361        if await self.exists(safe_name):
1362            return safe_name
1363        return None
1364
1365    async def _find_artist_path(
1366        self, candidate_name: str, album_dir: str | None, *, exact_only: bool = False
1367    ) -> str | None:
1368        """
1369        Return an artist folder for one name candidate: a root, ancestor, or known item's path.
1370
1371        :param candidate_name: The artist name (or a sort-name alias) to match against a folder.
1372        :param album_dir: The album directory whose ancestors are searched, if any.
1373        :param exact_only: Only accept an exact (normalized) match, skipping the relaxed
1374            (fuzzy) fallback built into the ancestor search.
1375        """
1376        if artist_path := await self._root_artist_path(candidate_name):
1377            return artist_path
1378        if album_dir and (
1379            artist_path := get_artist_dir(
1380                candidate_name, album_dir=album_dir, exact_only=exact_only
1381            )
1382        ):
1383            return artist_path
1384        # check if we have an existing item to retrieve the artist path
1385        async for item in self.mass.music.artists.iter_library_items(
1386            search=candidate_name, provider=self.instance_id
1387        ):
1388            if not compare_strings(candidate_name, item.name):
1389                continue
1390            for prov_mapping in item.provider_mappings:
1391                if prov_mapping.provider_instance == self.instance_id and prov_mapping.url:
1392                    return prov_mapping.url
1393        return None
1394
1395    async def _resolve_artist_representative_track(self, artist: Artist) -> str | None:
1396        """
1397        Return one of this artist's own track paths, to register a metadata-file baseline.
1398
1399        Needed for a manual "Refresh item", which doesn't go through `_parse_track`/
1400        `_parse_album` and would otherwise leave a freshly read artist.nfo/image
1401        unregistered. Falls back to an album-only artist's own albums (e.g. credited only
1402        as ALBUMARTIST, never as a track artist) - a bounded, first-success lookup; not
1403        optimized further since this only runs for a one-off manual refresh.
1404
1405        :param artist: The library artist whose own tracks are searched.
1406        """
1407        for track in await self.mass.music.artists.get_library_artist_tracks(
1408            artist.item_id, provider_filter=self.instance_id
1409        ):
1410            if path := self._track_representative_path(track):
1411                return path
1412        for album in await self.mass.music.artists.get_library_artist_albums(
1413            artist.item_id, provider_filter=self.instance_id
1414        ):
1415            for track in await self.mass.music.albums.get_library_album_tracks(
1416                album.item_id, provider_filter=[self.instance_id]
1417            ):
1418                if path := self._track_representative_path(track):
1419                    return path
1420        return None
1421
1422    def _track_representative_path(self, track: Track) -> str | None:
1423        """Return this instance's own resolvable path (or CUE sheet path) for one track."""
1424        for prov_mapping in track.provider_mappings:
1425            if prov_mapping.provider_instance != self.instance_id or not prov_mapping.available:
1426                # an unavailable mapping is a stale library row, not a resolvable path: using
1427                # it anyway could point the caller's next `_scandir`/`exists` at an already
1428                # removed folder instead of trying another candidate track
1429                continue
1430            # a CUE-derived track's mapping is a synthetic "<cue path>::<track>" id, not
1431            # itself a resolvable path; its CUE sheet is, and reprocessing that sheet
1432            # already refreshes every track (and this artist) it describes
1433            if parsed := parse_cue_track_id(prov_mapping.item_id):
1434                return parsed[0]
1435            return prov_mapping.item_id
1436        return None
1437
1438    async def _find_artist_by_folder_name(self, folder_name: str) -> Artist | None:
1439        """
1440        Return the one library artist (on this provider) whose name or sort-name matches.
1441
1442        Used to recover a synthetic artist's identity on the second, not-yet-persisted
1443        fetch of a path a normal folder/sort-name-alias match just resolved onto.
1444
1445        :param folder_name: The resolved folder's own basename.
1446        """
1447        matches = [
1448            item
1449            async for item in self.mass.music.artists.iter_library_items(provider=self.instance_id)
1450            if compare_strings(folder_name, item.name)
1451            or (item.sort_name and compare_strings(folder_name, item.sort_name))
1452        ]
1453        return matches[0] if len(matches) == 1 else None
1454
1455    async def _drop_stale_album_artist_caches(self) -> None:
1456        """
1457        Drop this provider's own short-lived album/artist/folder-image caches.
1458
1459        Called once, before this sync's batch starts processing, whenever at least one
1460        metadata-file change queued a representative for reparsing. A queued representative's
1461        reparse must see that change, not a stale object left by a read within the last 120
1462        seconds; without this, a concurrent parse of the same folder (whether the queued
1463        representative itself or an unrelated sibling track reprocessed for another reason in
1464        this same batch) could read - or worse, write back to the database - the pre-change
1465        data, silently undoing the refresh.
1466        """
1467        for stale_category in (
1468            CACHE_CATEGORY_ALBUM_INFO,
1469            CACHE_CATEGORY_ARTIST_INFO,
1470            CACHE_CATEGORY_FOLDER_IMAGES,
1471        ):
1472            await self.cache.delete(key=None, category=stale_category, provider=self.instance_id)
1473
1474    async def _queue_changed_metadata_files(
1475        self,
1476        metadata_files: list[FileSystemItem],
1477        file_checksums: dict[str, str],
1478        cue_file_checksums: dict[str, set[str]],
1479        items_to_process: list[tuple[FileSystemItem, str | None]],
1480        force_refresh_tracks: set[str],
1481    ) -> None:
1482        """
1483        Queue the representative track of each changed local metadata file for reparsing.
1484
1485        A metadata file (NFO or recognized folder image) with no registered cache entry is a
1486        new or never-parsed file and is ignored here; a full reparse (a manual refresh, or the
1487        next time its representative track changes for any other reason) registers it. A
1488        registered file whose token is unchanged is also ignored. Only a registered file whose
1489        token changed queues its representative track, deduplicated against tracks already
1490        queued by this sync (whether from another changed metadata file or their own change).
1491
1492        This never writes the metadata-file cache itself: only actually reparsing the
1493        representative track (which re-reads the file) advances its registered token, so a
1494        failed parse is retried next sync instead of being silently marked as handled. A
1495        changed recognized image also has its own image cache invalidated unconditionally,
1496        even when its representative reparse itself is deduplicated away.
1497
1498        :param metadata_files: Local metadata files collected by this sync's walk.
1499        :param file_checksums: Previously stored checksum per provider item id.
1500        :param cue_file_checksums: Previously stored track checksums keyed by CUE relative_path.
1501        :param items_to_process: The sync's changed/new items; receives queued representatives.
1502        :param force_refresh_tracks: Receives the relative_path of every representative that
1503            was actually queued (new or deduplicated against one already queued); a non-empty
1504            result tells the caller to drop this provider's own short-lived album/artist/
1505            folder-image caches before processing, so a concurrent parse of the same folder
1506            cannot hand back or write back the pre-change data. Left empty when nothing was
1507            actually queued (e.g. every representative failed to resolve).
1508        """
1509        # one bulk load instead of one query per metadata file, which otherwise would mean
1510        # thousands of sequential cache reads on a large library every single sync
1511        registrations = await self.cache.get_all(
1512            provider=self.instance_id, category=CACHE_CATEGORY_METADATA_FILE
1513        )
1514        queued_tracks = {item.relative_path for item, _ in items_to_process}
1515        for meta_item in metadata_files:
1516            cached = registrations.get(meta_item.relative_path)
1517            if not cached:
1518                continue
1519            if cached.get("token") == meta_item.metadata_change_token:
1520                continue
1521            track_path = cached.get("track")
1522            if not track_path:
1523                continue
1524            if is_image_file(meta_item):
1525                # invalidate unconditionally, even when the representative reparse below is
1526                # deduplicated away (e.g. the track itself also changed, or another metadata
1527                # file already queued it this sync): the image keeps the same (provider, path)
1528                # identity, so skipping this here would leave its old bytes cached regardless
1529                await self.mass.metadata.invalidate_image_cache(
1530                    self.instance_id, meta_item.relative_path
1531                )
1532            if track_path in queued_tracks:
1533                # already queued for another reason (its own content changed, or another
1534                # metadata file got here first): still needs the stale-cache drop below, since
1535                # that reparse must see this change too
1536                force_refresh_tracks.add(track_path)
1537                continue
1538            try:
1539                track_item = await self.resolve(track_path)
1540            except MediaNotFoundError, OSError:
1541                # the representative no longer resolves: leave the old token in place so a
1542                # future sync (once a fresh representative registers) or a manual refresh
1543                # can recover; this change is deferred, not lost - and since nothing is
1544                # actually queued, the stale-cache drop below is not needed either
1545                continue
1546            if track_item.is_dir:
1547                continue
1548            queued_tracks.add(track_path)
1549            force_refresh_tracks.add(track_path)
1550            if track_item.ext in CUE_EXTENSIONS:
1551                # a CUE sheet's own checksum is not tracked under its path in file_checksums,
1552                # only the synthetic per-track ids it expands into are
1553                prev_checksum = min(cue_file_checksums.get(track_path, set()), default=None)
1554            else:
1555                prev_checksum = file_checksums.get(track_path)
1556            items_to_process.append((track_item, prev_checksum))
1557
1558    async def _register_metadata_file(
1559        self, item: FileSystemItem, representative_track: str | None
1560    ) -> None:
1561        """
1562        Remember a local metadata file's current token and its representative track.
1563
1564        Called whenever album/artist parsing reads an NFO file or enumerates a recognized
1565        folder image, so a later sync can detect the file changing on disk and reparse just
1566        that one representative track. This cache is derivative, not authoritative: it is
1567        never consulted by parsing itself, only by the sync walk's change detection.
1568
1569        :param item: The metadata file (NFO or recognized folder image) that was just read.
1570        :param representative_track: The track whose reparse rebuilds this item; skipped when
1571            falsy, since there is then nothing useful to reparse later.
1572        """
1573        if not representative_track:
1574            return
1575        await self.cache.set(
1576            key=item.relative_path,
1577            data={"token": item.metadata_change_token, "track": representative_track},
1578            provider=self.instance_id,
1579            category=CACHE_CATEGORY_METADATA_FILE,
1580            expiration=METADATA_FILE_CACHE_EXPIRATION,
1581            # this is functional registration data the feature depends on, not a disposable
1582            # cache: keep it out of a generic "clear cache" action
1583            persistent=True,
1584        )
1585
1586    @staticmethod
1587    def _add_nfo_candidate(by_name: dict[str, FileSystemItem], item: FileSystemItem) -> None:
1588        """
1589        Add one recognized NFO file to a per-directory candidate map, order-independently.
1590
1591        Two files differing only in case (e.g. ``album.nfo`` and ``ALBUM.NFO`` in the same
1592        directory - an unusual layout, but possible on a case-sensitive filesystem) must resolve
1593        to the same one regardless of directory-listing order, so the sync's walk and an
1594        on-demand listing never disagree: the lexicographically first literal filename always
1595        wins, deterministically.
1596        """
1597        name = item.filename.lower()
1598        if name not in NFO_FILENAMES:
1599            return
1600        existing = by_name.get(name)
1601        if existing is None or item.filename < existing.filename:
1602            by_name[name] = item
1603
1604    @staticmethod
1605    def _build_nfo_index(
1606        metadata_files: list[FileSystemItem],
1607    ) -> dict[str, dict[str, FileSystemItem]]:
1608        """Group this sync's walked album.nfo/artist.nfo by parent directory."""
1609        index: dict[str, dict[str, FileSystemItem]] = {}
1610        for item in metadata_files:
1611            # metadata_files also carries folder artwork; skip those before creating a
1612            # per-directory entry, or every image-bearing folder would get an empty one
1613            if item.filename.lower() not in NFO_FILENAMES:
1614                continue
1615            LocalFileSystemProvider._add_nfo_candidate(
1616                index.setdefault(item.relative_parent_path, {}), item
1617            )
1618        return index
1619
1620    @contextlib.contextmanager
1621    def _ondemand_listing_scope(self) -> Iterator[None]:
1622        """
1623        Memoize on-demand folder listings for one outermost parse.
1624
1625        A no-op once the sync's own NFO index is ready (lookups are already O(1) from it) or
1626        when already inside an outer scope, so nested album/artist resolution within one
1627        parse shares a single memo and reuses a folder's already-completed listing - though
1628        two concurrent lookups racing the very first listing of a folder can still both list
1629        it, bounded by concurrency - including for a track processed while the sync's walk is
1630        still in progress, or the whole batch of tracks processed while an incomplete scan
1631        left that index unready. A forced refresh never takes that index shortcut regardless
1632        (see :meth:`_nfo_item_for`),
1633        so it still needs - and gets - its own memo here.
1634        """
1635        if (self._sync_nfo_index_ready and not BYPASS_CACHE.get()) or (
1636            _ONDEMAND_NFO_ITEMS.get() is not None
1637        ):
1638            yield
1639            return
1640        items_token = _ONDEMAND_NFO_ITEMS.set({})
1641        try:
1642            yield
1643        finally:
1644            _ONDEMAND_NFO_ITEMS.reset(items_token)
1645
1646    async def _nfo_item_for(self, folder: str, filename: str) -> FileSystemItem | None:
1647        """
1648        Return a folder's named NFO file, from the sync index or a listing on demand.
1649
1650        Listing (not a direct path probe) matches the sync's own case-insensitive NFO
1651        recognition. A forced refresh (``BYPASS_CACHE``) always lists directly, even during
1652        a concurrent background sync, since that sync's index is a provider-wide snapshot
1653        that could otherwise still serve the stale listing the refresh was meant to bypass.
1654        Outside that index (before it is built, or an incomplete scan left it unready), the
1655        per-parse on-demand memo below still applies, since `sync_library` wraps its whole
1656        batch in one shared :meth:`_ondemand_listing_scope`, not a scope per track.
1657
1658        :param folder: The candidate directory to look in.
1659        :param filename: ``album.nfo`` or ``artist.nfo``.
1660        """
1661        if self._sync_nfo_index_ready and not BYPASS_CACHE.get():
1662            return self._sync_nfo_by_dir.get(folder, {}).get(filename)
1663        memo = _ONDEMAND_NFO_ITEMS.get()
1664        if memo is not None:
1665            if folder not in memo:
1666                memo[folder] = await self._list_nfo_candidates(folder)
1667            candidates = memo[folder]
1668        else:
1669            candidates = await self._list_nfo_candidates(folder)
1670        return candidates.get(filename)
1671
1672    async def _list_nfo_candidates(self, folder: str) -> dict[str, FileSystemItem]:
1673        """
1674        Return a folder's recognized NFO files, keyed by lowercase filename.
1675
1676        A listing failure is not caught here: every bounded candidate directory is either the
1677        track's own directory or an ancestor of it, so it necessarily exists; a raised error is
1678        therefore a genuine transient storage failure and must propagate just like a `_read_file`
1679        failure, so the caller (and, during a sync, `_process_item_async`) can defer and retry
1680        instead of silently treating the folder as having no NFO.
1681        """
1682        # bypass a cloud-backed provider's own short-lived listing cache during an explicit
1683        # "Refresh item", so an NFO just added to disk is seen right away instead of only
1684        # after that cache naturally expires
1685        items = await self._scandir(folder, use_cache=not BYPASS_CACHE.get())
1686        by_name: dict[str, FileSystemItem] = {}
1687        for item in items:
1688            if not item.is_dir:
1689                self._add_nfo_candidate(by_name, item)
1690        return by_name
1691
1692    async def _load_nfo_root(
1693        self, nfo_item: FileSystemItem, root_tag: str
1694    ) -> dict[str, Any] | None:
1695        """
1696        Read and parse one NFO file, returning its root element or None when malformed.
1697
1698        :param nfo_item: The NFO file to read.
1699        :param root_tag: The expected root element name (``album`` or ``artist``).
1700        """
1701        raw = await self._read_file(nfo_item.relative_path)
1702        return await asyncio.to_thread(parse_nfo_root, raw, root_tag)
1703
1704    def _nfo_applies_cleanly(self, root: dict[str, Any], kind: str) -> bool:
1705        """
1706        Return True when an NFO root's consumed fields are well-shaped and apply without error.
1707
1708        Used before trusting an NFO as folder identity, so a title/id that happens to match but
1709        carries an invalid field never resolves a directory that enrichment would then reject
1710        anyway. Every field ``parse_album_nfo``/``parse_artist_nfo`` reads must be a plain scalar
1711        (not a repeated/nested XML element reaching a string-only assignment or helper without
1712        raising), and a present MusicBrainz id must actually be a valid UUID, not merely absent.
1713
1714        :param root: The parsed NFO root.
1715        :param kind: ``album`` or ``artist``.
1716        """
1717        fields, mbid_fields = (
1718            (_ALBUM_NFO_FIELDS, _ALBUM_MBID_FIELDS)
1719            if kind == "album"
1720            else (_ARTIST_NFO_FIELDS, _ARTIST_MBID_FIELDS)
1721        )
1722        for field in fields:
1723            value = root.get(field)
1724            if value is None:
1725                continue
1726            if field in _LIST_ALLOWED_NFO_FIELDS:
1727                if not isinstance(value, str) and not (
1728                    isinstance(value, list | tuple) and all(isinstance(v, str) for v in value)
1729                ):
1730                    return False
1731            elif not isinstance(value, str):
1732                return False
1733        for field in mbid_fields:
1734            raw = root.get(field)
1735            if isinstance(raw, str) and raw.strip() and clean_mbid(raw) is None:
1736                return False
1737        try:
1738            if kind == "album":
1739                scratch_album = Album(
1740                    item_id="", provider=self.instance_id, name="", provider_mappings=set()
1741                )
1742                parse_album_nfo(scratch_album, root)
1743            else:
1744                scratch_artist = Artist(
1745                    item_id="", provider=self.instance_id, name="", provider_mappings=set()
1746                )
1747                parse_artist_nfo(scratch_artist, root)
1748        except ValueError, TypeError, AttributeError:
1749            # AttributeError covers a non-scalar field shape (e.g. a repeated/nested XML
1750            # element) reaching a string-only helper such as split_items
1751            return False
1752        return True
1753
1754    async def _resolve_album_dir_via_nfo(
1755        self, track_dir: str, tags: AudioTags, rejected: set[str] | None = None
1756    ) -> tuple[str, FileSystemItem, dict[str, Any]] | None:
1757        """
1758        Return ``(album_dir, nfo_item, root)`` resolved from a validated album.nfo, or None.
1759
1760        Bounded to ``track_dir`` and its immediate parent: a recognized disc subfolder's own
1761        album.nfo is never trusted as identity, only its parent's is, and the provider's own
1762        root is never a candidate either (like the normal, non-NFO folder match, it can never
1763        identify one specific album out of the many the root may contain). ``track_dir`` itself
1764        is tried first - it is the nearer, more specific candidate - before falling back to the
1765        parent, so a same-title album.nfo one level up (e.g. a stray leftover from a prior,
1766        flatter layout) can never outrank the track's own, definitively correct one. A
1767        candidate's album.nfo must positively match this track's MusicBrainz album/release-
1768        group id or its album title, and apply cleanly, before it is trusted; a malformed or
1769        non-matching NFO leaves the album unresolved (synthetic, tag-only).
1770
1771        :param track_dir: The directory the track file lives in.
1772        :param tags: The track's audio tags, matched against a candidate NFO.
1773        :param rejected: When given, every folder whose own album.nfo exists but fails to
1774            positively identify this album is added here, so a later relaxed (fuzzy/layout/
1775            date-prefix) match landing on that same folder knows not to trust that same
1776            rejected file.
1777        """
1778        album_id = clean_mbid(tags.musicbrainz_albumid, tags.filename)
1779        rg_id = clean_mbid(tags.musicbrainz_releasegroupid, tags.filename)
1780        album_name, album_version = (
1781            parse_title_and_version(tags.album) if tags.album else (None, "")
1782        )
1783        album_artist_ids = tuple(
1784            cleaned
1785            for raw_id in tags.musicbrainz_albumartistids
1786            if (cleaned := clean_mbid(raw_id, tags.filename))
1787        )
1788        candidates: list[str] = []
1789        if track_dir and not is_disc_dir(Path(track_dir).name):
1790            candidates.append(track_dir)
1791        candidates.extend(d for d in (os.path.dirname(track_dir),) if d)
1792        for folder in candidates:
1793            nfo_item = await self._nfo_item_for(folder, "album.nfo")
1794            if nfo_item is None:
1795                continue
1796            root = await self._load_nfo_root(nfo_item, "album")
1797            if (
1798                root is not None
1799                and self._album_nfo_matches(
1800                    root, album_id, rg_id, album_name, album_artist_ids, album_version
1801                )
1802                and self._nfo_applies_cleanly(root, "album")
1803            ):
1804                return folder, nfo_item, root
1805            if root is not None and rejected is not None:
1806                rejected.add(folder)
1807        return None
1808
1809    @staticmethod
1810    def _album_nfo_matches(
1811        root: dict[str, Any],
1812        album_id: str | None,
1813        rg_id: str | None,
1814        album_name: str | None,
1815        album_artist_ids: tuple[str, ...] = (),
1816        album_version: str = "",
1817    ) -> bool:
1818        """Return True when an album.nfo positively identifies this track's album."""
1819        nfo_artist_id = clean_mbid(root.get("musicbrainzalbumartistid"), "musicbrainzalbumartistid")
1820        if nfo_artist_id and album_artist_ids and nfo_artist_id not in album_artist_ids:
1821            # the NFO names a different album artist than this track's tags: reject even a
1822            # matching title or album id, since it may belong to another artist's same-named
1823            # or same-catalog-numbered album
1824            return False
1825        comparisons: list[bool] = []
1826        for field, tag_id in (
1827            ("musicbrainzalbumid", album_id),
1828            ("musicbrainzreleasegroupid", rg_id),
1829        ):
1830            if not tag_id:
1831                continue
1832            nfo_id = clean_mbid(root.get(field), field)
1833            if nfo_id:
1834                comparisons.append(nfo_id == tag_id)
1835        if comparisons:
1836            # every comparable id must agree; one mismatch rejects even if another id matched
1837            return all(comparisons)
1838        if not album_name:
1839            return False
1840        nfo_title = root.get("title") or root.get("name")
1841        if not nfo_title:
1842            return False
1843        # strip the NFO title's own edition/version suffix the same way the track's tag-derived
1844        # album name already was (matching how parse_album_nfo treats the title once applied),
1845        # so e.g. "Album (Deluxe Edition)" compares as "Album" on both sides
1846        nfo_name, nfo_version = parse_title_and_version(str(nfo_title))
1847        if album_version and nfo_version and not compare_strings(album_version, nfo_version, False):
1848            # both sides name a specific edition (e.g. "Live" vs "Remix"): never a match on
1849            # the base title alone, even though it is otherwise identical on both sides
1850            return False
1851        # strict comparison: folder identity must not be granted on a fuzzy/near match (e.g.
1852        # "Album 1" vs "Album 2"), only an (almost) exact one after normalization
1853        return compare_strings(nfo_name, album_name)
1854
1855    async def _resolve_artist_dir_via_nfo(
1856        self, album_dir: str, name: str, mbid: str | None, rejected: set[str] | None = None
1857    ) -> tuple[str, FileSystemItem, dict[str, Any]] | None:
1858        """
1859        Return ``(artist_path, nfo_item, root)`` resolved from a validated artist.nfo, or None.
1860
1861        Walks the same bounded ancestor levels as the normal folder-name lookup, from the album
1862        directory's parent upward, but never as far as the provider's own root: like the normal
1863        folder-name match, an artist.nfo there could not identify one specific artist out of the
1864        many the root may contain. The first ancestor whose artist.nfo positively matches this
1865        artist's MusicBrainz id or name - and applies cleanly - is trusted; a mismatching,
1866        malformed or marker-only (no id or name) NFO is skipped and the walk continues.
1867
1868        :param album_dir: The album directory whose ancestors are searched.
1869        :param name: The artist name to match.
1870        :param mbid: The artist's cleaned MusicBrainz id, if any.
1871        :param rejected: When given, every folder whose own artist.nfo exists but fails to
1872            positively identify this artist is added here, so a later relaxed (fuzzy/alias)
1873            match landing on that same folder knows not to trust that same rejected file.
1874        """
1875        parentdir = os.path.dirname(album_dir)
1876        for _ in range(3):
1877            if not parentdir:
1878                break
1879            nfo_item = await self._nfo_item_for(parentdir, "artist.nfo")
1880            if nfo_item is not None:
1881                root = await self._load_nfo_root(nfo_item, "artist")
1882                if (
1883                    root is not None
1884                    and self._artist_nfo_matches(root, name, mbid)
1885                    and self._nfo_applies_cleanly(root, "artist")
1886                ):
1887                    return parentdir, nfo_item, root
1888                if rejected is not None:
1889                    rejected.add(parentdir)
1890            parentdir = os.path.dirname(parentdir)
1891        return None
1892
1893    @staticmethod
1894    def _artist_nfo_matches(root: dict[str, Any], name: str, mbid: str | None) -> bool:
1895        """Return True when an artist.nfo positively identifies this artist by id or name."""
1896        if mbid:
1897            nfo_mbid = clean_mbid(root.get("musicbrainzartistid"), "artist.nfo")
1898            if nfo_mbid:
1899                return nfo_mbid == mbid
1900        nfo_name = root.get("title") or root.get("name")
1901        # strict comparison: folder identity must not be granted on a fuzzy/near match (e.g.
1902        # "Artist 1" vs "Artist 2"), only an (almost) exact one after normalization
1903        return bool(nfo_name) and compare_strings(str(nfo_name), name)
1904
1905    async def _iter_album_tracks(self, prov_album_id: str) -> AsyncGenerator[Track]:
1906        """
1907        Yield an album's tracks, lazily, so a caller needing only the first can stop early.
1908
1909        Served from the database when this id is already mapped; otherwise (the second,
1910        id-changed fetch of a manual "Refresh item" that has just resolved a previously
1911        synthetic album onto its real folder, before that mapping is persisted) the folder's
1912        own tracks are parsed directly, so the refresh still succeeds instead of raising on a
1913        mapping that is only about to exist.
1914
1915        :param prov_album_id: This provider's album id (a path, once resolved).
1916        """
1917        db_album = await self.mass.music.albums.get_library_item_by_prov_id(
1918            prov_album_id, self.instance_id
1919        )
1920        if db_album is not None:
1921            album_tracks = await self.mass.music.albums.get_library_album_tracks(db_album.item_id)
1922            for track in album_tracks:
1923                if any(x.provider_instance == self.instance_id for x in track.provider_mappings):
1924                    yield track
1925            return
1926        if not await self.exists(prov_album_id):
1927            msg = f"Album not found: {prov_album_id}"
1928            raise MediaNotFoundError(msg)
1929        # one shared scope for the whole scan, not one per track: every _parse_track call below
1930        # would otherwise open and close its own scope, re-listing the same candidate ancestor
1931        # folders once per track instead of once for the album
1932        with self._ondemand_listing_scope():
1933            async for track in self._scan_folder_tracks(prov_album_id):
1934                yield track
1935
1936    async def _scan_folder_tracks(self, folder: str) -> AsyncIterator[Track]:
1937        """
1938        Yield every parseable track under a folder or one of its immediate child directories.
1939
1940        The folder itself is scanned first; a child directory is only listed if the caller
1941        needs more than the folder alone provided, so a single-track lookup (:meth:`get_album`)
1942        skips subfolder listings entirely on a cloud/WebDAV backend. Every child is tried, not
1943        only regex-recognized disc folders, since the album may live in an arbitrarily named
1944        subfolder; the ``track.album.item_id == folder`` filter keeps only tracks belonging to
1945        this exact folder.
1946
1947        :param folder: The folder to scan; also every one of its immediate child directories.
1948        """
1949        # shared across this folder and every child directory below, so a single CUE sheet
1950        # naming a child folder's companion audio file (e.g. one CUE covering the whole
1951        # album, split across "Disc 1"/"Disc 2" subfolders) still excludes that companion
1952        # once its own subfolder is scanned, instead of parsing it again as a duplicate track
1953        cue_stems: set[str] = set()
1954        entries = await self._scandir(folder)
1955        async for track in self._yield_scanned_tracks(folder, entries, cue_stems):
1956            yield track
1957        for entry in entries:
1958            if entry.is_dir:
1959                child_entries = await self._scandir(entry.relative_path)
1960                async for track in self._yield_scanned_tracks(folder, child_entries, cue_stems):
1961                    yield track
1962
1963    async def _yield_scanned_tracks(
1964        self, folder: str, items: list[FileSystemItem], cue_stems: set[str]
1965    ) -> AsyncIterator[Track]:
1966        """
1967        Yield every parseable track among one directory's own (already listed) items.
1968
1969        A CUE's own file and the companion audio file it names are never yielded separately -
1970        only the CUE's own segmented tracks represent them, mirroring the sync/browse paths'
1971        CUE-stem exclusion. A missing companion audio file or unparsable CUE sheet is checked
1972        explicitly before constructing the CUE's album, and only that is treated as "this file
1973        is unreadable" and skipped. A failure past that point (constructing the album via this
1974        same NFO resolution) propagates instead, since a cloud/WebDAV `_read_file` also raises
1975        `MediaNotFoundError` for a transient read failure, not only a genuinely missing file.
1976        Used only by :meth:`_scan_folder_tracks`.
1977
1978        :param folder: The top-level folder this album resolved to, matched against each
1979            parsed track's own album identity.
1980        :param items: One directory's own listed entries (never a nested listing).
1981        :param cue_stems: Companion-audio stems already absorbed by a CUE sheet found in this
1982            folder or an earlier one in this same scan; grown here with this folder's own
1983            CUEs and still shared with every directory scanned afterwards.
1984        """
1985        cue_sheets: dict[str, CueSheet] = {}
1986        for item in items:
1987            if item.is_dir or item.ext not in CUE_EXTENSIONS:
1988                continue
1989            try:
1990                cue_sheet = await self._cue.load_cue_sheet(item)
1991            except InvalidDataError:
1992                continue
1993            if not cue_sheet.tracks:
1994                # parses cleanly but names no tracks (e.g. truncated/malformed content):
1995                # excluding its same-named companion here would lose that otherwise
1996                # playable file entirely, since this CUE itself yields nothing below
1997                continue
1998            cue_stems.add(item.absolute_path.rsplit(".", 1)[0])
1999            cue_sheets[item.relative_path] = cue_sheet
2000            if companion_stem := cue_referenced_audio_stem(item, cue_sheet):
2001                cue_stems.add(companion_stem)
2002        for item in items:
2003            if item.is_dir:
2004                continue
2005            if item.ext in CUE_EXTENSIONS:
2006                loaded_cue_sheet = cue_sheets.get(item.relative_path)
2007                if loaded_cue_sheet is None:
2008                    self.logger.warning("Skipping unreadable CUE sheet %s", item.relative_path)
2009                    continue
2010                if await self._cue.find_audio_file(item, loaded_cue_sheet) is None:
2011                    self.logger.warning(
2012                        "Skipping CUE sheet with missing companion audio file: %s",
2013                        item.relative_path,
2014                    )
2015                    continue
2016                try:
2017                    # InvalidDataError here covers the audio file's own unreadable/corrupt
2018                    # tags (e.g. no determinable duration); a MediaNotFoundError past this
2019                    # point can only come from constructing this CUE's album (this call's own
2020                    # NFO resolution reading an NFO file) or a similarly transient storage
2021                    # failure, and must propagate instead of being mistaken for a bad CUE
2022                    cue_tracks = await self._cue.parse_tracks(item)
2023                except InvalidDataError as err:
2024                    self.logger.warning(
2025                        "Skipping unreadable CUE sheet %s: %s", item.relative_path, err
2026                    )
2027                    continue
2028                for track in cue_tracks:
2029                    if isinstance(track.album, Album) and track.album.item_id == folder:
2030                        yield track
2031            elif item.ext in TRACK_EXTENSIONS:
2032                if item.absolute_path.rsplit(".", 1)[0] in cue_stems:
2033                    continue  # absorbed into its CUE sheet's own segmented tracks
2034                try:
2035                    tags = await async_parse_tags(item.absolute_path, item.file_size)
2036                except InvalidDataError as err:
2037                    self.logger.warning("Skipping unreadable track %s: %s", item.relative_path, err)
2038                    continue
2039                track = await self._parse_track(item, tags)
2040                if isinstance(track.album, Album) and track.album.item_id == folder:
2041                    yield track
2042
2043    def _set_available(self, available: bool) -> None:
2044        """Update the provider availability and notify listeners on change."""
2045        if self.available == available:
2046            return
2047        self.available = available
2048        if available:
2049            self._cancel_availability_probe()
2050        else:
2051            self._schedule_availability_probe()
2052        self.mass.signal_event(EventType.PROVIDERS_UPDATED, data=self.mass.get_providers())
2053
2054    async def _is_reachable(self) -> bool:
2055        """Return whether the storage backing this provider can be read."""
2056        return bool(await isdir(self.base_path))
2057
2058    @property
2059    def _availability_probe_id(self) -> str:
2060        """Return the timer id of this provider's reachability checks."""
2061        return f"filesystem_availability_probe_{self.instance_id}"
2062
2063    def _schedule_availability_probe(self) -> None:
2064        """Arm the next reachability check."""
2065        self.mass.call_later(
2066            AVAILABILITY_PROBE_INTERVAL,
2067            self._probe_availability,
2068            task_id=self._availability_probe_id,
2069        )
2070
2071    def _cancel_availability_probe(self) -> None:
2072        """Stop checking for the storage coming back."""
2073        self.mass.cancel_timer(self._availability_probe_id)
2074
2075    async def _probe_availability(self) -> None:
2076        """Mark the provider available again once its storage can be read."""
2077        try:
2078            reachable = await self._is_reachable()
2079        except MusicAssistantError as err:
2080            # storage that is simply still gone, which is what this loop waits for
2081            self.logger.debug("%s is still unreachable: %s", self.name, err)
2082            reachable = False
2083        except Exception:
2084            # an unexpected failure must not end the loop, since it is what brings the
2085            # provider back, but it is a defect rather than an outage so it is logged loudly
2086            self.logger.exception("Reachability check for %s failed", self.name)
2087            reachable = False
2088        if self.unloading:
2089            # the provider was torn down while this check was running; re-arming here
2090            # would leave a timer firing against an instance nothing owns anymore
2091            return
2092        if reachable:
2093            self.logger.info("%s is reachable again", self.name)
2094            self._set_available(True)
2095            return
2096        self._schedule_availability_probe()
2097
2098    async def _process_item_async(
2099        self,
2100        item: FileSystemItem,
2101        prev_checksum: str | None,
2102        cur_filenames: set[str] | None = None,
2103        cue_stems: set[str] | None = None,
2104        prev_filenames: set[str] | None = None,
2105    ) -> bool:
2106        """
2107        Process a single item asynchronously.
2108
2109        :param item: The filesystem item to process.
2110        :param prev_checksum: Previous checksum from the database, or None for new items.
2111        :param cur_filenames: Set of current filenames being tracked (for CUE track IDs).
2112        :param cue_stems: Absolute paths (without extension) of CUE sheets in this scan,
2113            used to detect companion-CUE audio files without a filesystem stat.
2114        :param prev_filenames: The ids/paths the previous scan found, used to keep the
2115            ids of a CUE sheet that fails to parse.
2116        """
2117        try:
2118            self.logger.log(VERBOSE_LOG_LEVEL, "Processing: %s", item.relative_path)
2119
2120            if prev_checksum is not None:
2121                # the file changed on disk: drop cached artwork derived from it
2122                # (thumbnails, source bytes, palette) so re-read embedded art is
2123                # served fresh, for both reference forms of the image path
2124                await self.mass.metadata.invalidate_image_cache(
2125                    self.instance_id, item.relative_path
2126                )
2127                await self.mass.metadata.invalidate_image_cache(
2128                    self.instance_id, self._versioned_image_path(item.relative_path, prev_checksum)
2129                )
2130
2131            if item.ext in CUE_EXTENSIONS and self.media_content_type == "music":
2132                tracks = await self._cue.parse_tracks(item)
2133                for track in tracks:
2134                    track.favorite = False
2135                    await self.mass.music.tracks.add_item_to_library(
2136                        track, overwrite_existing=prev_checksum is not None
2137                    )
2138                    if cur_filenames is not None:
2139                        cur_filenames.add(track.item_id)
2140                return True
2141
2142            if item.ext in TRACK_EXTENSIONS and self.media_content_type == "music":
2143                if not self._sync_tracks:
2144                    return False
2145                # skip audio files that have a companion CUE sheet
2146                if cue_stems is not None and item.absolute_path.rsplit(".", 1)[0] in cue_stems:
2147                    return False
2148                tags = await async_parse_tags(item.absolute_path, item.file_size)
2149                track = await self._parse_track(item, tags)
2150                track.favorite = False  # TODO: implement favorite status based on rating ?
2151                await self.mass.music.tracks.add_item_to_library(
2152                    track, overwrite_existing=prev_checksum is not None
2153                )
2154                return True
2155
2156            if item.ext in AUDIOBOOK_EXTENSIONS and self.media_content_type == "audiobooks":
2157                tags = await async_parse_tags(item.absolute_path, item.file_size)
2158                try:
2159                    audiobook = await self._parse_audiobook(item, tags)
2160                except IsChapterFile:
2161                    return True
2162                await self.mass.music.audiobooks.add_item_to_library(
2163                    audiobook, overwrite_existing=prev_checksum is not None
2164                )
2165                return True
2166
2167            if item.ext in PODCAST_EPISODE_EXTENSIONS and self.media_content_type == "podcasts":
2168                tags = await async_parse_tags(item.absolute_path, item.file_size)
2169                episode = await self._parse_podcast_episode(item, tags)
2170                assert isinstance(episode.podcast, Podcast)
2171                await self.mass.music.podcasts.add_item_to_library(
2172                    episode.podcast, overwrite_existing=prev_checksum is not None
2173                )
2174                return True
2175
2176            if item.ext in PLAYLIST_EXTENSIONS and self.media_content_type == "music":
2177                if not self._sync_playlists:
2178                    return False
2179                playlist = await self.get_playlist(item.relative_path)
2180                await self.mass.music.playlists.add_item_to_library(
2181                    playlist, overwrite_existing=prev_checksum is not None
2182                )
2183                return True
2184
2185        except Exception as err:
2186            # we don't want the whole sync to crash on one file so we catch all exceptions here
2187            self.logger.error(
2188                "Error processing %s - %s",
2189                item.relative_path,
2190                str(err),
2191                exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
2192            )
2193            report_current_task_failure(f"Failed to process {item.relative_path}: {err}")
2194            # the file is still on the storage, so keep it in the scan result:
2195            # leaving it out makes the deletion step treat it as removed
2196            self._keep_failed_item(item, cur_filenames, prev_filenames)
2197        return False
2198
2199    def _keep_failed_item(
2200        self,
2201        item: FileSystemItem,
2202        cur_filenames: set[str] | None,
2203        prev_filenames: set[str] | None,
2204    ) -> None:
2205        """
2206        Keep an item that could not be processed in the scan result.
2207
2208        :param item: The item that failed to process.
2209        :param cur_filenames: Receives the ids/paths present in this scan.
2210        :param prev_filenames: The ids/paths the previous scan found.
2211        """
2212        if cur_filenames is None:
2213            return
2214        cur_filenames.add(item.relative_path)
2215        if (
2216            item.ext not in CUE_EXTENSIONS
2217            or self.media_content_type != "music"
2218            or not prev_filenames
2219        ):
2220            return
2221        # a CUE sheet stands in for one id per track it describes and those cannot be
2222        # rebuilt without parsing it, so carry over the ids of the previous scan
2223        cur_filenames.update(
2224            item_id
2225            for item_id in prev_filenames
2226            if (parsed := parse_cue_track_id(item_id)) and parsed[0] == item.relative_path
2227        )
2228
2229    async def _process_orphaned_albums_and_artists(self) -> None:
2230        """Process deletion of orphaned albums and artists."""
2231        assert self.mass.music.database
2232        # Remove albums without any tracks
2233        query = (
2234            f"SELECT item_id FROM {DB_TABLE_ALBUMS} "
2235            f"WHERE item_id not in ( SELECT album_id from {DB_TABLE_ALBUM_TRACKS}) "
2236            f"AND item_id in ( SELECT item_id from {DB_TABLE_PROVIDER_MAPPINGS} "
2237            f"WHERE provider_instance = '{self.instance_id}' and media_type = 'album' )"
2238        )
2239        for db_row in await self.mass.music.database.get_rows_from_query(
2240            query,
2241            limit=100000,
2242        ):
2243            await self.mass.music.albums.remove_item_from_library(db_row["item_id"])
2244
2245        # Remove artists without any tracks or albums
2246        query = (
2247            f"SELECT item_id FROM {DB_TABLE_ARTISTS} "
2248            f"WHERE item_id not in "
2249            f"( select artist_id from {DB_TABLE_TRACK_ARTISTS} "
2250            f"UNION SELECT artist_id from {DB_TABLE_ALBUM_ARTISTS} ) "
2251            f"AND item_id in ( SELECT item_id from {DB_TABLE_PROVIDER_MAPPINGS} "
2252            f"WHERE provider_instance = '{self.instance_id}' and media_type = 'artist' )"
2253        )
2254        for db_row in await self.mass.music.database.get_rows_from_query(
2255            query,
2256            limit=100000,
2257        ):
2258            await self.mass.music.artists.remove_item_from_library(db_row["item_id"])
2259
2260    async def _process_deletions(self, deleted_files: set[str]) -> None:
2261        """Process all deletions."""
2262        # process deleted tracks/playlists
2263        album_ids = set()
2264        artist_ids = set()
2265        for file_path in deleted_files:
2266            if parse_cue_track_id(file_path) is not None and self.media_content_type == "music":
2267                controller = self.mass.music.get_controller(MediaType.TRACK)
2268            elif "." not in file_path:
2269                continue
2270            else:
2271                _, ext = file_path.rsplit(".", 1)
2272                if ext in PODCAST_EPISODE_EXTENSIONS and self.media_content_type == "podcasts":
2273                    controller = self.mass.music.get_controller(MediaType.PODCAST_EPISODE)
2274                elif ext in AUDIOBOOK_EXTENSIONS and self.media_content_type == "audiobooks":
2275                    controller = self.mass.music.get_controller(MediaType.AUDIOBOOK)
2276                elif ext in PLAYLIST_EXTENSIONS and self.media_content_type == "music":
2277                    controller = self.mass.music.get_controller(MediaType.PLAYLIST)
2278                elif ext in TRACK_EXTENSIONS and self.media_content_type == "music":
2279                    controller = self.mass.music.get_controller(MediaType.TRACK)
2280                else:
2281                    # unsupported file extension?
2282                    continue
2283
2284            if library_item := await controller.get_library_item_by_prov_id(
2285                file_path, self.instance_id
2286            ):
2287                if is_track(library_item):
2288                    if library_item.album:
2289                        album_ids.add(library_item.album.item_id)
2290                        # need to fetch the library album to resolve the itemmapping
2291                        db_album = await self.mass.music.albums.get_library_item(
2292                            library_item.album.item_id
2293                        )
2294                        for artist in db_album.artists:
2295                            artist_ids.add(artist.item_id)
2296                    for artist in library_item.artists:
2297                        artist_ids.add(artist.item_id)
2298                await controller.remove_item_from_library(library_item.item_id)
2299        # check if any albums need to be cleaned up
2300        for album_id in album_ids:
2301            if not await self.mass.music.albums.tracks(album_id, "library"):
2302                await self.mass.music.albums.remove_item_from_library(album_id)
2303        # check if any artists need to be cleaned up
2304        for artist_id in artist_ids:
2305            artist_albums = await self.mass.music.artists.albums(artist_id, "library")
2306            artist_tracks = await self.mass.music.artists.tracks(artist_id, "library")
2307            if not (artist_albums or artist_tracks):
2308                await self.mass.music.artists.remove_item_from_library(artist_id)
2309
2310    async def _get_playlist_local_image(self, file_item: FileSystemItem) -> MediaItemImage | None:
2311        """Return a local image alongside the playlist file (matching basename) if any."""
2312        cache_key = f"playlist_image.{file_item.relative_path}"
2313        cached = await self.cache.get(
2314            key=cache_key,
2315            provider=self.instance_id,
2316            category=CACHE_CATEGORY_FOLDER_IMAGES,
2317            base_class=MediaItemImage,
2318        )
2319        if cached is not None:
2320            return cached[0] if cached else None
2321        try:
2322            folder_files = await self._scandir(file_item.relative_parent_path)
2323        except OSError, MusicAssistantError:
2324            return None
2325        target = file_item.name.lower()
2326        result: MediaItemImage | None = None
2327        for item in folder_files:
2328            if item.is_dir or not item.ext:
2329                continue
2330            if item.ext.lower() not in IMAGE_EXTENSIONS:
2331                continue
2332            if item.name.lower() != target:
2333                continue
2334            result = MediaItemImage(
2335                type=ImageType.THUMB,
2336                path=item.relative_path,
2337                provider=self.instance_id,
2338                remotely_accessible=False,
2339            )
2340            break
2341        await self.cache.set(
2342            key=cache_key,
2343            data=[result.to_dict()] if result is not None else [],
2344            provider=self.instance_id,
2345            category=CACHE_CATEGORY_FOLDER_IMAGES,
2346            expiration=120,
2347        )
2348        return result
2349
2350    async def _parse_playlist_line(self, line: str, playlist_path: str) -> Track | None:
2351        """Try to parse a track from a playlist line."""
2352        try:
2353            line = line.replace("file://", "").strip()
2354            # try to resolve the filename (both normal and url decoded):
2355            # - relative to the playlist folder (normpath resolves parent .. references)
2356            # - as-is: an absolute path, or relative to our base path
2357            # candidates stay relative so subclasses with virtual paths (cloud,
2358            # webdav) resolve them too, instead of leaking the server CWD
2359            for _line in (line, urllib.parse.unquote(line)):
2360                if playlist_path:
2361                    normalized = posixpath.normpath(f"{playlist_path}/{_line}")
2362                    with contextlib.suppress(FileNotFoundError, MediaNotFoundError):
2363                        file_item = await self.resolve(normalized)
2364                        return await self._get_playlist_line_track(file_item)
2365                with contextlib.suppress(FileNotFoundError, MediaNotFoundError):
2366                    file_item = await self.resolve(_line)
2367                    return await self._get_playlist_line_track(file_item)
2368            # all attempts failed
2369            raise MediaNotFoundError("Invalid path/uri")
2370
2371        except MusicAssistantError as err:
2372            self.logger.warning("Could not parse %s to track: %s", line, str(err))
2373
2374        return None
2375
2376    async def _get_playlist_line_track(self, file_item: FileSystemItem) -> Track:
2377        """
2378        Return the track for a resolved playlist entry.
2379
2380        :param file_item: The resolved file the playlist entry points at.
2381        """
2382        # filesystem tracks are synced into the library, so prefer the database over
2383        # (expensive) tag parsing - this keeps loading large playlists fast
2384        library_track = await self.mass.music.tracks.get_library_item_by_prov_id(
2385            file_item.relative_path, self.instance_id
2386        )
2387        # only trust the library item if its mapping for this file is available: the file
2388        # just resolved, so an unavailable mapping is stale (e.g. the file was missing
2389        # during the last scan) and would wrongly exclude the track from playback
2390        if library_track is not None and any(
2391            mapping.provider_instance == self.instance_id
2392            and mapping.item_id == file_item.relative_path
2393            and mapping.available
2394            for mapping in library_track.provider_mappings
2395        ):
2396            # callers expect the provider item identity here (not the library one),
2397            # e.g. for duplicate detection when editing the playlist
2398            library_track.item_id = file_item.relative_path
2399            library_track.provider = self.instance_id
2400            library_track.uri = create_uri(
2401                MediaType.TRACK, self.instance_id, file_item.relative_path
2402            )
2403            return library_track
2404        # not (yet) in the library: parse the file tags
2405        tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
2406        return await self._parse_track(file_item, tags)
2407
2408    @staticmethod
2409    def _versioned_image_path(relative_path: str, checksum: str | None) -> str:
2410        """Append the file checksum so the image cache busts when the file is replaced."""
2411        if checksum:
2412            return f"{relative_path}?cs={checksum}"
2413        return relative_path
2414
2415    @staticmethod
2416    def _codec_type_from_tags(tags: AudioTags) -> ContentType:
2417        """Return the audio codec detected by ffprobe, if any."""
2418        if tags.raw and (streams := tags.raw.get("streams")):
2419            if codec_name := streams[0].get("codec_name"):
2420                return ContentType.try_parse(codec_name)
2421        return ContentType.UNKNOWN
2422
2423    async def _parse_track(
2424        self, file_item: FileSystemItem, tags: AudioTags, full_album_metadata: bool = False
2425    ) -> Track:
2426        """Parse full track details from file tags."""
2427        with self._ondemand_listing_scope():
2428            return await self._parse_track_impl(file_item, tags, full_album_metadata)
2429
2430    async def _parse_track_impl(
2431        self, file_item: FileSystemItem, tags: AudioTags, full_album_metadata: bool = False
2432    ) -> Track:
2433        """Parse full track details from file tags (implementation, see :meth:`_parse_track`)."""
2434        # ruff: noqa: PLR0915
2435        name, version = parse_title_and_version(tags.title, tags.version)
2436        track = Track(
2437            item_id=file_item.relative_path,
2438            provider=self.instance_id,
2439            name=name,
2440            sort_name=tags.title_sort,
2441            version=version,
2442            provider_mappings={
2443                ProviderMapping(
2444                    item_id=file_item.relative_path,
2445                    provider_domain=self.domain,
2446                    provider_instance=self.instance_id,
2447                    audio_format=AudioFormat(
2448                        content_type=ContentType.try_parse(file_item.ext or tags.format),
2449                        codec_type=self._codec_type_from_tags(tags),
2450                        sample_rate=tags.sample_rate,
2451                        bit_depth=tags.bits_per_sample,
2452                        channels=tags.channels,
2453                        bit_rate=tags.bit_rate,
2454                    ),
2455                    details=file_item.checksum,
2456                    in_library=True,
2457                )
2458            },
2459            disc_number=tags.disc or 0,
2460            track_number=tags.track or 0,
2461            date_added=(
2462                datetime.fromtimestamp(file_item.created_at, tz=UTC)
2463                if file_item.created_at
2464                else None
2465            ),
2466        )
2467
2468        if isrc_tags := tags.isrc:
2469            for isrsc in isrc_tags:
2470                track.external_ids.add((ExternalID.ISRC, isrsc))
2471
2472        if acoustid := tags.get("acoustid"):
2473            track.external_ids.add((ExternalID.ACOUSTID, acoustid))
2474
2475        # album
2476        album = track.album = (
2477            await self._parse_album(
2478                track_path=file_item.relative_path,
2479                track_tags=tags,
2480                track_created_at=file_item.created_at,
2481            )
2482            if tags.album
2483            else None
2484        )
2485
2486        # track artist(s)
2487        resolved_track_artists = await self._resolve_artists_with_mbids(
2488            tags.artists,
2489            tags.musicbrainz_artistids,
2490            tags.artist_sort_names,
2491            log_label="ARTISTS tag",
2492        )
2493        for name, mbid, sort_name in resolved_track_artists:
2494            # prefer the existing album artist object when it's the same artist
2495            if album_artist_match := self._match_album_artist(album, name, mbid):
2496                track.artists.append(album_artist_match)
2497                continue
2498            artist = await self._parse_artist(
2499                name, sort_name=sort_name, mbid=mbid, representative_track=file_item.relative_path
2500            )
2501            track.artists.append(artist)
2502
2503        # handle embedded cover image
2504        if tags.has_cover_image:
2505            # we do not actually embed the image in the metadata because that would consume too
2506            # much space and bandwidth. Instead we set the filename as value so the image can
2507            # be retrieved later in realtime.
2508            track.metadata.images = UniqueList(
2509                [
2510                    MediaItemImage(
2511                        type=ImageType.THUMB,
2512                        path=file_item.relative_path,
2513                        provider=self.instance_id,
2514                        remotely_accessible=False,
2515                    )
2516                ]
2517            )
2518
2519        # copy (embedded) album image from track (if the album itself doesn't have an image)
2520        if album and not album.image and track.image:
2521            album.metadata.images = UniqueList([track.image])
2522
2523        # parse other info
2524        track.duration = int(tags.duration or 0)
2525        track.metadata.genres = set(tags.genres)
2526        if tags.disc:
2527            track.disc_number = tags.disc
2528        if tags.track:
2529            track.track_number = tags.track
2530        track.metadata.copyright = tags.get("copyright")
2531        track.metadata.lyrics = tags.lyrics
2532        track.metadata.grouping = tags.get("grouping")
2533        track.metadata.description = tags.get("comment")
2534        explicit_tag = tags.get("itunesadvisory")
2535        if explicit_tag is not None:
2536            track.metadata.explicit = explicit_tag == "1"
2537        if recording_mbid := clean_mbid(tags.musicbrainz_recordingid, tags.filename):
2538            track.mbid = recording_mbid
2539
2540        # handle (optional) loudness measurement tag(s)
2541        if tags.track_loudness is not None:
2542            self.mass.create_task(
2543                self.mass.streams.audio_analysis.set_track_loudness(
2544                    track.item_id,
2545                    self.instance_id,
2546                    tags.track_loudness,
2547                    tags.track_album_loudness,
2548                )
2549            )
2550
2551        # possible lrclib metadata
2552        # synced lyrics are saved as "filename.lrc" by lrcget alongside
2553        # the actual file location - just change the file extension
2554        assert file_item.ext is not None  # for type checking
2555        lrc_path = f"{file_item.relative_path.removesuffix(file_item.ext)}lrc"
2556        if await self.exists(lrc_path):
2557            try:
2558                raw = await self._read_file(lrc_path)
2559                track.metadata.lrc_lyrics = raw.decode("utf-8")
2560            except Exception as err:
2561                self.logger.warning(
2562                    "Failed to read lyrics file %s: %s",
2563                    lrc_path,
2564                    str(err),
2565                )
2566        elif syn_lyrics := tags.synchronized_lyrics:
2567            track.metadata.lrc_lyrics = lyrics.convert_to_lrc_lyrics(syn_lyrics)
2568
2569        return track
2570
2571    async def _resolve_artists_with_mbids(
2572        self,
2573        parsed_names: tuple[str, ...],
2574        mbids: tuple[str, ...],
2575        sort_names: tuple[str, ...],
2576        log_label: str,
2577    ) -> list[tuple[str, str | None, str | None]]:
2578        """
2579        Return ``(name, mbid, sort_name)`` triples for a track's or album's artists.
2580
2581        When the parsed name count and the MBID count disagree, canonical names
2582        are looked up from MusicBrainz; otherwise the tag-parsed names are used.
2583
2584        :param parsed_names: Tag-parsed artist names.
2585        :param mbids: MusicBrainz artist IDs from the tag.
2586        :param sort_names: Sort names from the corresponding *sort tag.
2587        :param log_label: Tag name used in warning messages (e.g. "ARTISTS tag").
2588        """
2589
2590        def _sort_name(index: int) -> str | None:
2591            return sort_names[index] if index < len(sort_names) else None
2592
2593        def _from_tags() -> list[tuple[str, str | None, str | None]]:
2594            return [
2595                (
2596                    name,
2597                    mbids[i] if i < len(mbids) else None,
2598                    _sort_name(i),
2599                )
2600                for i, name in enumerate(parsed_names)
2601            ]
2602
2603        if not mbids or len(parsed_names) == len(mbids):
2604            return _from_tags()
2605
2606        mb_provider = cast("MusicbrainzProvider | None", self.mass.get_provider("musicbrainz"))
2607        if mb_provider is None:
2608            self.logger.warning(
2609                "%s count (%d) doesn't match MBID count (%d) and MusicBrainz "
2610                "provider is not loaded; using tag-parsed names: %s",
2611                log_label,
2612                len(parsed_names),
2613                len(mbids),
2614                parsed_names,
2615            )
2616            return _from_tags()
2617
2618        mb_results = await mb_provider.resolve_artists_from_mbids(mbids)
2619        # counts disagree, so positional fallback to a tag name is unreliable;
2620        # drop any MBID whose lookup failed (already logged per-MBID)
2621        resolved: list[tuple[str, str | None, str | None]] = [
2622            mb_result for mb_result in mb_results if mb_result is not None
2623        ]
2624        if not resolved:
2625            self.logger.warning(
2626                "%s count (%d) didn't match MBID count (%d) and every MusicBrainz "
2627                "lookup failed; falling back to tag-parsed names: %s",
2628                log_label,
2629                len(parsed_names),
2630                len(mbids),
2631                parsed_names,
2632            )
2633            return _from_tags()
2634        self.logger.info(
2635            "%s count (%d) didn't match MBID count (%d); resolved canonical names "
2636            "via MusicBrainz: %s",
2637            log_label,
2638            len(parsed_names),
2639            len(mbids),
2640            [r[0] for r in resolved],
2641        )
2642        return resolved
2643
2644    def _match_album_artist(
2645        self, album: Album | None, name: str, mbid: str | None
2646    ) -> Artist | ItemMapping | None:
2647        """
2648        Return an existing album artist representing the same artist, if any.
2649
2650        Matches on MusicBrainz ID when available (names may differ when only one
2651        side was resolved against MusicBrainz), otherwise on exact name.
2652
2653        :param album: The track's album, if known.
2654        :param name: Resolved track-artist name.
2655        :param mbid: Resolved track-artist MusicBrainz ID, if any.
2656        """
2657        if not album:
2658            return None
2659        return next(
2660            (x for x in album.artists if (mbid and x.mbid == mbid) or x.name == name),
2661            None,
2662        )
2663
2664    async def _parse_artist(
2665        self,
2666        name: str,
2667        album_dir: str | None = None,
2668        sort_name: str | None = None,
2669        mbid: str | None = None,
2670        artist_path: str | None = None,
2671        representative_track: str | None = None,
2672    ) -> Artist:
2673        """Parse full (album) Artist."""
2674        nfo_item: FileSystemItem | None = None
2675        nfo_root: dict[str, Any] | None = None
2676        # folders whose own artist.nfo was already read and rejected by the validated NFO
2677        # tier below; a later relaxed/fuzzy match landing on one of these must not blindly
2678        # trust that same rejected file during enrichment further down
2679        rejected_nfo_folders: set[str] = set()
2680        cleaned_mbid = clean_mbid(mbid, f"tags of artist {name}")
2681        if not artist_path:
2682            # exact tier: an already-known mapped identity or an exact (normalized) plain-name
2683            # match at a root/ancestor folder always wins - a validated artist.nfo (below)
2684            # still outranks a *relaxed* (fuzzy or sort-name-alias) match, but never this
2685            artist_path = await self._find_artist_path(name, album_dir, exact_only=True)
2686            if not artist_path and album_dir:
2687                # exact matching found nothing at all: fall back to a bounded validated
2688                # artist.nfo before trying any relaxed heuristic
2689                for candidate_name in (n for n in (name, sort_name) if n):
2690                    resolved = await self._resolve_artist_dir_via_nfo(
2691                        album_dir, candidate_name, cleaned_mbid, rejected_nfo_folders
2692                    )
2693                    if resolved:
2694                        artist_path, nfo_item, nfo_root = resolved
2695                        break
2696            if not artist_path:
2697                # relaxed tier: plain name is tried at every location before the sort-name
2698                # alias is tried anywhere - a sort-name match must never outrank a (relaxed)
2699                # plain-name match found elsewhere
2700                for candidate_name in (n for n in (name, sort_name) if n):
2701                    artist_path = await self._find_artist_path(candidate_name, album_dir)
2702                    if artist_path:
2703                        break
2704
2705        # prefer (short lived) cache for a bit more speed
2706        if artist_path and (
2707            cache := await self.cache.get(
2708                key=artist_path,
2709                provider=self.instance_id,
2710                category=CACHE_CATEGORY_ARTIST_INFO,
2711                base_class=Artist,
2712            )
2713        ):
2714            return cache  # type: ignore[no-any-return]
2715
2716        prov_artist_id = artist_path or name
2717        artist = Artist(
2718            item_id=prov_artist_id,
2719            provider=self.instance_id,
2720            name=name,
2721            sort_name=sort_name,
2722            provider_mappings={
2723                ProviderMapping(
2724                    item_id=prov_artist_id,
2725                    provider_domain=self.domain,
2726                    provider_instance=self.instance_id,
2727                    url=artist_path,
2728                    in_library=True,
2729                )
2730            },
2731        )
2732        if cleaned_mbid:
2733            artist.mbid = cleaned_mbid
2734        if not artist_path or not await self.exists(artist_path):
2735            return artist
2736
2737        # grab additional metadata within the Artist's folder
2738        if nfo_item is not None and nfo_root is not None:
2739            # already read and validated while resolving this artist's folder above
2740            parse_artist_nfo(artist, nfo_root, nfo_item.relative_path)
2741            await self._register_metadata_file(nfo_item, representative_track)
2742        elif artist_path not in rejected_nfo_folders and (
2743            read_nfo_item := await self._nfo_item_for(artist_path, "artist.nfo")
2744        ):
2745            # found NFO file with metadata; read and parse it. Skipped when this folder's
2746            # own artist.nfo was already read and rejected by the validated NFO tier above,
2747            # so a relaxed (fuzzy/sort-name-alias) match landing here can't silently trust
2748            # that same rejected file
2749            if read_root := await self._load_nfo_root(read_nfo_item, "artist"):
2750                parse_artist_nfo(artist, read_root, read_nfo_item.relative_path)
2751                # only a successful parse counts as having read this NFO: registering on a
2752                # malformed file would advance its token and treat the bad edit as handled,
2753                # permanently masking it (until unrelated changes trigger a full reparse)
2754                await self._register_metadata_file(read_nfo_item, representative_track)
2755            else:
2756                self.logger.warning(
2757                    "Failed to parse artist NFO file %s", read_nfo_item.relative_path
2758                )
2759        # find local images
2760        if images := await self._get_local_images(
2761            artist_path, extra_thumb_names=("artist",), representative_track=representative_track
2762        ):
2763            artist.metadata.images = UniqueList(images)
2764
2765        await self.cache.set(
2766            key=artist_path,
2767            data=artist.to_dict(),
2768            provider=self.instance_id,
2769            category=CACHE_CATEGORY_ARTIST_INFO,
2770            expiration=120,
2771        )
2772
2773        return artist
2774
2775    async def _parse_audiobook(self, file_item: FileSystemItem, tags: AudioTags) -> Audiobook:
2776        """
2777        Parse Audiobook details from file tags.
2778
2779        Audiobooks can be single files with embedded chapters or multiple files per folder.
2780        Only the first file (by track number or alphabetically) is processed as the audiobook.
2781        """
2782        # Skip files that aren't the first chapter.
2783        # A file carrying its own embedded chapter markers is a standalone audiobook,
2784        # so it should never be treated as a chapter file of another book.
2785        track_tag = tags.tags.get("track")
2786        if track_tag:
2787            track_num = try_parse_int(str(track_tag).split("/")[0], None)
2788            if track_num and track_num > 1 and not tags.chapters:
2789                raise IsChapterFile
2790        elif not tags.chapters:
2791            # No track tag and no embedded chapters -
2792            # assume part of a multi-file audiobook, only process the first file alphabetically
2793            items = await self._scandir(file_item.relative_parent_path)
2794            # Sort by filename for alphabetical ordering
2795            items.sort(key=lambda x: x.filename.lower())
2796            for item in items:
2797                if item.is_dir or item.ext not in AUDIOBOOK_EXTENSIONS:
2798                    continue
2799                if item.absolute_path != file_item.absolute_path:
2800                    raise IsChapterFile
2801                break
2802
2803        # For multi-file audiobooks, album tag is the book name, title is the chapter name
2804        if tags.album:
2805            book_name = tags.album
2806            sort_name = tags.album_sort
2807        elif (title := tags.tags.get("title")) and tags.track is None:
2808            book_name = title
2809            sort_name = tags.title_sort
2810        else:
2811            # file(s) without tags, use foldername
2812            book_name = file_item.parent_name
2813            sort_name = None
2814
2815        # collect all chapters
2816        total_duration, chapters = await self._get_chapters_for_audiobook(file_item, tags)
2817
2818        audio_book = Audiobook(
2819            item_id=file_item.relative_path,
2820            provider=self.instance_id,
2821            name=book_name,
2822            sort_name=sort_name,
2823            version=tags.version,
2824            duration=total_duration or int(tags.duration or 0),
2825            provider_mappings={
2826                ProviderMapping(
2827                    item_id=file_item.relative_path,
2828                    provider_domain=self.domain,
2829                    provider_instance=self.instance_id,
2830                    audio_format=AudioFormat(
2831                        content_type=ContentType.try_parse(file_item.ext or tags.format),
2832                        codec_type=self._codec_type_from_tags(tags),
2833                        sample_rate=tags.sample_rate,
2834                        bit_depth=tags.bits_per_sample,
2835                        channels=tags.channels,
2836                        bit_rate=tags.bit_rate,
2837                    ),
2838                    details=file_item.checksum,
2839                    in_library=True,
2840                )
2841            },
2842        )
2843        audio_book.metadata.chapters = chapters
2844
2845        # handle embedded cover image
2846        if tags.has_cover_image:
2847            # we do not actually embed the image in the metadata because that would consume too
2848            # much space and bandwidth. Instead we set the filename as value so the image can
2849            # be retrieved later in realtime.
2850            audio_book.metadata.add_image(
2851                MediaItemImage(
2852                    type=ImageType.THUMB,
2853                    path=self._versioned_image_path(file_item.relative_path, file_item.checksum),
2854                    provider=self.instance_id,
2855                    remotely_accessible=False,
2856                )
2857            )
2858
2859        # parse other info
2860        audio_book.authors.set(tags.writers or tags.album_artists or tags.artists)
2861        audio_book.metadata.genres = (
2862            set(tags.genres) if tags.genres else {DEFAULT_AUDIOBOOK_PODCAST_GENRE}
2863        )
2864        audio_book.metadata.copyright = tags.get("copyright")
2865        audio_book.metadata.lyrics = tags.lyrics
2866        audio_book.metadata.description = tags.get("comment")
2867        explicit_tag = tags.get("itunesadvisory")
2868        if explicit_tag is not None:
2869            audio_book.metadata.explicit = explicit_tag == "1"
2870        if recording_mbid := clean_mbid(tags.musicbrainz_recordingid, tags.filename):
2871            audio_book.mbid = recording_mbid
2872
2873        # try to fetch additional metadata from the folder
2874        if not audio_book.image or not audio_book.metadata.description:
2875            # try to get an image by traversing files in the same folder
2876            for _item in await self._scandir(file_item.relative_parent_path):
2877                if "." not in _item.relative_path or _item.is_dir:
2878                    continue
2879                if _item.ext in IMAGE_EXTENSIONS and not audio_book.image:
2880                    audio_book.metadata.add_image(
2881                        MediaItemImage(
2882                            type=ImageType.THUMB,
2883                            path=self._versioned_image_path(_item.relative_path, _item.checksum),
2884                            provider=self.instance_id,
2885                            remotely_accessible=False,
2886                        )
2887                    )
2888                if _item.ext == "txt" and not audio_book.metadata.description:
2889                    # try to parse a description from a text file
2890                    try:
2891                        raw = await self._read_file(_item.relative_path)
2892                        audio_book.metadata.description = raw.decode("utf-8")
2893                    except Exception as err:
2894                        self.logger.warning(
2895                            "Could not read description from file %s: %s",
2896                            _item.relative_path,
2897                            str(err),
2898                        )
2899
2900        # handle (optional) loudness measurement tag(s)
2901        if tags.track_loudness is not None:
2902            self.mass.create_task(
2903                self.mass.streams.audio_analysis.set_track_loudness(
2904                    audio_book.item_id,
2905                    self.instance_id,
2906                    tags.track_loudness,
2907                    tags.track_album_loudness,
2908                    media_type=MediaType.AUDIOBOOK,
2909                )
2910            )
2911        return audio_book
2912
2913    async def _parse_podcast_episode(
2914        self, file_item: FileSystemItem, tags: AudioTags
2915    ) -> PodcastEpisode:
2916        """Parse full PodcastEpisode details from file tags."""
2917        # ruff: noqa: PLR0915
2918        podcast_name = tags.album or file_item.parent_name
2919        podcast_path = file_item.relative_parent_path
2920        episode = PodcastEpisode(
2921            item_id=file_item.relative_path,
2922            provider=self.instance_id,
2923            name=tags.title,
2924            sort_name=tags.title_sort,
2925            provider_mappings={
2926                ProviderMapping(
2927                    item_id=file_item.relative_path,
2928                    provider_domain=self.domain,
2929                    provider_instance=self.instance_id,
2930                    audio_format=AudioFormat(
2931                        content_type=ContentType.try_parse(file_item.ext or tags.format),
2932                        codec_type=self._codec_type_from_tags(tags),
2933                        sample_rate=tags.sample_rate,
2934                        bit_depth=tags.bits_per_sample,
2935                        channels=tags.channels,
2936                        bit_rate=tags.bit_rate,
2937                    ),
2938                    details=file_item.checksum,
2939                    in_library=True,
2940                )
2941            },
2942            position=tags.track or 0,
2943            duration=try_parse_int(tags.duration) or 0,
2944            podcast=Podcast(
2945                item_id=podcast_path,
2946                provider=self.instance_id,
2947                name=podcast_name,
2948                sort_name=tags.album_sort,
2949                publisher=tags.tags.get("publisher"),
2950                provider_mappings={
2951                    ProviderMapping(
2952                        item_id=podcast_path,
2953                        provider_domain=self.domain,
2954                        provider_instance=self.instance_id,
2955                        in_library=True,
2956                    )
2957                },
2958            ),
2959        )
2960        # handle embedded cover image
2961        if tags.has_cover_image:
2962            # we do not actually embed the image in the metadata because that would consume too
2963            # much space and bandwidth. Instead we set the filename as value so the image can
2964            # be retrieved later in realtime.
2965            episode.metadata.add_image(
2966                MediaItemImage(
2967                    type=ImageType.THUMB,
2968                    path=file_item.relative_path,
2969                    provider=self.instance_id,
2970                    remotely_accessible=False,
2971                )
2972            )
2973        # parse other info
2974        episode.metadata.genres = (
2975            set(tags.genres) if tags.genres else {DEFAULT_AUDIOBOOK_PODCAST_GENRE}
2976        )
2977        episode.metadata.copyright = tags.get("copyright")
2978        episode.metadata.lyrics = tags.lyrics
2979        episode.metadata.description = tags.get("comment")
2980        explicit_tag = tags.get("itunesadvisory")
2981        if explicit_tag is not None:
2982            episode.metadata.explicit = explicit_tag == "1"
2983
2984        # handle (optional) chapters
2985        if tags.chapters:
2986            episode.metadata.chapters = [
2987                MediaItemChapter(
2988                    position=chapter.chapter_id,
2989                    name=chapter.title or f"Chapter {chapter.chapter_id}",
2990                    start=chapter.position_start,
2991                    end=chapter.position_end,
2992                )
2993                for chapter in tags.chapters
2994            ]
2995
2996        # try to fetch additional Podcast metadata from the folder
2997        assert isinstance(episode.podcast, Podcast)
2998        if images := await self._get_local_images(file_item.relative_parent_path):
2999            episode.podcast.metadata.images = images
3000        if metadata := await self._get_podcast_metadata(file_item.relative_parent_path):
3001            if title := metadata.get("title"):
3002                episode.podcast.name = title
3003            if sort_name := metadata.get("sorttitle"):
3004                episode.podcast.sort_name = sort_name
3005            if description := metadata.get("description"):
3006                episode.podcast.metadata.description = description
3007            if genres := metadata.get("genres"):
3008                episode.podcast.metadata.genres = set(genres)
3009            if publisher := metadata.get("publisher"):
3010                episode.podcast.publisher = publisher
3011            if image := metadata.get("imageURL"):
3012                episode.podcast.metadata.add_image(
3013                    MediaItemImage(
3014                        type=ImageType.THUMB,
3015                        path=image,
3016                        provider=self.instance_id,
3017                        remotely_accessible=True,
3018                    )
3019                )
3020        # copy (embedded) image from episode (or vice versa)
3021        if not episode.podcast.image and episode.image:
3022            episode.podcast.metadata.add_image(episode.image)
3023        elif not episode.image and episode.podcast.image:
3024            episode.metadata.add_image(episode.podcast.image)
3025        # ensure podcast has a default genre if none set
3026        if not episode.podcast.metadata.genres:
3027            episode.podcast.metadata.genres = {DEFAULT_AUDIOBOOK_PODCAST_GENRE}
3028
3029        # handle (optional) loudness measurement tag(s)
3030        if tags.track_loudness is not None:
3031            self.mass.create_task(
3032                self.mass.streams.audio_analysis.set_track_loudness(
3033                    episode.item_id,
3034                    self.instance_id,
3035                    tags.track_loudness,
3036                    tags.track_album_loudness,
3037                    media_type=MediaType.PODCAST_EPISODE,
3038                )
3039            )
3040        return episode
3041
3042    async def _parse_sound_effect(self, file_item: FileSystemItem, tags: AudioTags) -> SoundEffect:
3043        """Parse full sound effect details from file tags."""
3044        sound_effect = SoundEffect(
3045            item_id=file_item.relative_path,
3046            provider=self.instance_id,
3047            name=tags.title,
3048            sort_name=tags.title_sort,
3049            duration=int(tags.duration or 0),
3050            provider_mappings={
3051                ProviderMapping(
3052                    item_id=file_item.relative_path,
3053                    provider_domain=self.domain,
3054                    provider_instance=self.instance_id,
3055                    audio_format=AudioFormat(
3056                        content_type=ContentType.try_parse(file_item.ext or tags.format),
3057                        codec_type=self._codec_type_from_tags(tags),
3058                        sample_rate=tags.sample_rate,
3059                        bit_depth=tags.bits_per_sample,
3060                        channels=tags.channels,
3061                        bit_rate=tags.bit_rate,
3062                    ),
3063                    details=file_item.checksum,
3064                    in_library=True,
3065                )
3066            },
3067        )
3068        sound_effect.metadata.description = tags.get("comment")
3069        # handle embedded cover image
3070        if tags.has_cover_image:
3071            # we do not actually embed the image in the metadata because that would consume too
3072            # much space and bandwidth. Instead we set the filename as value so the image can
3073            # be retrieved later in realtime.
3074            sound_effect.metadata.add_image(
3075                MediaItemImage(
3076                    type=ImageType.THUMB,
3077                    path=file_item.relative_path,
3078                    provider=self.instance_id,
3079                    remotely_accessible=False,
3080                )
3081            )
3082        return sound_effect
3083
3084    async def _get_or_parse_sound_effect(self, file_item: FileSystemItem) -> SoundEffect:
3085        """Return the (cached) SoundEffect for the given file, parsing tags when needed."""
3086        cache_key = f"sound_effect.{file_item.relative_path}"
3087        cached_data: SoundEffect | None = await self.cache.get(
3088            cache_key,
3089            provider=self.instance_id,
3090            checksum=file_item.checksum,
3091            category=CACHE_CATEGORY_SOUND_EFFECTS,
3092            base_class=SoundEffect,
3093        )
3094        if cached_data is not None:
3095            return cached_data
3096        tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
3097        sound_effect = await self._parse_sound_effect(file_item, tags)
3098        await self.cache.set(
3099            cache_key,
3100            sound_effect.to_dict(),
3101            expiration=3600 * 24 * 365,  # File timestamp checksum handles invalidation
3102            provider=self.instance_id,
3103            checksum=file_item.checksum,
3104            category=CACHE_CATEGORY_SOUND_EFFECTS,
3105        )
3106        return sound_effect
3107
3108    async def _parse_album(
3109        self,
3110        track_path: str,
3111        track_tags: AudioTags,
3112        track_created_at: int | None = None,
3113        representative_track: str | None = None,
3114    ) -> Album:
3115        """
3116        Parse Album metadata from Track tags.
3117
3118        :param track_path: Path to the track file.
3119        :param track_tags: Audio tags from the track.
3120        :param track_created_at: Creation timestamp of the track file (Unix epoch).
3121        :param representative_track: The path to register for metadata-file change detection,
3122            when it differs from `track_path` itself (a CUE sheet's tracks are parsed from their
3123            companion audio file, but that companion is not a synced item on its own, so the CUE
3124            sheet's own path is the one that must be re-queued when the album/artist folder's
3125            NFO or images change). Defaults to `track_path`.
3126        """
3127        assert track_tags.album
3128        representative_track = representative_track or track_path
3129        # work out if we have an album and/or disc folder
3130        # track_dir is the folder level where the tracks are located
3131        # this may be a separate disc folder (Disc 1, Disc 2 etc) underneath the album folder
3132        # or this is an album folder with the disc attached
3133        track_dir = os.path.dirname(track_path)
3134        # exact tier: an exact (normalized) plain-name folder match always wins, so a
3135        # validated album.nfo elsewhere never displaces an already-obvious folder
3136        album_dir = get_album_dir(track_dir, track_tags.album, exact_only=True)
3137        nfo_item: FileSystemItem | None = None
3138        nfo_root: dict[str, Any] | None = None
3139        # folders whose own album.nfo was already read and rejected by the validated NFO
3140        # tier below; a later relaxed/fuzzy match landing on one of these must not blindly
3141        # trust that same rejected file during enrichment further down
3142        rejected_nfo_folders: set[str] = set()
3143        if not album_dir:
3144            # exact matching found nothing: fall back to a bounded validated album.nfo
3145            # (the immediate parent, and the track directory itself unless it is a disc
3146            # subfolder, whose own album.nfo is never trusted as identity) before trying
3147            # any relaxed (fuzzy/layout-variant/date-prefixed/sort-name-alias) heuristic
3148            resolved = await self._resolve_album_dir_via_nfo(
3149                track_dir, track_tags, rejected_nfo_folders
3150            )
3151            if resolved:
3152                album_dir, nfo_item, nfo_root = resolved
3153        if not album_dir:
3154            album_dir = get_album_dir(track_dir, track_tags.album, track_tags.album_sort)
3155
3156        if album_dir and (
3157            cache := await self.cache.get(
3158                key=album_dir,
3159                provider=self.instance_id,
3160                category=CACHE_CATEGORY_ALBUM_INFO,
3161                base_class=Album,
3162            )
3163        ):
3164            return cache  # type: ignore[no-any-return]
3165
3166        # album artist(s)
3167        # anchor the artist lookup on the track's own directory when the album itself never
3168        # resolved (e.g. no matching album.nfo), so the artist can still be found independently
3169        artist_lookup_dir = album_dir or track_dir
3170        album_artists: UniqueList[Artist | ItemMapping] = UniqueList()
3171        if track_tags.album_artists:
3172            resolved_album_artists = await self._resolve_artists_with_mbids(
3173                track_tags.album_artists,
3174                track_tags.musicbrainz_albumartistids,
3175                track_tags.album_artist_sort_names,
3176                log_label="ALBUMARTIST tag",
3177            )
3178            for name, mbid, sort_name in resolved_album_artists:
3179                artist = await self._parse_artist(
3180                    name,
3181                    album_dir=artist_lookup_dir,
3182                    sort_name=sort_name,
3183                    mbid=mbid,
3184                    representative_track=representative_track,
3185                )
3186                album_artists.append(artist)
3187        else:
3188            # album artist tag is missing, determine fallback
3189            fallback_action = self.config.get_value(CONF_ENTRY_MISSING_ALBUM_ARTIST.key)
3190            if fallback_action == "folder_name" and album_dir:
3191                possible_artist_folder = os.path.dirname(album_dir)
3192                self.logger.warning(
3193                    "%s is missing ID3 tag [albumartist], using foldername %s as fallback",
3194                    track_path,
3195                    possible_artist_folder,
3196                )
3197                album_artist_str = Path(possible_artist_folder).name
3198                album_artists = UniqueList(
3199                    [
3200                        await self._parse_artist(
3201                            name=album_artist_str,
3202                            album_dir=album_dir,
3203                            representative_track=representative_track,
3204                        )
3205                    ]
3206                )
3207            # fallback to track artists (if defined by user)
3208            elif fallback_action == "track_artist":
3209                self.logger.warning(
3210                    "%s is missing ID3 tag [albumartist], using track artist(s) as fallback",
3211                    track_path,
3212                )
3213                album_artists = UniqueList(
3214                    [
3215                        await self._parse_artist(
3216                            name=track_artist_str,
3217                            album_dir=artist_lookup_dir,
3218                            representative_track=representative_track,
3219                        )
3220                        for track_artist_str in track_tags.artists
3221                    ]
3222                )
3223            # all other: fallback to various artists
3224            else:
3225                self.logger.warning(
3226                    "%s is missing ID3 tag [albumartist], using %s as fallback",
3227                    track_path,
3228                    VARIOUS_ARTISTS_NAME,
3229                )
3230                album_artists = UniqueList(
3231                    [
3232                        await self._parse_artist(
3233                            name=VARIOUS_ARTISTS_NAME,
3234                            mbid=VARIOUS_ARTISTS_MBID,
3235                            representative_track=representative_track,
3236                        )
3237                    ]
3238                )
3239
3240        if album_dir:  # noqa: SIM108
3241            # prefer the path as id
3242            item_id = album_dir
3243        else:
3244            # create fake item_id based on artist + album
3245            item_id = album_artists[0].name + os.sep + track_tags.album
3246
3247        name, version = parse_title_and_version(track_tags.album)
3248        album = Album(
3249            item_id=item_id,
3250            provider=self.instance_id,
3251            name=name,
3252            version=version,
3253            sort_name=track_tags.album_sort,
3254            artists=album_artists,
3255            provider_mappings={
3256                ProviderMapping(
3257                    item_id=item_id,
3258                    provider_domain=self.domain,
3259                    provider_instance=self.instance_id,
3260                    url=album_dir,
3261                    in_library=True,
3262                )
3263            },
3264            date_added=(
3265                datetime.fromtimestamp(track_created_at, tz=UTC) if track_created_at else None
3266            ),
3267        )
3268        if track_tags.barcode:
3269            album.external_ids.add((ExternalID.BARCODE, track_tags.barcode))
3270
3271        if album_mbid := clean_mbid(track_tags.musicbrainz_albumid, track_tags.filename):
3272            album.mbid = album_mbid
3273        if releasegroup_mbid := clean_mbid(
3274            track_tags.musicbrainz_releasegroupid, track_tags.filename
3275        ):
3276            album.add_external_id(ExternalID.MB_RELEASEGROUP, releasegroup_mbid)
3277        if track_tags.year:
3278            album.year = track_tags.year
3279        album.album_type = track_tags.album_type
3280
3281        # hunt for additional metadata and images in the folder structure
3282        if not album_dir:
3283            return album
3284
3285        for folder_path in dict.fromkeys((track_dir, album_dir)):
3286            if not folder_path or not await self.exists(folder_path):
3287                continue
3288            if nfo_item is not None and nfo_root is not None:
3289                # identity was established through the bounded, validated NFO resolution
3290                # fallback above: only that one winning NFO ever applies. An unrelated
3291                # album.nfo the other candidate folder (track_dir or album_dir) happens to
3292                # also have was never validated against this track and must not silently
3293                # overwrite the resolved album's metadata.
3294                if folder_path == nfo_item.relative_parent_path:
3295                    parse_album_nfo(album, nfo_root, nfo_item.relative_path)
3296                    await self._register_metadata_file(nfo_item, representative_track)
3297            elif folder_path not in rejected_nfo_folders and (
3298                read_nfo_item := await self._nfo_item_for(folder_path, "album.nfo")
3299            ):
3300                # found NFO file with metadata; read and parse it. Skipped when this folder's
3301                # own album.nfo was already read and rejected by the validated NFO tier above,
3302                # so a relaxed (fuzzy/layout/date-prefix) match landing here can't silently
3303                # trust that same rejected file
3304                if read_root := await self._load_nfo_root(read_nfo_item, "album"):
3305                    parse_album_nfo(album, read_root, read_nfo_item.relative_path)
3306                    # only a successful parse counts as having read this NFO: registering on
3307                    # a malformed file would advance its token and treat the bad edit as
3308                    # handled, permanently masking it (until unrelated changes trigger a full
3309                    # reparse)
3310                    await self._register_metadata_file(read_nfo_item, representative_track)
3311                else:
3312                    self.logger.warning(
3313                        "Failed to parse album NFO file %s", read_nfo_item.relative_path
3314                    )
3315
3316            # find local images
3317            if images := await self._get_local_images(
3318                folder_path, extra_thumb_names=("album",), representative_track=representative_track
3319            ):
3320                if album.metadata.images is None:
3321                    album.metadata.images = UniqueList(images)
3322                else:
3323                    album.metadata.images += images
3324
3325        await self.cache.set(
3326            key=album_dir,
3327            data=album.to_dict(),
3328            provider=self.instance_id,
3329            category=CACHE_CATEGORY_ALBUM_INFO,
3330            expiration=120,
3331        )
3332        return album
3333
3334    async def _get_local_images(
3335        self,
3336        folder: str,
3337        extra_thumb_names: tuple[str, ...] | None = None,
3338        representative_track: str | None = None,
3339    ) -> UniqueList[MediaItemImage]:
3340        """
3341        Return local images found in a given folderpath.
3342
3343        :param folder: The folder to look for images in.
3344        :param extra_thumb_names: Extra image stems (besides folder/cover) treated as a thumbnail.
3345        :param representative_track: The track whose reparse rebuilds an album/artist that reads
3346            this folder, so each recognized image found here can register itself for change
3347            detection. Omit for folders unrelated to that (e.g. podcast folders).
3348        """
3349        if (
3350            cache := await self.cache.get(
3351                key=folder,
3352                provider=self.instance_id,
3353                category=CACHE_CATEGORY_FOLDER_IMAGES,
3354                base_class=MediaItemImage,
3355            )
3356        ) is not None:
3357            return UniqueList(cache)
3358        if extra_thumb_names is None:
3359            extra_thumb_names = ()
3360        images: UniqueList[MediaItemImage] = UniqueList()
3361        folder_files = await self._scandir(folder)
3362        for item in folder_files:
3363            if "." not in item.relative_path or item.is_dir or not item.ext:
3364                continue
3365            if item.ext.lower() not in IMAGE_EXTENSIONS:
3366                continue
3367            # try match on filename = one of our imagetypes
3368            if item.name.lower() in ImageType:
3369                images.append(
3370                    MediaItemImage(
3371                        type=ImageType(item.name),
3372                        path=item.relative_path,
3373                        provider=self.instance_id,
3374                        remotely_accessible=False,
3375                    )
3376                )
3377                await self._register_metadata_file(item, representative_track)
3378
3379        # try alternative names for thumbs
3380        extra_thumb_names = ("folder", "cover", *extra_thumb_names)
3381        for item in folder_files:
3382            if "." not in item.relative_path or item.is_dir or not item.ext:
3383                continue
3384            if item.ext.lower() not in IMAGE_EXTENSIONS:
3385                continue
3386            if item.name.lower() not in extra_thumb_names:
3387                continue
3388            images.append(
3389                MediaItemImage(
3390                    type=ImageType.THUMB,
3391                    path=item.relative_path,
3392                    provider=self.instance_id,
3393                    remotely_accessible=False,
3394                )
3395            )
3396            await self._register_metadata_file(item, representative_track)
3397
3398        await self.cache.set(
3399            key=folder,
3400            data=[img.to_dict() for img in images],
3401            provider=self.instance_id,
3402            category=CACHE_CATEGORY_FOLDER_IMAGES,
3403            expiration=120,
3404        )
3405        return images
3406
3407    async def _get_stream_details_for_track(self, item_id: str) -> StreamDetails:
3408        """Return the streamdetails for a track/song."""
3409        if parse_cue_track_id(item_id) is not None:
3410            return await self._cue.get_stream_details(item_id)
3411
3412        library_item = await self.mass.music.tracks.get_library_item_by_prov_id(
3413            item_id, self.instance_id
3414        )
3415        if library_item is None:
3416            # this could be a file that has just been added, try parsing it
3417            file_item = await self.resolve(item_id)
3418            tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
3419            if not (library_item := await self._parse_track(file_item, tags)):
3420                msg = f"Item not found: {item_id}"
3421                raise MediaNotFoundError(msg)
3422
3423        prov_mapping = next(x for x in library_item.provider_mappings if x.item_id == item_id)
3424        file_item = await self.resolve(item_id)
3425
3426        return StreamDetails(
3427            provider=self.instance_id,
3428            item_id=item_id,
3429            audio_format=prov_mapping.audio_format,
3430            media_type=MediaType.TRACK,
3431            stream_type=StreamType.LOCAL_FILE,
3432            duration=library_item.duration,
3433            size=file_item.file_size,
3434            data=file_item,
3435            path=file_item.absolute_path,
3436            can_seek=True,
3437            allow_seek=True,
3438        )
3439
3440    async def _get_stream_details_for_podcast_episode(self, item_id: str) -> StreamDetails:
3441        """Return the streamdetails for a podcast episode."""
3442        # podcasts episodes are never stored in the library so we need to parse the file
3443        file_item = await self.resolve(item_id)
3444        tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
3445        return StreamDetails(
3446            provider=self.instance_id,
3447            item_id=item_id,
3448            audio_format=AudioFormat(
3449                content_type=ContentType.try_parse(file_item.ext or tags.format),
3450                codec_type=self._codec_type_from_tags(tags),
3451                sample_rate=tags.sample_rate,
3452                bit_depth=tags.bits_per_sample,
3453                channels=tags.channels,
3454                bit_rate=tags.bit_rate,
3455            ),
3456            media_type=MediaType.PODCAST_EPISODE,
3457            stream_type=StreamType.LOCAL_FILE,
3458            duration=try_parse_int(tags.duration or 0),
3459            size=file_item.file_size,
3460            data=file_item,
3461            path=file_item.absolute_path,
3462            allow_seek=True,
3463            can_seek=True,
3464        )
3465
3466    async def _get_stream_details_for_sound_effect(self, item_id: str) -> StreamDetails:
3467        """Return the streamdetails for a sound effect."""
3468        # sound effects are never stored in the library so we parse the file,
3469        # served from cache unless the file changed on disk
3470        file_item = await self.resolve(item_id)
3471        sound_effect = await self._get_or_parse_sound_effect(file_item)
3472        prov_mapping = next(x for x in sound_effect.provider_mappings if x.item_id == item_id)
3473        return StreamDetails(
3474            provider=self.instance_id,
3475            item_id=item_id,
3476            audio_format=prov_mapping.audio_format,
3477            media_type=MediaType.SOUND_EFFECT,
3478            stream_type=StreamType.LOCAL_FILE,
3479            duration=sound_effect.duration,
3480            size=file_item.file_size,
3481            data=file_item,
3482            path=file_item.absolute_path,
3483            allow_seek=True,
3484            can_seek=True,
3485        )
3486
3487    async def _get_stream_details_for_audiobook(self, item_id: str) -> StreamDetails:
3488        """Return the streamdetails for an audiobook."""
3489        library_item = await self.mass.music.audiobooks.get_library_item_by_prov_id(
3490            item_id, self.instance_id
3491        )
3492        if library_item is None:
3493            # this could be a file that has just been added, try parsing it
3494            file_item = await self.resolve(item_id)
3495            tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
3496            if not (library_item := await self._parse_audiobook(file_item, tags)):
3497                msg = f"Item not found: {item_id}"
3498                raise MediaNotFoundError(msg)
3499
3500        prov_mapping = next(x for x in library_item.provider_mappings if x.item_id == item_id)
3501        file_item = await self.resolve(item_id)
3502        duration = library_item.duration
3503        file_based_chapters: list[tuple[str, float]] | None = await self.cache.get(
3504            key=file_item.relative_path,
3505            provider=self.instance_id,
3506            category=CACHE_CATEGORY_AUDIOBOOK_CHAPTERS,
3507        )
3508        if file_based_chapters is None:
3509            # no cache available for this audiobook, we need to parse the chapters
3510            tags = await async_parse_tags(file_item.absolute_path, file_item.file_size)
3511            await self._parse_audiobook(file_item, tags)
3512            file_based_chapters = await self.cache.get(
3513                key=file_item.relative_path,
3514                provider=self.instance_id,
3515                category=CACHE_CATEGORY_AUDIOBOOK_CHAPTERS,
3516            )
3517
3518        if file_based_chapters:
3519            # this is a multi-file audiobook
3520            return StreamDetails(
3521                provider=self.instance_id,
3522                item_id=item_id,
3523                audio_format=prov_mapping.audio_format,
3524                media_type=MediaType.AUDIOBOOK,
3525                stream_type=StreamType.LOCAL_FILE,
3526                duration=duration,
3527                path=[
3528                    MultiPartPath(
3529                        path=self._get_chapter_path(chapter_path),
3530                        duration=chapter_duration,
3531                    )
3532                    for chapter_path, chapter_duration in file_based_chapters
3533                ],
3534                allow_seek=True,
3535            )
3536
3537        # regular single-file streaming, simply let ffmpeg deal with the file directly
3538        return StreamDetails(
3539            provider=self.instance_id,
3540            item_id=item_id,
3541            audio_format=prov_mapping.audio_format,
3542            media_type=MediaType.AUDIOBOOK,
3543            stream_type=StreamType.LOCAL_FILE,
3544            duration=library_item.duration,
3545            size=file_item.file_size,
3546            data=file_item,
3547            path=file_item.absolute_path,
3548            allow_seek=True,
3549            can_seek=True,
3550        )
3551
3552    def _get_chapter_path(self, relative_path: str) -> str:
3553        """Return absolute path for a chapter file. Override for network storage."""
3554        return self.get_absolute_path(relative_path)
3555
3556    async def _get_chapters_for_audiobook(
3557        self, audiobook_file_item: FileSystemItem, tags: AudioTags
3558    ) -> tuple[int, list[MediaItemChapter]]:
3559        """
3560        Return chapters for an audiobook.
3561
3562        Chapter sources in order of preference:
3563        1. Multiple files with track tags - sorted by track number
3564        2. Single file with embedded chapters - use embedded chapter markers
3565        3. Multiple files without track tags - sorted alphabetically (fallback)
3566        """
3567        chapters: list[MediaItemChapter] = []
3568        all_chapter_files: list[tuple[str, float]] = []
3569        total_duration = 0.0
3570
3571        # Scan folder for chapter files, separating tagged from untagged
3572        chapter_file_items: list[tuple[FileSystemItem, AudioTags]] = []
3573        untagged_file_items: list[tuple[FileSystemItem, AudioTags]] = []
3574
3575        items = await self._scandir(audiobook_file_item.relative_parent_path)
3576        # Sort by filename for consistent alphabetical ordering
3577        items.sort(key=lambda x: x.filename.lower())
3578
3579        for item in items:
3580            if "." not in item.relative_path or item.is_dir:
3581                continue
3582            if item.ext not in AUDIOBOOK_EXTENSIONS:
3583                continue
3584            item_tags = await async_parse_tags(item.absolute_path, item.file_size)
3585            if not (tags.album == item_tags.album or (item_tags.tags.get("title") is None)):
3586                continue
3587            if item_tags.tags.get("track") is None:
3588                untagged_file_items.append((item, item_tags))
3589            else:
3590                chapter_file_items.append((item, item_tags))
3591
3592        # Determine chapter source
3593        use_embedded = False
3594        use_alphabetical = False
3595
3596        if len(chapter_file_items) > 1:
3597            chapter_file_items.sort(key=lambda x: (x[1].disc or 0, x[1].track or 0))
3598        elif len(chapter_file_items) <= 1 and tags.chapters:
3599            use_embedded = True
3600        elif len(untagged_file_items) > 1:
3601            use_alphabetical = True
3602            chapter_file_items = untagged_file_items
3603            self.logger.info(
3604                "Audiobook files have no track tags, using alphabetical order: %s",
3605                tags.album,
3606            )
3607
3608        if use_embedded:
3609            chapters = [
3610                MediaItemChapter(
3611                    position=chapter.chapter_id,
3612                    name=chapter.title or f"Chapter {chapter.chapter_id}",
3613                    start=chapter.position_start,
3614                    end=chapter.position_end,
3615                )
3616                for chapter in tags.chapters
3617            ]
3618            total_duration = try_parse_int(tags.duration) or 0
3619            self.logger.log(
3620                VERBOSE_LOG_LEVEL,
3621                "Audiobook '%s': %d embedded chapters, duration=%d",
3622                tags.album,
3623                len(chapters),
3624                int(total_duration),
3625            )
3626        else:
3627            for position, (chapter_item, chapter_tags) in enumerate(chapter_file_items, start=1):
3628                if chapter_tags.duration is None:
3629                    self.logger.warning(
3630                        "Chapter file has no duration, skipping: %s",
3631                        chapter_item.relative_path,
3632                    )
3633                    continue
3634                self.logger.debug("Chapter filename: %s", chapter_item.relative_path)
3635                chapters.append(
3636                    MediaItemChapter(
3637                        position=position,
3638                        name=chapter_tags.title,
3639                        start=total_duration,
3640                        end=total_duration + chapter_tags.duration,
3641                    )
3642                )
3643                all_chapter_files.append(
3644                    (
3645                        chapter_item.relative_path,
3646                        chapter_tags.duration,
3647                    )
3648                )
3649                total_duration += chapter_tags.duration
3650            sort_method = "alphabetical" if use_alphabetical else "track"
3651            self.logger.log(
3652                VERBOSE_LOG_LEVEL,
3653                "Audiobook '%s': %d files (%s order), duration=%d",
3654                tags.album,
3655                len(chapters),
3656                sort_method,
3657                int(total_duration),
3658            )
3659        # Cache chapter files for streaming
3660        await self.cache.set(
3661            key=audiobook_file_item.relative_path,
3662            data=all_chapter_files,
3663            provider=self.instance_id,
3664            category=CACHE_CATEGORY_AUDIOBOOK_CHAPTERS,
3665        )
3666        return int(total_duration), chapters
3667
3668    async def _get_podcast_metadata(self, podcast_folder: str) -> dict[str, Any]:
3669        """Return metadata for a podcast."""
3670        if (
3671            cache := await self.cache.get(
3672                key=podcast_folder,
3673                provider=self.instance_id,
3674                category=CACHE_CATEGORY_PODCAST_METADATA,
3675            )
3676        ) is not None:
3677            return cast("dict[str, Any]", cache)
3678        data: dict[str, Any] = {}
3679        metadata_file = os.path.join(podcast_folder, "metadata.json")
3680        if await self.exists(metadata_file):
3681            # found json file with metadata
3682            raw = await self._read_file(metadata_file)
3683            data.update(json_loads(raw.decode("utf-8")))
3684        await self.cache.set(
3685            key=podcast_folder,
3686            data=data,
3687            provider=self.instance_id,
3688            category=CACHE_CATEGORY_PODCAST_METADATA,
3689        )
3690        return data
3691
3692    async def _scandir(self, path: str, use_cache: bool = True) -> list[FileSystemItem]:
3693        """
3694        List directory contents in natural sort order.
3695
3696        :param use_cache: Unused for local disk (a read is always fresh); accepted so shared,
3697            cache-sensitive callers (e.g. on-demand NFO/image lookups honoring a manual "Refresh
3698            item") can request a bypass the same way regardless of which subclass overrides
3699            this, since a cloud-backed provider serves its own short-lived listing cache here.
3700        """
3701        # raw scandir order depends on the underlying filesystem (e.g. hash order
3702        # on ext4) so sort to make browse and folder playback order deterministic
3703        abs_path = self.get_absolute_path(path)
3704        return await asyncio.to_thread(sorted_scandir, self.base_path, abs_path, sort=True)
3705
3706    async def _read_file(self, path: str) -> bytes:
3707        """Read file contents. Override for network storage."""
3708        async with aiofiles.open(self.get_absolute_path(path), mode="rb") as f:
3709            return cast("bytes", await f.read())
3710