/
/
/
1"""
2Google Drive File System Provider for Music Assistant.
3
4All filesystem/sync/streaming logic lives in CloudFileSystemProvider; this
5module only supplies the Google-specific parts: OAuth2 auth (see auth.py),
6the folder-ID config check and the three Drive API hooks.
7"""
8
9from __future__ import annotations
10
11from contextlib import contextmanager
12from typing import TYPE_CHECKING, Any, cast
13
14from aiohttp import ClientError
15from google_drive_api.api import GoogleDriveApi
16from google_drive_api.exceptions import AuthException, GoogleDriveApiError
17from music_assistant_models.config_entries import ConfigEntry
18from music_assistant_models.enums import ConfigEntryType
19from music_assistant_models.errors import (
20 LoginFailed,
21 ProviderUnavailableError,
22 SetupFailedError,
23)
24
25from music_assistant.providers.filesystem_cloud.base import (
26 CONF_CLIENT_ID,
27 CONF_CLIENT_SECRET,
28 CONF_FOLDER_ID,
29 CONF_REFRESH_TOKEN,
30 CloudFileSystemProvider,
31 read_setup_value,
32)
33from music_assistant.providers.filesystem_local.constants import (
34 CONF_CONTENT_TYPE,
35 CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
36 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
37 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
38 CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
39 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
40 CONF_ENTRY_MISSING_ALBUM_ARTIST,
41 CONF_ENTRY_PROPAGATE_GENRES,
42)
43
44from .auth import MAGoogleDriveAuth
45from .constants import FOLDER_MIME_TYPE
46
47if TYPE_CHECKING:
48 from collections.abc import Generator
49
50 from aiohttp import ClientResponse
51 from music_assistant_models.config_entries import ProviderConfig
52 from music_assistant_models.provider import ProviderManifest
53
54 from music_assistant.mass import MusicAssistant
55 from music_assistant.providers.filesystem_cloud.base import RawItem
56
57# fields we ask Google to return for each file
58_FILE_FIELDS = "id, name, mimeType, size, modifiedTime"
59
60
61class GoogleDriveFileSystemProvider(CloudFileSystemProvider):
62 """Google Drive File System Provider for Music Assistant."""
63
64 def __init__(
65 self,
66 mass: MusicAssistant,
67 manifest: ProviderManifest,
68 config: ProviderConfig,
69 ) -> None:
70 """Initialize Google Drive FileSystem Provider."""
71 # the "root" for this provider is a Google Drive folder ID; read it setup-data-aware
72 # here since the instance (and self.get_setup_value) does not exist yet
73 super().__init__(
74 mass,
75 manifest,
76 config,
77 cast("str", read_setup_value(mass, config, CONF_FOLDER_ID) or "root"),
78 )
79 self.auth = MAGoogleDriveAuth(
80 mass,
81 cast("str", self.get_setup_value(CONF_CLIENT_ID)),
82 cast("str", self.get_setup_value(CONF_CLIENT_SECRET)),
83 cast("str", self.get_setup_value(CONF_REFRESH_TOKEN)),
84 )
85 self.api = GoogleDriveApi(self.auth)
86 self._root_folder_name: str | None = None
87
88 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
89 """
90 Return Config entries to setup this provider.
91
92 Credentials, the content type and root folder are collected by the setup flow (see
93 setup_flow.py); only the genuine sync options are configurable here.
94 """
95 # the content type is set by the setup flow; surface it read-only so the sync
96 # options' depends_on chains still resolve
97 content_type = getattr(self, "media_content_type", "music")
98 return (
99 ConfigEntry(key=CONF_CONTENT_TYPE, type=ConfigEntryType.LABEL, value=content_type),
100 CONF_ENTRY_MISSING_ALBUM_ARTIST,
101 CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
102 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
103 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
104 CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
105 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
106 CONF_ENTRY_PROPAGATE_GENRES,
107 )
108
109 @property
110 def instance_name_postfix(self) -> str | None:
111 """Return a (default) instance name postfix for this provider instance."""
112 return self._root_folder_name
113
114 async def handle_async_init(self) -> None:
115 """Handle async initialization of the provider."""
116 # verify auth works early so setup fails clearly if creds are wrong
117 try:
118 await self.api.get_user(params={"fields": "user(emailAddress)"})
119 except AuthException as err:
120 raise LoginFailed(f"Google Drive authentication failed: {err}") from err
121 except GoogleDriveApiError as err:
122 raise SetupFailedError(f"Unable to connect to Google Drive: {err}") from err
123 # verify the configured folder ID early: users tend to enter the folder
124 # name here, but Drive only accepts the opaque ID from the folder URL
125 if self.root_folder_id != "root":
126 try:
127 meta = await self._get_file_meta(self.root_folder_id, "id, name, mimeType")
128 except GoogleDriveApiError as err:
129 msg = (
130 f"Drive folder ID '{self.root_folder_id}' not found. Use the ID from the "
131 "folder URL (drive.google.com/drive/folders/<ID>), not the folder name."
132 )
133 raise SetupFailedError(msg) from err
134 if meta.get("mimeType") != FOLDER_MIME_TYPE:
135 msg = f"Drive item '{self.root_folder_id}' is not a folder."
136 raise SetupFailedError(msg)
137 self._root_folder_name = cast("str | None", meta.get("name"))
138 await self._post_init()
139
140 # ------------------------------------------------------------------
141 # cloud API hooks
142 # ------------------------------------------------------------------
143
144 async def _api_list_children(self, folder_id: str) -> list[RawItem]:
145 """List a Drive folder's children via the files.list API, following pagination."""
146 items: list[RawItem] = []
147 page_token: str | None = None
148 with _translate_errors():
149 while True:
150 params = {
151 "q": f"'{folder_id}' in parents and trashed = false",
152 "fields": f"nextPageToken, files({_FILE_FIELDS})",
153 "pageSize": 1000,
154 }
155 if page_token:
156 params["pageToken"] = page_token
157 result = await self.api.list_files(params=params)
158 for f in result.get("files", []):
159 items.append(
160 (
161 f["id"],
162 f["name"],
163 f.get("mimeType") == FOLDER_MIME_TYPE,
164 f.get("modifiedTime", "unknown"),
165 int(f["size"]) if f.get("size") else None,
166 )
167 )
168 page_token = result.get("nextPageToken")
169 if not page_token:
170 break
171 return items
172
173 async def _api_download_bytes(self, file_id: str) -> bytes:
174 """Download a Drive file's full contents."""
175 with _translate_errors():
176 resp = await self.api.get_file_content(file_id)
177 data: bytes = await resp.read()
178 return data
179
180 async def _api_download_response(self, file_id: str, headers: dict[str, str]) -> ClientResponse:
181 """Open a streaming download for a Drive file."""
182 with _translate_errors():
183 return await self.api.get_file_content(file_id, headers=headers)
184
185 async def _get_file_meta(self, file_id: str, fields: str) -> dict[str, Any]:
186 """
187 Fetch a single file's metadata by (Drive) ID.
188
189 The library has no get-single-file helper, so we use its auth object
190 (same approach the library uses internally for get_user).
191 """
192 url = f"https://www.googleapis.com/drive/v3/files/{file_id}"
193 result: dict[str, Any] = await self.auth.get_json(url, params={"fields": fields})
194 return result
195
196
197@contextmanager
198def _translate_errors() -> Generator[None]:
199 """Translate Drive client errors into MA errors."""
200 try:
201 yield
202 except AuthException as err:
203 raise LoginFailed(f"Google Drive authentication failed: {err}") from err
204 except (GoogleDriveApiError, ClientError) as err:
205 raise ProviderUnavailableError(f"Google Drive API error: {err}") from err
206