/
/
/
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.errors import (
18 LoginFailed,
19 ProviderUnavailableError,
20 SetupFailedError,
21)
22
23from music_assistant.providers.filesystem_cloud.base import (
24 CONF_CLIENT_ID,
25 CONF_CLIENT_SECRET,
26 CONF_FOLDER_ID,
27 CONF_REFRESH_TOKEN,
28 CloudFileSystemProvider,
29 read_setup_value,
30)
31from music_assistant.providers.filesystem_local.constants import (
32 CONF_CONTENT_TYPE,
33 CONF_ENTRY_CONTENT_TYPE,
34 CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
35 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
36 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
37 CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
38 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
39 CONF_ENTRY_MISSING_ALBUM_ARTIST,
40 CONF_ENTRY_PROPAGATE_GENRES,
41 content_type_config_entry,
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 ConfigEntry, 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 = str(
98 self.get_setup_value(CONF_CONTENT_TYPE, CONF_ENTRY_CONTENT_TYPE.default_value)
99 )
100 return (
101 content_type_config_entry(content_type),
102 CONF_ENTRY_MISSING_ALBUM_ARTIST,
103 CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
104 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
105 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
106 CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
107 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
108 CONF_ENTRY_PROPAGATE_GENRES,
109 )
110
111 @property
112 def instance_name_postfix(self) -> str | None:
113 """Return a (default) instance name postfix for this provider instance."""
114 return self._root_folder_name
115
116 async def handle_async_init(self) -> None:
117 """Handle async initialization of the provider."""
118 # verify auth works early so setup fails clearly if creds are wrong
119 try:
120 await self.api.get_user(params={"fields": "user(emailAddress)"})
121 except AuthException as err:
122 raise LoginFailed(f"Google Drive authentication failed: {err}") from err
123 except GoogleDriveApiError as err:
124 raise SetupFailedError(f"Unable to connect to Google Drive: {err}") from err
125 # verify the configured folder ID early: users tend to enter the folder
126 # name here, but Drive only accepts the opaque ID from the folder URL
127 if self.root_folder_id != "root":
128 try:
129 meta = await self._get_file_meta(self.root_folder_id, "id, name, mimeType")
130 except GoogleDriveApiError as err:
131 msg = (
132 f"Drive folder ID '{self.root_folder_id}' not found. Use the ID from the "
133 "folder URL (drive.google.com/drive/folders/<ID>), not the folder name."
134 )
135 raise SetupFailedError(msg) from err
136 if meta.get("mimeType") != FOLDER_MIME_TYPE:
137 msg = f"Drive item '{self.root_folder_id}' is not a folder."
138 raise SetupFailedError(msg)
139 self._root_folder_name = cast("str | None", meta.get("name"))
140 await self._post_init()
141
142 # ------------------------------------------------------------------
143 # cloud API hooks
144 # ------------------------------------------------------------------
145
146 async def _api_list_children(self, folder_id: str) -> list[RawItem]:
147 """List a Drive folder's children via the files.list API, following pagination."""
148 items: list[RawItem] = []
149 page_token: str | None = None
150 with _translate_errors():
151 while True:
152 params = {
153 "q": f"'{folder_id}' in parents and trashed = false",
154 "fields": f"nextPageToken, files({_FILE_FIELDS})",
155 "pageSize": 1000,
156 }
157 if page_token:
158 params["pageToken"] = page_token
159 result = await self.api.list_files(params=params)
160 for f in result.get("files", []):
161 items.append(
162 (
163 f["id"],
164 f["name"],
165 f.get("mimeType") == FOLDER_MIME_TYPE,
166 f.get("modifiedTime", "unknown"),
167 int(f["size"]) if f.get("size") else None,
168 None,
169 )
170 )
171 page_token = result.get("nextPageToken")
172 if not page_token:
173 break
174 return items
175
176 async def _api_download_bytes(self, file_id: str) -> bytes:
177 """Download a Drive file's full contents."""
178 with _translate_errors():
179 resp = await self.api.get_file_content(file_id)
180 data: bytes = await resp.read()
181 return data
182
183 async def _api_download_response(self, file_id: str, headers: dict[str, str]) -> ClientResponse:
184 """Open a streaming download for a Drive file."""
185 with _translate_errors():
186 return await self.api.get_file_content(file_id, headers=headers)
187
188 async def _get_file_meta(self, file_id: str, fields: str) -> dict[str, Any]:
189 """
190 Fetch a single file's metadata by (Drive) ID.
191
192 The library has no get-single-file helper, so we use its auth object
193 (same approach the library uses internally for get_user).
194 """
195 url = f"https://www.googleapis.com/drive/v3/files/{file_id}"
196 result: dict[str, Any] = await self.auth.get_json(url, params={"fields": fields})
197 return result
198
199
200@contextmanager
201def _translate_errors() -> Generator[None]:
202 """Translate Drive client errors into MA errors."""
203 try:
204 yield
205 except AuthException as err:
206 raise LoginFailed(f"Google Drive authentication failed: {err}") from err
207 except (GoogleDriveApiError, ClientError) as err:
208 raise ProviderUnavailableError(f"Google Drive API error: {err}") from err
209