music-assistant-server

1"""
2Base class for cloud-storage filesystem providers (Google Drive, OneDrive, ...).
3
4Extends LocalFileSystemProvider with path <-> cloud-file-ID resolution, API-backed
5directory listings and streaming through a dynamic MA URL, so short-lived cloud
6auth tokens stay fresh. Concrete providers implement the _api_* hooks and their
7own auth/setup.
8"""
9
10from __future__ import annotations
11
12import posixpath
13import time
14from dataclasses import replace
15from typing import TYPE_CHECKING, cast
16from urllib.parse import quote
17
18from aiohttp import ClientError, web
19from music_assistant_models.config_entries import ConfigEntry
20from music_assistant_models.enums import ConfigEntryType
21from music_assistant_models.errors import (
22    LoginFailed,
23    MediaNotFoundError,
24    ProviderUnavailableError,
25)
26
27from music_assistant.controllers.tasks.context import update_current_task_progress_text
28from music_assistant.helpers.tags import get_embedded_image
29from music_assistant.models.setup_flow import SetupFlowError
30from music_assistant.providers.filesystem_local import LocalFileSystemProvider
31from music_assistant.providers.filesystem_local.constants import (
32    AUDIOBOOK_EXTENSIONS,
33    CONF_CONTENT_TYPE,
34    CONF_ENTRY_CONTENT_TYPE,
35    CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
36    PODCAST_EPISODE_EXTENSIONS,
37    SUPPORTED_EXTENSIONS,
38    TRACK_EXTENSIONS,
39)
40from music_assistant.providers.filesystem_local.helpers import FileSystemItem, ScanErrors
41
42if TYPE_CHECKING:
43    from collections.abc import Awaitable, Callable
44
45    from aiohttp import ClientResponse
46    from music_assistant_models.config_entries import ConfigValueType, ProviderConfig
47    from music_assistant_models.provider import ProviderManifest
48
49    from music_assistant.mass import MusicAssistant
50    from music_assistant.models.setup_flow import SetupSession
51
52# (id, name, is_dir, checksum, size) as returned by _api_list_children
53RawItem = tuple[str, str, bool, str, int | None]
54
55# extensions the stream route will serve; playlists/cue/images are read
56# server-side and never fetched over HTTP, so audio is all it needs to proxy
57AUDIO_STREAM_EXTENSIONS = TRACK_EXTENSIONS | AUDIOBOOK_EXTENSIONS | PODCAST_EPISODE_EXTENSIONS
58
59# config keys shared by the cloud filesystem providers, all collected by the setup flow
60CONF_CLIENT_ID = "client_id"
61CONF_CLIENT_SECRET = "client_secret"
62CONF_REFRESH_TOKEN = "refresh_token"
63CONF_FOLDER_ID = "folder_id"
64
65
66async def run_cloud_setup(
67    session: SetupSession,
68    authorize: Callable[[SetupSession, str, str], Awaitable[str]],
69) -> None:
70    """
71    Drive the setup flow shared by the cloud filesystem providers.
72
73    Collects the content type, OAuth client credentials and root folder, runs the
74    provider-specific OAuth ``authorize`` step for a refresh token and persists it all.
75
76    :param session: The setup session driving the flow.
77    :param authorize: Provider-specific coroutine that runs the OAuth consent for the given
78        (client_id, client_secret) and returns the resulting refresh token.
79    """
80    setup_data = dict(session.context.setup_data)
81    # a secure value is never echoed back into a flow step, so on reconfigure the user may
82    # leave the client secret blank to reuse the previously stored one
83    stored_secret = str(session.context.setup_data.get(CONF_CLIENT_SECRET) or "")
84    errors: dict[str, str] | None = None
85    while True:
86        entries = [
87            replace(entry, value=setup_data.get(entry.key, entry.value))
88            for entry in _cloud_setup_entries(has_stored_secret=bool(stored_secret))
89        ]
90        submitted = await session.form(entries, step_id="user", errors=errors)
91        setup_data.update(submitted)
92        client_id = str(setup_data.get(CONF_CLIENT_ID) or "")
93        client_secret = str(setup_data.get(CONF_CLIENT_SECRET) or "") or stored_secret
94        setup_data[CONF_CLIENT_SECRET] = client_secret
95        # a blank secret on a retry means "the one just tried", not the original stored one
96        stored_secret = client_secret
97        try:
98            if not client_secret:
99                raise SetupFlowError("A client secret is required", translation_key="required")
100            setup_data[CONF_REFRESH_TOKEN] = await authorize(session, client_id, client_secret)
101            await session.finish(setup_data)
102            return
103        except SetupFlowError as err:
104            errors = {"base": err.translation_key or str(err)}
105
106
107def read_setup_value(
108    mass: MusicAssistant, config: ProviderConfig, key: str, default: ConfigValueType = None
109) -> ConfigValueType:
110    """
111    Read a setup_data value from a config not yet attached to a provider instance.
112
113    Mirrors Provider.get_setup_value for the __init__ window (before super().__init__),
114    decrypting strings and reading through to legacy config values for pre-flow installs.
115
116    :param mass: The MusicAssistant instance.
117    :param config: The provider config being loaded.
118    :param key: The setup data key to read.
119    :param default: Value to return when the key is not present anywhere.
120    """
121    value = config.setup_data.get(key)
122    if value is not None:
123        return mass.config.decrypt_string(value) if isinstance(value, str) else value
124    return config.get_value(key, default)
125
126
127def _cloud_setup_entries(*, has_stored_secret: bool) -> tuple[ConfigEntry, ...]:
128    """Return the config entries collected by the shared cloud setup form."""
129    return (
130        CONF_ENTRY_CONTENT_TYPE,
131        ConfigEntry(key=CONF_CLIENT_ID, type=ConfigEntryType.STRING, required=True),
132        ConfigEntry(
133            key=CONF_CLIENT_SECRET,
134            type=ConfigEntryType.SECURE_STRING,
135            # optional on reconfigure (a stored secret can be reused), required on first setup
136            required=not has_stored_secret,
137        ),
138        ConfigEntry(
139            key=CONF_FOLDER_ID, type=ConfigEntryType.STRING, required=False, default_value="root"
140        ),
141    )
142
143
144class CloudFileSystemProvider(LocalFileSystemProvider):
145    """Base class for filesystem providers backed by a cloud storage API."""
146
147    # cloud APIs generally struggle with the default 16 parallel tag-parse downloads
148    _SYNC_CONCURRENCY = 4
149    # how long a folder listing may be served from cache; keeps interactive
150    # browsing snappy (no API round trip per click). Library syncs always fetch
151    # fresh listings, so new cloud content is never missed because of this.
152    _DIR_CACHE_TTL = 300
153
154    def __init__(
155        self,
156        mass: MusicAssistant,
157        manifest: ProviderManifest,
158        config: ProviderConfig,
159        root_folder_id: str,
160    ) -> None:
161        """
162        Initialize the cloud filesystem provider.
163
164        :param root_folder_id: The cloud provider's opaque ID of the root folder to serve.
165        """
166        # base_path is unused for us, but the parent expects something
167        super().__init__(mass, manifest, config, root_folder_id)
168        # the content type is collected by the setup flow (setup_data); the parent reads it
169        # from the legacy config values, so re-resolve it setup-data-aware (read-through keeps
170        # pre-flow installs working)
171        self.media_content_type = cast(
172            "str", self.get_setup_value(CONF_CONTENT_TYPE, CONF_ENTRY_CONTENT_TYPE.default_value)
173        )
174        self.root_folder_id = root_folder_id
175        self._unregister_stream_route: Callable[[], None] | None = None
176        # per-folder listing cache: folder path -> {child name -> (cloud id, item)};
177        # every path->id lookup is answered from here, so sibling probes by the
178        # inherited logic (artwork, lyrics, playlists) cost no extra API calls
179        self._dir_cache: dict[str, dict[str, tuple[str, FileSystemItem]]] = {}
180        # monotonic deadline per folder path until which _scandir may serve the
181        # cached listing; path->id lookups deliberately never expire (IDs are stable)
182        self._dir_cache_expiry: dict[str, float] = {}
183
184    async def unload(self, is_removed: bool = False) -> None:
185        """Handle unload/close of the provider."""
186        if self._unregister_stream_route is not None:
187            self._unregister_stream_route()
188
189    async def resolve(self, file_path: str) -> FileSystemItem:
190        """Resolve a relative path to a FileSystemItem."""
191        file_path = self._normalize_path(file_path)
192        if entry := await self._lookup(file_path):
193            return entry[1]
194        raise MediaNotFoundError(f"Cloud path not found: {file_path}")
195
196    async def exists(self, file_path: str) -> bool:
197        """Check if a cloud file/folder exists."""
198        if not file_path:
199            return False
200        try:
201            return await self._lookup(self._normalize_path(file_path)) is not None
202        except ProviderUnavailableError, MediaNotFoundError:
203            return False
204
205    async def resolve_image(self, path: str) -> str | bytes:
206        """Return raw image bytes for a cloud image file or embedded cover art."""
207        # drop the cache-busting suffix the parent appends for embedded images
208        path = path.split("?cs=", 1)[0]
209        ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
210        if ext in SUPPORTED_EXTENSIONS:
211            # audio file: extract the embedded art with ffmpeg over our stream URL
212            if img_data := await get_embedded_image(self._stream_url(path)):
213                return img_data
214            raise MediaNotFoundError(f"No embedded image found: {path}")
215        return await self._read_file(path)
216
217    # ------------------------------------------------------------------
218    # API hooks (implemented by the concrete cloud provider); hooks must
219    # translate client library errors into MA errors: ProviderUnavailableError
220    # for API/transport failures, LoginFailed for authentication problems
221    # ------------------------------------------------------------------
222
223    async def _api_list_children(self, folder_id: str) -> list[RawItem]:
224        """
225        List the children of a cloud folder, following pagination if needed.
226
227        :param folder_id: The cloud provider's opaque folder ID.
228        :return: One (id, name, is_dir, checksum, size) tuple per child.
229        """
230        raise NotImplementedError
231
232    async def _api_download_bytes(self, file_id: str) -> bytes:
233        """
234        Download a (small) cloud file's full contents.
235
236        :param file_id: The cloud provider's opaque file ID.
237        """
238        raise NotImplementedError
239
240    async def _api_download_response(self, file_id: str, headers: dict[str, str]) -> ClientResponse:
241        """
242        Open a streaming download for a cloud file.
243
244        :param file_id: The cloud provider's opaque file ID.
245        :param headers: Extra request headers to forward (e.g. Range for seeking).
246        """
247        raise NotImplementedError
248
249    # ------------------------------------------------------------------
250    # initialization helpers
251    # ------------------------------------------------------------------
252
253    async def _post_init(self) -> None:
254        """Complete common initialization; call at the end of handle_async_init."""
255        self._register_stream_route()
256
257    def _register_stream_route(self) -> None:
258        """Register the dynamic route that proxies cloud downloads with fresh auth."""
259        self._unregister_stream_route = self.mass.streams.register_dynamic_route(
260            f"/{self.instance_id}_stream", self._handle_stream_request
261        )
262
263    # ------------------------------------------------------------------
264    # filesystem hooks (these are what the parent calls)
265    # ------------------------------------------------------------------
266
267    async def _scandir(self, path: str, use_cache: bool = True) -> list[FileSystemItem]:
268        """
269        List the children of a cloud folder.
270
271        `path` is the relative path of the folder ("" means this provider's root).
272        `use_cache` allows serving a recent cached listing; pass False to force
273        a fresh fetch from the cloud API.
274        """
275        path = self._normalize_path(path)
276        # serve recently fetched listings from cache so browsing back and forth
277        # through folders doesn't cost an API round trip per click
278        if (
279            use_cache
280            and (cached := self._dir_cache.get(path)) is not None
281            and time.monotonic() < self._dir_cache_expiry.get(path, 0)
282        ):
283            return [entry[1] for entry in cached.values()]
284        folder_id = await self._resolve_id(path)
285        children: dict[str, tuple[str, FileSystemItem]] = {}
286        items: list[FileSystemItem] = []
287        for raw in await self._api_list_children(folder_id):
288            # slashes in cloud file names would corrupt our path scheme
289            name = raw[1].replace("/", "_")
290            if name in children:
291                # some clouds (e.g. Google Drive) allow duplicate names in a folder; paths can't
292                self.logger.warning(
293                    "Duplicate name '%s' in folder '%s' - ignoring all but the first",
294                    name,
295                    path or "(root)",
296                )
297                continue
298            item = self._to_item(raw, path, name)
299            children[name] = (raw[0], item)
300            items.append(item)
301        self._dir_cache[path] = children
302        self._dir_cache_expiry[path] = time.monotonic() + self._DIR_CACHE_TTL
303        return items
304
305    async def _enumerate_files_for_sync(
306        self,
307        *,
308        file_checksums: dict[str, str],
309        cue_file_checksums: dict[str, str],
310        cur_filenames: set[str],
311        items_to_process: list[tuple[FileSystemItem, str | None]],
312        unchanged_cue_items: list[FileSystemItem],
313        cue_stems: set[str],
314        scan_errors: ScanErrors,
315    ) -> None:
316        """Walk the cloud folder tree via the API and populate the sync buckets."""
317        ignore_album_playlists = self.media_content_type == "music" and bool(
318            self.config.get_value(CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS.key)
319        )
320        # mutable counter for the nested coroutine
321        scanned = [0]
322        # a cloud folder may be reachable twice (e.g. Drive multi-parent), so
323        # guard against re-visiting
324        visited: set[str] = set()
325
326        async def _walk(path: str, is_root: bool) -> None:
327            if path in visited:
328                return
329            visited.add(path)
330            try:
331                # always fetch fresh during a sync so new cloud content is
332                # picked up no matter how recently a folder was browsed
333                items = await self._scandir(path, use_cache=False)
334            except ProviderUnavailableError as err:
335                # a root-level failure aborts the sync right away, subfolder failures only
336                # once too many happen in a row, matching the local-filesystem walker
337                if not is_root:
338                    self.logger.warning("Error scanning folder %s: %s", path, err)
339                scan_errors.record_dir_error(err, is_root=is_root, path=path)
340                return
341            scan_errors.record_dir_read()
342            for item in items:
343                if item.is_dir:
344                    await _walk(item.relative_path, is_root=False)
345                    if scan_errors.aborted:
346                        return
347                    continue
348                if item.ext not in SUPPORTED_EXTENSIONS:
349                    continue
350                scanned[0] += 1
351                if scanned[0] % 500 == 0:
352                    update_current_task_progress_text(f"Scanning files: {scanned[0]} found")
353                self._classify_scan_item(
354                    item,
355                    file_checksums=file_checksums,
356                    cue_file_checksums=cue_file_checksums,
357                    cur_filenames=cur_filenames,
358                    items_to_process=items_to_process,
359                    unchanged_cue_items=unchanged_cue_items,
360                    cue_stems=cue_stems,
361                    ignore_album_playlists=ignore_album_playlists,
362                )
363
364        await _walk("", is_root=True)
365
366    async def _read_file(self, path: str) -> bytes:
367        """Download a (small text) file's bytes: nfo, m3u, lrc, etc."""
368        file_id = await self._resolve_id(self._normalize_path(path))
369        try:
370            return await self._api_download_bytes(file_id)
371        except ProviderUnavailableError as err:
372            raise MediaNotFoundError(f"Unable to read cloud file {path}: {err}") from err
373
374    def _get_chapter_path(self, relative_path: str) -> str:
375        """Return the streamable URL for an audiobook chapter file."""
376        return self._stream_url(relative_path)
377
378    # ------------------------------------------------------------------
379    # streaming
380    # ------------------------------------------------------------------
381
382    def _stream_url(self, path: str) -> str:
383        """Build the MA-hosted URL that proxies this cloud file."""
384        base = f"{self.mass.streams.base_url}/{self.instance_id}_stream"
385        return f"{base}?path={quote(path)}"
386
387    async def _handle_stream_request(self, request: web.Request) -> web.StreamResponse:
388        """
389        Proxy a cloud download through MA, adding a fresh auth header.
390
391        Because this runs per request, the token is always valid - so even a
392        multi-hour audiobook can't outlive it.
393        """
394        path = self._normalize_path(request.query.get("path") or "")
395        if not path:
396            raise web.HTTPBadRequest(text="Missing path")
397        # the streamserver is unauthenticated: only proxy audio files so this route
398        # can't be used to download arbitrary files from the cloud account
399        # (same 404 as a missing file, so blocked paths are indistinguishable)
400        ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
401        if ext not in AUDIO_STREAM_EXTENSIONS:
402            raise web.HTTPNotFound(text="File not found")
403        try:
404            file_id = await self._resolve_id(path)
405        except MediaNotFoundError as err:
406            self.logger.debug("Cloud stream path not found: %s (%s)", path, err)
407            raise web.HTTPNotFound(text="File not found") from err
408        # forward Range header so players can seek
409        headers = {}
410        if rng := request.headers.get("Range"):
411            headers["Range"] = rng
412        try:
413            cloud_resp = await self._api_download_response(file_id, headers)
414        except (ProviderUnavailableError, LoginFailed) as err:
415            self.logger.warning("Cloud provider unavailable while streaming %s: %s", path, err)
416            raise web.HTTPBadGateway(text="Upstream provider unavailable") from err
417
418        response = web.StreamResponse(status=cloud_resp.status)
419        # copy content-type / length / range headers back to the player
420        for h in ("Content-Type", "Content-Length", "Content-Range", "Accept-Ranges"):
421            if h in cloud_resp.headers:
422                response.headers[h] = cloud_resp.headers[h]
423        try:
424            await response.prepare(request)
425            async for chunk in cloud_resp.content.iter_chunked(64 * 1024):
426                await response.write(chunk)
427            await response.write_eof()
428        except ConnectionError:
429            # client hung up early (e.g. ffmpeg closes as soon as it has read
430            # the tags); perfectly normal, not an error
431            self.logger.debug("Client disconnected while streaming %s", path)
432        except ClientError as err:
433            # the cloud side dropped mid-transfer
434            self.logger.warning("Cloud download interrupted for %s: %s", path, err)
435        finally:
436            # abort the cloud download so we don't keep pulling unneeded bytes
437            cloud_resp.close()
438        return response
439
440    # ------------------------------------------------------------------
441    # path resolution helpers
442    # ------------------------------------------------------------------
443
444    def _normalize_path(self, path: str) -> str:
445        """Normalize a relative path (collapse ./.. segments from playlist entries)."""
446        path = path.strip("/")
447        if path:
448            path = posixpath.normpath(path)
449            if path == ".":
450                path = ""
451        return path
452
453    async def _lookup(self, path: str) -> tuple[str, FileSystemItem] | None:
454        """
455        Return the cached (cloud id, item) tuple for a relative path, if it exists.
456
457        Lists the parent folder (once) on a cache miss.
458        """
459        if not path:
460            return None
461        parent, _, name = path.rpartition("/")
462        if (children := self._dir_cache.get(parent)) is None:
463            await self._scandir(parent)
464            children = self._dir_cache.get(parent, {})
465        return children.get(name)
466
467    async def _resolve_id(self, path: str) -> str:
468        """Resolve a relative path to its cloud file ID."""
469        if not path:
470            return self.root_folder_id
471        if entry := await self._lookup(path):
472            return entry[0]
473        raise MediaNotFoundError(f"Cloud path not found: {path}")
474
475    def _to_item(self, raw: RawItem, parent_path: str, name: str) -> FileSystemItem:
476        """Convert a raw API listing entry to a FileSystemItem."""
477        _, _, is_dir, checksum, size = raw
478        relative_path = f"{parent_path}/{name}" if parent_path else name
479        return FileSystemItem(
480            filename=name,
481            relative_path=relative_path,
482            # absolute_path is what the parent hands to the tag parser (ffmpeg);
483            # point it at our streaming URL so tags are read over HTTP with a
484            # fresh token - no temp download needed. Folders don't stream.
485            absolute_path="" if is_dir else self._stream_url(relative_path),
486            is_dir=is_dir,
487            checksum=checksum,
488            file_size=size,
489        )
490