/
/
/
1"""
2OneDrive File System Provider for Music Assistant.
3
4All filesystem/sync/streaming logic lives in CloudFileSystemProvider; this
5module only supplies the OneDrive-specific parts: OAuth2 auth (see auth.py),
6the folder-path resolution and the three Graph API hooks.
7
8Listing/metadata goes through HA's onedrive-personal-sdk, but downloads go
9straight to Microsoft Graph so we can forward Range headers (needed for
10seeking) and read the response headers - the SDK's download helper exposes
11neither.
12"""
13
14from __future__ import annotations
15
16from typing import TYPE_CHECKING, cast
17from urllib.parse import quote
18
19from aiohttp import ClientError
20from music_assistant_models.config_entries import ConfigEntry
21from music_assistant_models.enums import ConfigEntryType
22from music_assistant_models.errors import (
23 LoginFailed,
24 ProviderUnavailableError,
25 SetupFailedError,
26)
27from onedrive_personal_sdk.clients.client import OneDriveClient
28from onedrive_personal_sdk.exceptions import AuthenticationError, OneDriveException
29from onedrive_personal_sdk.models.items import Folder
30
31from music_assistant.providers.filesystem_cloud.base import (
32 CONF_CLIENT_ID,
33 CONF_CLIENT_SECRET,
34 CONF_FOLDER_ID,
35 CONF_REFRESH_TOKEN,
36 CloudFileSystemProvider,
37 read_setup_value,
38)
39from music_assistant.providers.filesystem_local.constants import (
40 CONF_CONTENT_TYPE,
41 CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
42 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
43 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
44 CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
45 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
46 CONF_ENTRY_MISSING_ALBUM_ARTIST,
47 CONF_ENTRY_PROPAGATE_GENRES,
48)
49
50from .auth import MAOneDriveAuth
51from .constants import GRAPH_BASE_URL
52
53if TYPE_CHECKING:
54 from aiohttp import ClientResponse
55 from music_assistant_models.config_entries import ProviderConfig
56 from music_assistant_models.provider import ProviderManifest
57
58 from music_assistant.mass import MusicAssistant
59 from music_assistant.providers.filesystem_cloud.base import RawItem
60
61
62class OneDriveFileSystemProvider(CloudFileSystemProvider):
63 """OneDrive File System Provider for Music Assistant."""
64
65 def __init__(
66 self,
67 mass: MusicAssistant,
68 manifest: ProviderManifest,
69 config: ProviderConfig,
70 ) -> None:
71 """Initialize OneDrive FileSystem Provider."""
72 # the configured "root" is a folder path; handle_async_init resolves it to the Graph
73 # item ID everything else works off. Read it setup-data-aware here since the instance
74 # (and self.get_setup_value) does not exist yet
75 super().__init__(
76 mass,
77 manifest,
78 config,
79 cast("str", read_setup_value(mass, config, CONF_FOLDER_ID) or "root"),
80 )
81 self.auth = MAOneDriveAuth(
82 mass,
83 config.instance_id,
84 cast("str", self.get_setup_value(CONF_CLIENT_ID)),
85 cast("str", self.get_setup_value(CONF_CLIENT_SECRET)),
86 cast("str", self.get_setup_value(CONF_REFRESH_TOKEN)),
87 )
88 # the SDK just needs a coroutine that returns a fresh access token
89 self.client = OneDriveClient(self.auth.async_get_access_token, mass.http_session)
90 self._root_folder_name: str | None = None
91
92 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
93 """
94 Return Config entries to setup this provider.
95
96 Credentials, the content type and root folder are collected by the setup flow (see
97 setup_flow.py); only the genuine sync options are configurable here.
98 """
99 # the content type is set by the setup flow; surface it read-only so the sync
100 # options' depends_on chains still resolve
101 content_type = getattr(self, "media_content_type", "music")
102 return (
103 ConfigEntry(key=CONF_CONTENT_TYPE, type=ConfigEntryType.LABEL, value=content_type),
104 CONF_ENTRY_MISSING_ALBUM_ARTIST,
105 CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
106 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
107 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
108 CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
109 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
110 CONF_ENTRY_PROPAGATE_GENRES,
111 )
112
113 @property
114 def instance_name_postfix(self) -> str | None:
115 """Return a (default) instance name postfix for this provider instance."""
116 return self._root_folder_name
117
118 async def handle_async_init(self) -> None:
119 """Handle async initialization of the provider."""
120 # a single API call validates auth and (if configured) the folder path
121 if self.root_folder_id == "root":
122 try:
123 await self.client.get_drive_item("root")
124 except AuthenticationError as err:
125 raise LoginFailed(f"OneDrive authentication failed: {err}") from err
126 except OneDriveException as err:
127 raise SetupFailedError(f"Unable to connect to OneDrive: {err}") from err
128 else:
129 await self._resolve_root_folder()
130 await self._post_init()
131
132 # ------------------------------------------------------------------
133 # cloud API hooks
134 # ------------------------------------------------------------------
135
136 async def _api_list_children(self, folder_id: str) -> list[RawItem]:
137 """List a OneDrive folder's children."""
138 try:
139 # the SDK follows pagination internally
140 items = await self.client.list_drive_items(folder_id)
141 except AuthenticationError as err:
142 raise LoginFailed(f"OneDrive authentication failed: {err}") from err
143 except OneDriveException as err:
144 raise ProviderUnavailableError(f"OneDrive API error: {err}") from err
145 out: list[RawItem] = []
146 for item in items:
147 if isinstance(item, Folder):
148 out.append((item.id, item.name, True, "folder", item.size))
149 continue
150 # quickXorHash is a stable content hash; not every file has one,
151 # so fall back to the size
152 checksum = item.hashes.quick_xor_hash or str(item.size)
153 out.append((item.id, item.name, False, checksum, item.size))
154 return out
155
156 async def _api_download_bytes(self, file_id: str) -> bytes:
157 """Download a OneDrive file's full contents."""
158 try:
159 stream = await self.client.download_drive_item(file_id)
160 return await stream.read()
161 except AuthenticationError as err:
162 raise LoginFailed(f"OneDrive authentication failed: {err}") from err
163 except OneDriveException as err:
164 raise ProviderUnavailableError(f"OneDrive API error: {err}") from err
165
166 async def _api_download_response(self, file_id: str, headers: dict[str, str]) -> ClientResponse:
167 """Open a streaming download for a OneDrive file."""
168 # go direct to Graph instead of through the SDK so Range headers are
169 # forwarded and the response headers stay available
170 token = await self.auth.async_get_access_token()
171 url = f"{GRAPH_BASE_URL}/me/drive/items/{file_id}/content"
172 req_headers = {"Authorization": f"Bearer {token}", **headers}
173 try:
174 # Graph 302-redirects to a pre-signed download URL; aiohttp follows
175 # it and drops the auth header on the cross-host hop
176 return await self.mass.http_session.get(url, headers=req_headers)
177 except ClientError as err:
178 raise ProviderUnavailableError(f"OneDrive API error: {err}") from err
179
180 async def _resolve_root_folder(self) -> None:
181 """Resolve the configured folder path to the Graph item ID it maps to."""
182 path = self.root_folder_id.strip("/")
183 token = await self.auth.async_get_access_token()
184 # the SDK's /items/{id} syntax only takes item IDs, so address the
185 # folder by path via Graph's root-relative syntax
186 url = f"{GRAPH_BASE_URL}/me/drive/root:/{quote(path)}"
187 try:
188 async with self.mass.http_session.get(
189 url, headers={"Authorization": f"Bearer {token}"}
190 ) as resp:
191 if resp.status in (401, 403):
192 raise LoginFailed(f"OneDrive authentication failed: {await resp.text()}")
193 if resp.status == 404:
194 raise SetupFailedError(f"Folder '{path}' not found in your OneDrive")
195 resp.raise_for_status()
196 item = await resp.json()
197 except ClientError as err:
198 raise SetupFailedError(f"Unable to connect to OneDrive: {err}") from err
199 if "folder" not in item:
200 raise SetupFailedError(f"OneDrive item '{path}' is not a folder")
201 self.root_folder_id = item["id"]
202 self._root_folder_name = item["name"]
203