/
/
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 await super().unload(is_removed)
187 if self._unregister_stream_route is not None:
188 self._unregister_stream_route()
189
190 async def resolve(self, file_path: str) -> FileSystemItem:
191 """Resolve a relative path to a FileSystemItem."""
192 file_path = self._normalize_path(file_path)
193 if entry := await self._lookup(file_path):
194 return entry[1]
195 raise MediaNotFoundError(f"Cloud path not found: {file_path}")
196
197 async def exists(self, file_path: str) -> bool:
198 """Check if a cloud file/folder exists."""
199 if not file_path:
200 return False
201 try:
202 return await self._lookup(self._normalize_path(file_path)) is not None
203 except ProviderUnavailableError, MediaNotFoundError:
204 return False
205
206 async def resolve_image(self, path: str) -> str | bytes:
207 """Return raw image bytes for a cloud image file or embedded cover art."""
208 # drop the cache-busting suffix the parent appends for embedded images
209 path = path.split("?cs=", 1)[0]
210 ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
211 if ext in SUPPORTED_EXTENSIONS:
212 # audio file: extract the embedded art with ffmpeg over our stream URL
213 if img_data := await get_embedded_image(self._stream_url(path)):
214 return img_data
215 raise MediaNotFoundError(f"No embedded image found: {path}")
216 return await self._read_file(path)
217
218 # ------------------------------------------------------------------
219 # API hooks (implemented by the concrete cloud provider); hooks must
220 # translate client library errors into MA errors: ProviderUnavailableError
221 # for API/transport failures, LoginFailed for authentication problems
222 # ------------------------------------------------------------------
223
224 async def _api_list_children(self, folder_id: str) -> list[RawItem]:
225 """
226 List the children of a cloud folder, following pagination if needed.
227
228 :param folder_id: The cloud provider's opaque folder ID.
229 :return: One (id, name, is_dir, checksum, size) tuple per child.
230 """
231 raise NotImplementedError
232
233 async def _api_download_bytes(self, file_id: str) -> bytes:
234 """
235 Download a (small) cloud file's full contents.
236
237 :param file_id: The cloud provider's opaque file ID.
238 """
239 raise NotImplementedError
240
241 async def _api_download_response(self, file_id: str, headers: dict[str, str]) -> ClientResponse:
242 """
243 Open a streaming download for a cloud file.
244
245 :param file_id: The cloud provider's opaque file ID.
246 :param headers: Extra request headers to forward (e.g. Range for seeking).
247 """
248 raise NotImplementedError
249
250 # ------------------------------------------------------------------
251 # initialization helpers
252 # ------------------------------------------------------------------
253
254 async def _post_init(self) -> None:
255 """Complete common initialization; call at the end of handle_async_init."""
256 self._register_stream_route()
257
258 def _register_stream_route(self) -> None:
259 """Register the dynamic route that proxies cloud downloads with fresh auth."""
260 self._unregister_stream_route = self.mass.streams.register_dynamic_route(
261 f"/{self.instance_id}_stream", self._handle_stream_request
262 )
263
264 # ------------------------------------------------------------------
265 # filesystem hooks (these are what the parent calls)
266 # ------------------------------------------------------------------
267
268 async def _is_reachable(self) -> bool:
269 """Return whether the cloud storage can be read."""
270 # this provider has no local path to stat, so ask the API for the root listing;
271 # an outage (or expired credentials) surfaces as a raised error
272 await self._scandir("", use_cache=False)
273 return True
274
275 async def _scandir(self, path: str, use_cache: bool = True) -> list[FileSystemItem]:
276 """
277 List the children of a cloud folder.
278
279 `path` is the relative path of the folder ("" means this provider's root).
280 `use_cache` allows serving a recent cached listing; pass False to force
281 a fresh fetch from the cloud API.
282 """
283 path = self._normalize_path(path)
284 # serve recently fetched listings from cache so browsing back and forth
285 # through folders doesn't cost an API round trip per click
286 if (
287 use_cache
288 and (cached := self._dir_cache.get(path)) is not None
289 and time.monotonic() < self._dir_cache_expiry.get(path, 0)
290 ):
291 return [entry[1] for entry in cached.values()]
292 folder_id = await self._resolve_id(path)
293 children: dict[str, tuple[str, FileSystemItem]] = {}
294 items: list[FileSystemItem] = []
295 for raw in await self._api_list_children(folder_id):
296 # slashes in cloud file names would corrupt our path scheme
297 name = raw[1].replace("/", "_")
298 if name in children:
299 # some clouds (e.g. Google Drive) allow duplicate names in a folder; paths can't
300 self.logger.warning(
301 "Duplicate name '%s' in folder '%s' - ignoring all but the first",
302 name,
303 path or "(root)",
304 )
305 continue
306 item = self._to_item(raw, path, name)
307 children[name] = (raw[0], item)
308 items.append(item)
309 self._dir_cache[path] = children
310 self._dir_cache_expiry[path] = time.monotonic() + self._DIR_CACHE_TTL
311 return items
312
313 async def _enumerate_files_for_sync(
314 self,
315 *,
316 file_checksums: dict[str, str],
317 cue_file_checksums: dict[str, set[str]],
318 cur_filenames: set[str],
319 items_to_process: list[tuple[FileSystemItem, str | None]],
320 unchanged_cue_items: list[FileSystemItem],
321 cue_stems: set[str],
322 scan_errors: ScanErrors,
323 ) -> None:
324 """Walk the cloud folder tree via the API and populate the sync buckets."""
325 ignore_album_playlists = self.media_content_type == "music" and bool(
326 self.config.get_value(CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS.key)
327 )
328 # mutable counter for the nested coroutine
329 scanned = [0]
330 # a cloud folder may be reachable twice (e.g. Drive multi-parent), so
331 # guard against re-visiting
332 visited: set[str] = set()
333
334 async def _walk(path: str, is_root: bool) -> None:
335 if path in visited:
336 return
337 visited.add(path)
338 try:
339 # always fetch fresh during a sync so new cloud content is
340 # picked up no matter how recently a folder was browsed
341 items = await self._scandir(path, use_cache=False)
342 except ProviderUnavailableError as err:
343 # a root-level failure aborts the sync right away, subfolder failures only
344 # once too many happen in a row, matching the local-filesystem walker
345 if not is_root:
346 self.logger.warning("Error scanning folder %s: %s", path, err)
347 scan_errors.record_dir_error(err, is_root=is_root, path=path)
348 return
349 scan_errors.record_dir_read()
350 for item in items:
351 if item.is_dir:
352 await _walk(item.relative_path, is_root=False)
353 if scan_errors.aborted:
354 return
355 continue
356 if item.ext not in SUPPORTED_EXTENSIONS:
357 continue
358 scanned[0] += 1
359 if scanned[0] % 500 == 0:
360 update_current_task_progress_text(f"Scanning files: {scanned[0]} found")
361 self._classify_scan_item(
362 item,
363 file_checksums=file_checksums,
364 cue_file_checksums=cue_file_checksums,
365 cur_filenames=cur_filenames,
366 items_to_process=items_to_process,
367 unchanged_cue_items=unchanged_cue_items,
368 cue_stems=cue_stems,
369 ignore_album_playlists=ignore_album_playlists,
370 )
371
372 await _walk("", is_root=True)
373
374 async def _read_file(self, path: str) -> bytes:
375 """Download a (small text) file's bytes: nfo, m3u, lrc, etc."""
376 file_id = await self._resolve_id(self._normalize_path(path))
377 try:
378 return await self._api_download_bytes(file_id)
379 except ProviderUnavailableError as err:
380 raise MediaNotFoundError(f"Unable to read cloud file {path}: {err}") from err
381
382 def _get_chapter_path(self, relative_path: str) -> str:
383 """Return the streamable URL for an audiobook chapter file."""
384 return self._stream_url(relative_path)
385
386 # ------------------------------------------------------------------
387 # streaming
388 # ------------------------------------------------------------------
389
390 def _stream_url(self, path: str) -> str:
391 """Build the MA-hosted URL that proxies this cloud file."""
392 base = f"{self.mass.streams.base_url}/{self.instance_id}_stream"
393 return f"{base}?path={quote(path)}"
394
395 async def _handle_stream_request(self, request: web.Request) -> web.StreamResponse:
396 """
397 Proxy a cloud download through MA, adding a fresh auth header.
398
399 Because this runs per request, the token is always valid - so even a
400 multi-hour audiobook can't outlive it.
401 """
402 path = self._normalize_path(request.query.get("path") or "")
403 if not path:
404 raise web.HTTPBadRequest(text="Missing path")
405 # the streamserver is unauthenticated: only proxy audio files so this route
406 # can't be used to download arbitrary files from the cloud account
407 # (same 404 as a missing file, so blocked paths are indistinguishable)
408 ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
409 if ext not in AUDIO_STREAM_EXTENSIONS:
410 raise web.HTTPNotFound(text="File not found")
411 try:
412 file_id = await self._resolve_id(path)
413 except MediaNotFoundError as err:
414 self.logger.debug("Cloud stream path not found: %s (%s)", path, err)
415 raise web.HTTPNotFound(text="File not found") from err
416 # forward Range header so players can seek
417 headers = {}
418 if rng := request.headers.get("Range"):
419 headers["Range"] = rng
420 try:
421 cloud_resp = await self._api_download_response(file_id, headers)
422 except (ProviderUnavailableError, LoginFailed) as err:
423 self.logger.warning("Cloud provider unavailable while streaming %s: %s", path, err)
424 raise web.HTTPBadGateway(text="Upstream provider unavailable") from err
425
426 response = web.StreamResponse(status=cloud_resp.status)
427 # copy content-type / length / range headers back to the player
428 for h in ("Content-Type", "Content-Length", "Content-Range", "Accept-Ranges"):
429 if h in cloud_resp.headers:
430 response.headers[h] = cloud_resp.headers[h]
431 try:
432 await response.prepare(request)
433 async for chunk in cloud_resp.content.iter_chunked(64 * 1024):
434 await response.write(chunk)
435 await response.write_eof()
436 except ConnectionError:
437 # client hung up early (e.g. ffmpeg closes as soon as it has read
438 # the tags); perfectly normal, not an error
439 self.logger.debug("Client disconnected while streaming %s", path)
440 except ClientError as err:
441 # the cloud side dropped mid-transfer
442 self.logger.warning("Cloud download interrupted for %s: %s", path, err)
443 finally:
444 # abort the cloud download so we don't keep pulling unneeded bytes
445 cloud_resp.close()
446 return response
447
448 # ------------------------------------------------------------------
449 # path resolution helpers
450 # ------------------------------------------------------------------
451
452 def _normalize_path(self, path: str) -> str:
453 """Normalize a relative path (collapse ./.. segments from playlist entries)."""
454 path = path.strip("/")
455 if path:
456 path = posixpath.normpath(path)
457 if path == ".":
458 path = ""
459 return path
460
461 async def _lookup(self, path: str) -> tuple[str, FileSystemItem] | None:
462 """
463 Return the cached (cloud id, item) tuple for a relative path, if it exists.
464
465 Lists the parent folder (once) on a cache miss.
466 """
467 if not path:
468 return None
469 parent, _, name = path.rpartition("/")
470 if (children := self._dir_cache.get(parent)) is None:
471 await self._scandir(parent)
472 children = self._dir_cache.get(parent, {})
473 return children.get(name)
474
475 async def _resolve_id(self, path: str) -> str:
476 """Resolve a relative path to its cloud file ID."""
477 if not path:
478 return self.root_folder_id
479 if entry := await self._lookup(path):
480 return entry[0]
481 raise MediaNotFoundError(f"Cloud path not found: {path}")
482
483 def _to_item(self, raw: RawItem, parent_path: str, name: str) -> FileSystemItem:
484 """Convert a raw API listing entry to a FileSystemItem."""
485 _, _, is_dir, checksum, size = raw
486 relative_path = f"{parent_path}/{name}" if parent_path else name
487 return FileSystemItem(
488 filename=name,
489 relative_path=relative_path,
490 # absolute_path is what the parent hands to the tag parser (ffmpeg);
491 # point it at our streaming URL so tags are read over HTTP with a
492 # fresh token - no temp download needed. Folders don't stream.
493 absolute_path="" if is_dir else self._stream_url(relative_path),
494 is_dir=is_dir,
495 checksum=checksum,
496 file_size=size,
497 )
498