/
/
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.errors import (
21 LoginFailed,
22 ProviderUnavailableError,
23 SetupFailedError,
24)
25from onedrive_personal_sdk.clients.client import OneDriveClient
26from onedrive_personal_sdk.exceptions import AuthenticationError, OneDriveException
27from onedrive_personal_sdk.models.items import Folder
28
29from music_assistant.providers.filesystem_cloud.base import (
30 CONF_CLIENT_ID,
31 CONF_CLIENT_SECRET,
32 CONF_FOLDER_ID,
33 CONF_REFRESH_TOKEN,
34 CloudFileSystemProvider,
35 read_setup_value,
36)
37from music_assistant.providers.filesystem_local.constants import (
38 CONF_CONTENT_TYPE,
39 CONF_ENTRY_CONTENT_TYPE,
40 CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
41 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
42 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
43 CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
44 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
45 CONF_ENTRY_MISSING_ALBUM_ARTIST,
46 CONF_ENTRY_PROPAGATE_GENRES,
47 content_type_config_entry,
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 ConfigEntry, 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 = str(
102 self.get_setup_value(CONF_CONTENT_TYPE, CONF_ENTRY_CONTENT_TYPE.default_value)
103 )
104 return (
105 content_type_config_entry(content_type),
106 CONF_ENTRY_MISSING_ALBUM_ARTIST,
107 CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
108 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
109 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
110 CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
111 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
112 CONF_ENTRY_PROPAGATE_GENRES,
113 )
114
115 @property
116 def instance_name_postfix(self) -> str | None:
117 """Return a (default) instance name postfix for this provider instance."""
118 return self._root_folder_name
119
120 async def handle_async_init(self) -> None:
121 """Handle async initialization of the provider."""
122 # a single API call validates auth and (if configured) the folder path
123 if self.root_folder_id == "root":
124 try:
125 await self.client.get_drive_item("root")
126 except AuthenticationError as err:
127 raise LoginFailed(f"OneDrive authentication failed: {err}") from err
128 except OneDriveException as err:
129 raise SetupFailedError(f"Unable to connect to OneDrive: {err}") from err
130 else:
131 await self._resolve_root_folder()
132 await self._post_init()
133
134 # ------------------------------------------------------------------
135 # cloud API hooks
136 # ------------------------------------------------------------------
137
138 async def _api_list_children(self, folder_id: str) -> list[RawItem]:
139 """List a OneDrive folder's children."""
140 try:
141 # the SDK follows pagination internally
142 items = await self.client.list_drive_items(folder_id)
143 except AuthenticationError as err:
144 raise LoginFailed(f"OneDrive authentication failed: {err}") from err
145 except OneDriveException as err:
146 raise ProviderUnavailableError(f"OneDrive API error: {err}") from err
147 out: list[RawItem] = []
148 for item in items:
149 if isinstance(item, Folder):
150 out.append((item.id, item.name, True, "folder", item.size))
151 continue
152 # quickXorHash is a stable content hash; not every file has one,
153 # so fall back to the size
154 checksum = item.hashes.quick_xor_hash or str(item.size)
155 out.append((item.id, item.name, False, checksum, item.size))
156 return out
157
158 async def _api_download_bytes(self, file_id: str) -> bytes:
159 """Download a OneDrive file's full contents."""
160 try:
161 stream = await self.client.download_drive_item(file_id)
162 return await stream.read()
163 except AuthenticationError as err:
164 raise LoginFailed(f"OneDrive authentication failed: {err}") from err
165 except OneDriveException as err:
166 raise ProviderUnavailableError(f"OneDrive API error: {err}") from err
167
168 async def _api_download_response(self, file_id: str, headers: dict[str, str]) -> ClientResponse:
169 """Open a streaming download for a OneDrive file."""
170 # go direct to Graph instead of through the SDK so Range headers are
171 # forwarded and the response headers stay available
172 token = await self.auth.async_get_access_token()
173 url = f"{GRAPH_BASE_URL}/me/drive/items/{file_id}/content"
174 req_headers = {"Authorization": f"Bearer {token}", **headers}
175 try:
176 # Graph 302-redirects to a pre-signed download URL; aiohttp follows
177 # it and drops the auth header on the cross-host hop
178 return await self.mass.http_session.get(url, headers=req_headers)
179 except ClientError as err:
180 raise ProviderUnavailableError(f"OneDrive API error: {err}") from err
181
182 async def _resolve_root_folder(self) -> None:
183 """Resolve the configured folder path to the Graph item ID it maps to."""
184 path = self.root_folder_id.strip("/")
185 token = await self.auth.async_get_access_token()
186 # the SDK's /items/{id} syntax only takes item IDs, so address the
187 # folder by path via Graph's root-relative syntax
188 url = f"{GRAPH_BASE_URL}/me/drive/root:/{quote(path)}"
189 try:
190 async with self.mass.http_session.get(
191 url, headers={"Authorization": f"Bearer {token}"}
192 ) as resp:
193 if resp.status in (401, 403):
194 raise LoginFailed(f"OneDrive authentication failed: {await resp.text()}")
195 if resp.status == 404:
196 raise SetupFailedError(f"Folder '{path}' not found in your OneDrive")
197 resp.raise_for_status()
198 item = await resp.json()
199 except ClientError as err:
200 raise SetupFailedError(f"Unable to connect to OneDrive: {err}") from err
201 if "folder" not in item:
202 raise SetupFailedError(f"OneDrive item '{path}' is not a folder")
203 self.root_folder_id = item["id"]
204 self._root_folder_name = item["name"]
205