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