--- /dev/null
+"""YouSee Musik musicprovider support for MusicAssistant."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
+from music_assistant_models.enums import (
+ ConfigEntryType,
+ ProviderFeature,
+)
+
+from music_assistant.constants import (
+ CONF_PASSWORD,
+ CONF_USERNAME,
+)
+from music_assistant.providers.yousee.constants import CONF_QUALITY
+from music_assistant.providers.yousee.provider import YouSeeMusikProvider
+
+if TYPE_CHECKING:
+ from music_assistant_models.config_entries import ProviderConfig
+ from music_assistant_models.provider import ProviderManifest
+
+ from music_assistant.mass import MusicAssistant
+ from music_assistant.models import ProviderInstanceType
+
+
+SUPPORTED_FEATURES = {
+ ProviderFeature.BROWSE,
+ ProviderFeature.SEARCH,
+ ProviderFeature.RECOMMENDATIONS,
+ ProviderFeature.LIBRARY_ARTISTS,
+ ProviderFeature.LIBRARY_ALBUMS,
+ ProviderFeature.LIBRARY_TRACKS,
+ ProviderFeature.LIBRARY_PLAYLISTS,
+ ProviderFeature.ARTIST_ALBUMS,
+ ProviderFeature.ARTIST_TOPTRACKS,
+ ProviderFeature.LIBRARY_ARTISTS_EDIT,
+ ProviderFeature.LIBRARY_ALBUMS_EDIT,
+ ProviderFeature.LIBRARY_TRACKS_EDIT,
+ ProviderFeature.LIBRARY_PLAYLISTS_EDIT,
+ ProviderFeature.PLAYLIST_TRACKS_EDIT,
+ ProviderFeature.PLAYLIST_CREATE,
+ ProviderFeature.SIMILAR_TRACKS,
+ ProviderFeature.LYRICS,
+}
+
+
+async def setup(
+ mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
+) -> ProviderInstanceType:
+ """Initialize provider(instance) with given configuration."""
+ # setup is called when the user wants to setup a new provider instance.
+ # you are free to do any preflight checks here and but you must return
+ # an instance of the provider.
+ return YouSeeMusikProvider(mass, manifest, config, SUPPORTED_FEATURES)
+
+
+async def get_config_entries(
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
+) -> tuple[ConfigEntry, ...]:
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
+ return (
+ ConfigEntry(
+ key=CONF_USERNAME,
+ type=ConfigEntryType.STRING,
+ label="Username",
+ required=True,
+ ),
+ ConfigEntry(
+ key=CONF_PASSWORD,
+ type=ConfigEntryType.SECURE_STRING,
+ label="Password",
+ required=True,
+ ),
+ ConfigEntry(
+ key=CONF_QUALITY,
+ type=ConfigEntryType.INTEGER,
+ label="Stream Quality",
+ description="The streaming quality to use for playback",
+ default_value=320,
+ options=[
+ ConfigValueOption('"High" - MP4 320kbps', 320),
+ ConfigValueOption('"Normal" - MP4 192kbps', 192),
+ ],
+ ),
+ )
--- /dev/null
+"""API Client for YouSee Musik."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from music_assistant_models.errors import (
+ LoginFailed,
+)
+
+from music_assistant.constants import VERBOSE_LOG_LEVEL
+from music_assistant.helpers.json import json_dumps
+from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
+from music_assistant.providers.yousee.constants import MAX_PAGES_PAGINATED, PAGE_SIZE
+
+if TYPE_CHECKING:
+ from collections.abc import AsyncGenerator
+
+ from music_assistant.providers.yousee.provider import YouSeeMusikProvider
+
+
+JsonLike = dict[str, Any]
+
+
+class YouSeeGraphQLError(Exception):
+ """YouSee Musik GraphQL error."""
+
+ def __init__(self, data: JsonLike) -> None:
+ """Initialize YouSeeGraphQLError."""
+ super().__init__(json_dumps(data))
+
+
+class YouSeeAPIClient:
+ """Client for interacting with YouSee API."""
+
+ YOUSEE_GRAPHQL_ENDPOINT = "https://graphql-1458.api.247e.com/graphql"
+
+ # Unsure if yousee enforces rate limiting, this is just a sane precaution
+ throttler = ThrottlerManager(rate_limit=4, period=1)
+
+ def __init__(self, provider: YouSeeMusikProvider):
+ """Initialize API client."""
+ self.provider = provider
+ self.auth = provider.auth
+ self.logger = provider.logger
+ self.mass = provider.mass
+
+ @throttle_with_retries # type: ignore[type-var]
+ async def post_graphql(
+ self, query: str, variables: JsonLike, _headers: JsonLike | None = None
+ ) -> JsonLike:
+ """Post GraphQL query to YouSee endpoint with authorization."""
+ locale = self.mass.metadata.locale.split("_")[0]
+
+ async with self.mass.http_session.post(
+ self.YOUSEE_GRAPHQL_ENDPOINT,
+ json={"query": query, "variables": variables},
+ headers={
+ "Authorization": f"Bearer {await self.auth.auth_token()}",
+ "Accept-Language": locale,
+ }
+ | (_headers or {}),
+ ) as resp:
+ if resp.status in {401, 403}:
+ # Invalidate token
+ self.auth.invalidate()
+ raise LoginFailed("Authentication with YouSee failed")
+
+ resp.raise_for_status()
+
+ result = await resp.json()
+ if len(result.get("errors", [])) > 0:
+ raise YouSeeGraphQLError(result)
+
+ return dict(result)
+
+ async def paginate_graphql(
+ self,
+ query: str,
+ variables: JsonLike,
+ page_path: list[str],
+ variables_first_key: str = "first",
+ variables_after_key: str = "after",
+ ) -> AsyncGenerator[JsonLike, None]:
+ """Paginate GraphQL results."""
+ after = None
+ has_more = True
+ i = 0
+ while has_more and (i < MAX_PAGES_PAGINATED):
+ self.logger.log(VERBOSE_LOG_LEVEL, "Paginating GraphQL query, page %s", i + 1)
+ vars_with_pagination = variables | {
+ variables_first_key: PAGE_SIZE,
+ variables_after_key: after,
+ }
+ result = await self.post_graphql(query, vars_with_pagination)
+
+ # Navigate to the page containing items and pageInfo
+ page_data = result
+ for key in page_path:
+ page_data = page_data.get(key, {})
+
+ for item in page_data.get("items", []):
+ yield item
+
+ page_info = page_data.get("pageInfo", {})
+ has_more = page_info.get("hasNextPage", False)
+ after = page_info.get("endCursor", None)
+ i += 1
--- /dev/null
+"""YouSee Musik authentication manager."""
+
+import re
+import time
+from typing import TYPE_CHECKING
+
+from music_assistant.constants import CONF_PASSWORD, CONF_USERNAME
+from music_assistant.helpers.util import (
+ lock,
+ try_parse_int,
+)
+from music_assistant.providers.yousee.api_client import JsonLike
+
+if TYPE_CHECKING:
+ from music_assistant.providers.yousee.provider import YouSeeMusikProvider
+
+
+class YouSeeAccessToken:
+ """YouSee Musik access token wrapper."""
+
+ def __init__(self, access_token: str) -> None:
+ """Initialize YouSeeAccessToken."""
+ self._access_token = access_token
+ self._token_parts = self._parse_access_token(access_token)
+
+ def is_expired(self) -> bool:
+ """Return True if token is expired."""
+ expires_at = try_parse_int(self._token_parts.get("ExpiresOn", 0))
+ return not expires_at or expires_at <= time.time()
+
+ def _parse_access_token(self, token: str) -> JsonLike:
+ return dict(part.split("=", 1) for part in token.split("&") if "=" in part)
+
+ def __str__(self) -> str:
+ """Return string representation of the access token."""
+ return self._access_token
+
+
+class YouSeeAuthManager:
+ """YouSee Musik authentication manager."""
+
+ def __init__(self, provider: "YouSeeMusikProvider"):
+ """Initialize YouSeeAuthManager."""
+ self._access_token: YouSeeAccessToken | None = None
+ self._refresh_token: str | None = None
+ self.mass = provider.mass
+ self.provider = provider
+ self.logger = provider.logger
+
+ def invalidate(self) -> None:
+ """Invalidate current access token."""
+ self._access_token = None
+
+ @lock
+ async def auth_token(self) -> YouSeeAccessToken | None:
+ """Authenticate and return access token."""
+ if self._access_token and not self._access_token.is_expired():
+ return self._access_token
+
+ # Try refresh token flow first
+ if self._refresh_token:
+ self.logger.debug("Trying to fetch refresh token")
+
+ async with self.mass.http_session.post(
+ "https://musik.yousee.dk/api/token", data={"refresh_token": self._refresh_token}
+ ) as refresh_response:
+ refresh_result = await refresh_response.json()
+ if refresh_result.get("status", 4) == 0:
+ access_token = refresh_result["tokenResult"]["access_token"]
+
+ self.logger.debug("Refresh token flow success")
+ self._access_token = YouSeeAccessToken(access_token)
+ self._refresh_token = refresh_result["tokenResult"]["refresh_token"]
+ return self._access_token
+
+ async with (
+ self.mass.http_session.get(
+ "https://musik.yousee.dk/api/delegatedlogin"
+ ) as delegate_response,
+ ):
+ post_action_re = re.search('action="([^"]+)"', await delegate_response.text())
+ if not post_action_re:
+ return None
+
+ cookies = delegate_response.cookies
+
+ async with self.mass.http_session.post(
+ f"https://login.yousee.dk{post_action_re.group(1)}",
+ data={
+ "pf.username": self.provider.config.get_value(CONF_USERNAME),
+ "pf.pass": self.provider.config.get_value(CONF_PASSWORD),
+ "pf.ok": "clicked",
+ "pf.adapterId": "MusicUsernamePasswordAdapter",
+ },
+ cookies=cookies,
+ ) as login_response:
+ access_token_re = re.search(
+ r'localStorage.setItem\("accesstoken", "([^"]+)"',
+ await login_response.text(),
+ )
+
+ refresh_token_re = re.search(
+ r'localStorage.setItem\("refreshtoken", "([^"]+)"',
+ await login_response.text(),
+ )
+
+ if not access_token_re or not refresh_token_re:
+ return None
+
+ access_token = access_token_re.group(1)
+ self._refresh_token = refresh_token_re.group(1)
+
+ self._access_token = YouSeeAccessToken(access_token)
+ self.logger.debug("Got new auth token")
+
+ return self._access_token
--- /dev/null
+"""Constants for the YouSee Musik music provider."""
+
+VARIOUS_ARTISTS_ID = "1776"
+
+PAGE_SIZE = 50
+# to avoid infinite loops, this effectively limits any album/playlist to
+# PAGE_SIZE * MAX_PAGES_PAGINATED items (1000 items with the current settings)
+MAX_PAGES_PAGINATED = 20
+GET_POPULAR_TRACKS_LIMIT = 25
+
+IMAGE_SIZE = 512
+
+CONF_QUALITY = "yousee_quality"
--- /dev/null
+<svg width="256" height="256" viewBox="0 0 32 32" fill="none" xmlns="http://www.w3.org/2000/svg">
+<g clip-path="url(#clip0_4133_720)">
+<path d="M32 16C32 7.16344 24.8366 0 16 0C7.16344 0 0 7.16344 0 16C0 24.8366 7.16344 32 16 32C24.8366 32 32 24.8366 32 16Z" fill="#00DC00"/>
+<path fill-rule="evenodd" clip-rule="evenodd" d="M22.5353 6.99902L22.5348 19.4181C22.5348 20.2092 22.353 20.5992 21.9951 21.0596C21.5616 21.6172 20.9056 22.0619 20.1394 22.2998C19.3736 22.5375 18.5861 22.541 17.9221 22.3161C17.2578 22.0909 16.6655 21.6113 16.458 20.8706C16.2521 20.136 16.4997 19.4113 16.9307 18.8568C17.3642 18.2992 18.0202 17.8545 18.7864 17.6167C19.5522 17.3789 20.3397 17.3754 21.0037 17.6004C21.0456 17.6146 21.0873 17.6298 21.1285 17.646L21.129 9.25545C21.038 9.29277 20.9417 9.33139 20.8403 9.37097C20.0273 9.68831 18.8811 10.0697 17.5481 10.3245C16.2374 10.5749 15.1097 10.7003 14.3076 10.7631C14.0832 10.7807 13.8841 10.7934 13.7134 10.8026L13.7129 21.3799C13.7129 22.1592 13.5278 22.5522 13.1722 23.0096C12.7388 23.5672 12.0828 24.0119 11.3166 24.2498C10.5508 24.4875 9.76324 24.491 9.09931 24.2661C8.43495 24.041 7.8427 23.5613 7.63514 22.8206C7.42926 22.086 7.67689 21.3613 8.10793 20.8068C8.54138 20.2492 9.19738 19.8045 9.9636 19.5667C10.7294 19.3289 11.5169 19.3254 12.1809 19.5504C12.2231 19.5647 12.2651 19.58 12.3067 19.5964L12.3071 9.42022H13.0099L13.0127 9.42021L13.0257 9.42013C13.0378 9.42003 13.0568 9.41981 13.0824 9.41937C13.1336 9.41849 13.2113 9.41669 13.3132 9.41303C13.5171 9.4057 13.8178 9.39094 14.1978 9.36118C14.9581 9.30162 16.0334 9.18224 17.2841 8.9432C18.5125 8.70844 19.5743 8.35554 20.329 8.06098C20.7056 7.91397 21.0038 7.78223 21.2059 7.68829C21.3069 7.64134 21.3838 7.6039 21.4342 7.5788C21.4594 7.56625 21.478 7.5568 21.4897 7.55079L21.5021 7.5444L21.5039 7.54346L22.5353 6.99902ZM20.5524 18.9322C20.2044 18.8143 19.7232 18.7983 19.2033 18.9597C18.6839 19.1209 18.2807 19.4115 18.041 19.7198C17.7989 20.0313 17.7594 20.3035 17.812 20.4912C17.863 20.6729 18.0258 20.8664 18.3734 20.9842C18.7214 21.1021 19.2026 21.1181 19.7224 20.9567C20.2419 20.7955 20.6451 20.5049 20.8848 20.1966C21.1269 19.8851 21.1664 19.6129 21.1138 19.4253C21.0628 19.2436 20.9 19.05 20.5524 18.9322ZM11.7296 20.8822C11.3816 20.7643 10.9004 20.7483 10.3805 20.9097C9.86109 21.0709 9.45791 21.3615 9.21821 21.6698C8.9761 21.9813 8.93662 22.2535 8.98922 22.4412C9.04014 22.6229 9.20302 22.8164 9.55059 22.9342C9.89858 23.0521 10.3798 23.0681 10.8996 22.9067C11.4191 22.7455 11.8223 22.4549 12.062 22.1466C12.3041 21.8351 12.3435 21.563 12.2909 21.3753C12.24 21.1936 12.0771 21 11.7296 20.8822Z" fill="#323232"/>
+</g>
+<defs>
+<clipPath id="clip0_4133_720">
+<rect width="32" height="32" fill="white"/>
+</clipPath>
+</defs>
+</svg>
--- /dev/null
+<svg width="256" height="256" viewBox="0 0 32 32" fill="none" xmlns="http://www.w3.org/2000/svg">
+<path mask="url(#logo-mask)" d="M32 16C32 7.16344 24.8366 0 16 0C7.16344 0 0 7.16344 0 16C0 24.8366 7.16344 32 16 32C24.8366 32 32 24.8366 32 16Z" fill="white"/>
+<mask id="logo-mask">
+ <rect width="32" height="32" fill="white"/>
+ <path d="M22.5353 6.99902L22.5348 19.4181C22.5348 20.2092 22.353 20.5992 21.9951 21.0596C21.5616 21.6172 20.9056 22.0619 20.1394 22.2998C19.3736 22.5375 18.5861 22.541 17.9221 22.3161C17.2578 22.0909 16.6655 21.6113 16.458 20.8706C16.2521 20.136 16.4997 19.4113 16.9307 18.8568C17.3642 18.2992 18.0202 17.8545 18.7864 17.6167C19.5522 17.3789 20.3397 17.3754 21.0037 17.6004C21.0456 17.6146 21.0873 17.6298 21.1285 17.646L21.129 9.25545C21.038 9.29277 20.9417 9.33139 20.8403 9.37097C20.0273 9.68831 18.8811 10.0697 17.5481 10.3245C16.2374 10.5749 15.1097 10.7003 14.3076 10.7631C14.0832 10.7807 13.8841 10.7934 13.7134 10.8026L13.7129 21.3799C13.7129 22.1592 13.5278 22.5522 13.1722 23.0096C12.7388 23.5672 12.0828 24.0119 11.3166 24.2498C10.5508 24.4875 9.76324 24.491 9.09931 24.2661C8.43495 24.041 7.8427 23.5613 7.63514 22.8206C7.42926 22.086 7.67689 21.3613 8.10793 20.8068C8.54138 20.2492 9.19738 19.8045 9.9636 19.5667C10.7294 19.3289 11.5169 19.3254 12.1809 19.5504C12.2231 19.5647 12.2651 19.58 12.3067 19.5964L12.3071 9.42022H13.0099L13.0127 9.42021L13.0257 9.42013C13.0378 9.42003 13.0568 9.41981 13.0824 9.41937C13.1336 9.41849 13.2113 9.41669 13.3132 9.41303C13.5171 9.4057 13.8178 9.39094 14.1978 9.36118C14.9581 9.30162 16.0334 9.18224 17.2841 8.9432C18.5125 8.70844 19.5743 8.35554 20.329 8.06098C20.7056 7.91397 21.0038 7.78223 21.2059 7.68829C21.3069 7.64134 21.3838 7.6039 21.4342 7.5788C21.4594 7.56625 21.478 7.5568 21.4897 7.55079L21.5021 7.5444L21.5039 7.54346L22.5353 6.99902ZM20.5524 18.9322C20.2044 18.8143 19.7232 18.7983 19.2033 18.9597C18.6839 19.1209 18.2807 19.4115 18.041 19.7198C17.7989 20.0313 17.7594 20.3035 17.812 20.4912C17.863 20.6729 18.0258 20.8664 18.3734 20.9842C18.7214 21.1021 19.2026 21.1181 19.7224 20.9567C20.2419 20.7955 20.6451 20.5049 20.8848 20.1966C21.1269 19.8851 21.1664 19.6129 21.1138 19.4253C21.0628 19.2436 20.9 19.05 20.5524 18.9322ZM11.7296 20.8822C11.3816 20.7643 10.9004 20.7483 10.3805 20.9097C9.86109 21.0709 9.45791 21.3615 9.21821 21.6698C8.9761 21.9813 8.93662 22.2535 8.98922 22.4412C9.04014 22.6229 9.20302 22.8164 9.55059 22.9342C9.89858 23.0521 10.3798 23.0681 10.8996 22.9067C11.4191 22.7455 11.8223 22.4549 12.062 22.1466C12.3041 21.8351 12.3435 21.563 12.2909 21.3753C12.24 21.1936 12.0771 21 11.7296 20.8822Z" fill="black"/>
+</mask>
+</svg>
--- /dev/null
+"""Library management for YouSee Musik."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+from typing import TYPE_CHECKING
+
+from music_assistant_models.enums import MediaType
+from music_assistant_models.errors import InvalidDataError
+
+from music_assistant.constants import VERBOSE_LOG_LEVEL
+from music_assistant.providers.yousee.constants import IMAGE_SIZE
+from music_assistant.providers.yousee.parsers import (
+ parse_album,
+ parse_artist,
+ parse_playlist,
+ parse_track,
+)
+
+if TYPE_CHECKING:
+ from music_assistant_models.media_items import Album, Artist, MediaItemType, Playlist, Track
+
+ from music_assistant.providers.yousee.provider import YouSeeMusikProvider
+
+
+class YouSeeLibraryManager:
+ """Manages YouSee Musik library operations."""
+
+ def __init__(self, provider: YouSeeMusikProvider):
+ """Initialize library manager."""
+ self.provider = provider
+ self.api = provider.api
+ self.auth = provider.auth
+ self.logger = provider.logger
+
+ async def get_artists(self) -> AsyncGenerator[Artist, None]:
+ """Retrieve library artists from the provider."""
+ query = """
+ query favoriteArtists($first: Int!, $after: String, $imageSize: Int = 512) {
+ me {
+ favorites {
+ artists(first: $first, after: $after) {
+ totalCount,
+ pageInfo {
+ endCursor
+ hasNextPage
+ }
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ share
+ }
+ }
+ }
+ }
+ }
+ """
+ variables = {"imageSize": IMAGE_SIZE}
+
+ async for item in self.api.paginate_graphql(
+ query, variables, ["data", "me", "favorites", "artists"]
+ ):
+ self.logger.log(VERBOSE_LOG_LEVEL, "Parsing artist item: %s", item)
+ yield parse_artist(self.provider, item)
+
+ async def get_albums(self) -> AsyncGenerator[Album, None]:
+ """Retrieve library albums from the provider."""
+ query = """
+ query favoriteAlbums($first: Int!, $after: String, $imageSize: Int = 512) {
+ me {
+ favorites {
+ albums(first: $first, after: $after) {
+ totalCount,
+ pageInfo {
+ endCursor
+ hasNextPage
+ }
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ artist {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ }
+ }
+ }
+ }
+ }
+ """
+ variables = {"imageSize": IMAGE_SIZE}
+
+ async for item in self.api.paginate_graphql(
+ query, variables, ["data", "me", "favorites", "albums"]
+ ):
+ self.logger.log(VERBOSE_LOG_LEVEL, "Parsing album item: %s", item)
+ yield await parse_album(self.provider, item)
+
+ async def get_tracks(self) -> AsyncGenerator[Track, None]:
+ """Retrieve library tracks from the provider."""
+ query = """
+ query favoriteTracks($first: Int!, $after: String, $imageSize: Int = 512) {
+ me {
+ favorites {
+ tracks(first: $first, after: $after) {
+ totalCount
+ pageInfo {
+ endCursor
+ hasNextPage
+ }
+ items {
+ id
+ title
+ availableToStream
+ album {
+ id
+ title
+ }
+ artist {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ cover(size: $imageSize)
+ duration
+ share
+ genre
+ isrc
+ featuredArtists {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ """
+ variables = {"imageSize": IMAGE_SIZE}
+
+ async for item in self.api.paginate_graphql(
+ query, variables, ["data", "me", "favorites", "tracks"]
+ ):
+ self.logger.log(VERBOSE_LOG_LEVEL, "Parsing track item: %s", item)
+ yield await parse_track(self.provider, item)
+
+ async def get_playlists(self) -> AsyncGenerator[Playlist, None]:
+ """Retrieve library/subscribed playlists from the provider."""
+ query = """
+ query favoritePlaylists($first: Int!, $after: String, $imageSize: Int = 512) {
+ me {
+ playlists {
+ combinedPlaylists(first: $first, after: $after, orderBy: MODIFIED_DATE) {
+ totalCount
+ pageInfo {
+ hasNextPage
+ endCursor
+ }
+ items {
+ id
+ title
+ isOwned
+ share
+ cover(size: $imageSize)
+ description
+ }
+ }
+ }
+ }
+ }
+ """
+ variables = {"imageSize": IMAGE_SIZE}
+ async for item in self.api.paginate_graphql(
+ query, variables, ["data", "me", "playlists", "combinedPlaylists"]
+ ):
+ self.logger.log(VERBOSE_LOG_LEVEL, "Parsing playlist item: %s", item)
+ yield await parse_playlist(self.provider, item)
+
+ async def add_item(self, item: MediaItemType) -> bool:
+ """Add item to provider's library. Return true on success."""
+ if item.media_type not in (
+ MediaType.ARTIST,
+ MediaType.ALBUM,
+ MediaType.TRACK,
+ MediaType.PLAYLIST,
+ ):
+ raise InvalidDataError(
+ f"Cannot add media type {item.media_type} to library for provider "
+ f"{self.provider.name}"
+ )
+
+ media_type_str = item.media_type.capitalize()
+
+ query = f"""
+ mutation addToLibrary($id: ID!) {{
+ favorites {{
+ add{media_type_str} (id: $id) {{
+ ok
+ }}
+ }}
+ }}
+ """
+ variables = {"id": item.item_id}
+
+ result = await self.api.post_graphql(query, variables)
+
+ return bool(
+ result.get("data", {})
+ .get("favorites", {})
+ .get(f"add{media_type_str}", {})
+ .get("ok", False)
+ )
+
+ async def remove_item(self, prov_item_id: str, media_type: MediaType) -> bool:
+ """Remove item from provider's library. Return true on success."""
+ if media_type not in (
+ MediaType.ARTIST,
+ MediaType.ALBUM,
+ MediaType.TRACK,
+ MediaType.PLAYLIST,
+ ):
+ raise InvalidDataError(
+ f"Cannot remove media type {media_type} from library for provider "
+ f"{self.provider.name}"
+ )
+
+ media_type_str = media_type.capitalize()
+
+ query = f"""
+ mutation removeFromLibrary($id: ID!) {{
+ favorites {{
+ remove{media_type_str} (id: $id) {{
+ ok
+ }}
+ }}
+ }}
+ """
+ variables = {"id": prov_item_id}
+
+ result = await self.api.post_graphql(query, variables)
+
+ return bool(
+ result.get("data", {})
+ .get("favorites", {})
+ .get(f"remove{media_type_str}", {})
+ .get("ok", False)
+ )
--- /dev/null
+{
+ "type": "music",
+ "domain": "yousee",
+ "name": "YouSee Musik",
+ "stage": "experimental",
+ "description": "YouSee Musik, a Danish music streaming service.",
+ "codeowners": ["@math625f"],
+ "requirements": [],
+ "documentation": "https://music-assistant.io/music-providers/yousee/",
+ "multi_instance": true
+}
--- /dev/null
+"""Media retrieval operations for YouSee Musik."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from music_assistant_models.enums import (
+ MediaType,
+)
+from music_assistant_models.errors import MediaNotFoundError
+from music_assistant_models.media_items import Album, Artist, Playlist, SearchResults, Track
+
+from music_assistant.providers.yousee.api_client import JsonLike
+from music_assistant.providers.yousee.constants import (
+ GET_POPULAR_TRACKS_LIMIT,
+ IMAGE_SIZE,
+)
+from music_assistant.providers.yousee.parsers import (
+ parse_album,
+ parse_artist,
+ parse_lyrics,
+ parse_playlist,
+ parse_track,
+)
+
+if TYPE_CHECKING:
+ from music_assistant.providers.yousee.provider import YouSeeMusikProvider
+
+
+class YouSeeMediaManager:
+ """Handles retrieval of media items from YouSee Musik."""
+
+ def __init__(self, provider: YouSeeMusikProvider):
+ """Initialize media retriever."""
+ self.provider = provider
+ self.api = provider.api
+ self.logger = provider.logger
+
+ async def search(
+ self,
+ search_query: str,
+ media_types: list[MediaType],
+ limit: int = 5,
+ ) -> SearchResults:
+ """Perform search on musicprovider.
+
+ :param search_query: Search query.
+ :param media_types: A list of media_types to include.
+ :param limit: Number of items to return in the search (per type).
+ """
+ sections = {
+ MediaType.TRACK: """
+ tracks(first: $first) {
+ items {
+ id
+ title
+ availableToStream
+ album {
+ id
+ title
+ }
+ artist {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ cover(size: $imageSize)
+ duration
+ share
+ genre
+ isrc
+ featuredArtists {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ }
+ }
+ }
+ """,
+ MediaType.ALBUM: """
+ albums(first: $first) {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ artist {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ }
+ }
+ """,
+ MediaType.ARTIST: """
+ artists(first: $first) {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ share
+ }
+ }
+ """,
+ MediaType.PLAYLIST: """
+ playlists(first: $first) {
+ items {
+ id
+ title
+ isOwned
+ share
+ cover(size: $imageSize)
+ description
+ }
+ }
+ """,
+ }
+
+ search_result = SearchResults()
+
+ media_types = [x for x in media_types if x in (sections)]
+
+ if not media_types:
+ return search_result
+
+ query = """
+ query searchMixedSections($criterion: String!, $imageSize: Int = 512, $first: Int = 5) {
+ search(criterion: $criterion) {
+ TRACK_SECTION
+ ALBUM_SECTION
+ PLAYLIST_SECTION
+ ARTIST_SECTION
+ }
+ }
+ """
+ for media_type, section in sections.items():
+ if media_type in media_types:
+ query = query.replace(f"{media_type.name}_SECTION", section)
+ else:
+ query = query.replace(f"{media_type.name}_SECTION", "")
+
+ variables = {
+ "criterion": search_query,
+ "imageSize": IMAGE_SIZE,
+ "first": limit,
+ }
+
+ result = await self.api.post_graphql(query, variables)
+
+ result = result.get("data", {}).get("search", {})
+
+ if not result:
+ return search_result
+
+ if "artists" in result:
+ search_result.artists = [
+ parse_artist(self.provider, item) for item in result["artists"].get("items", [])
+ ]
+ if "albums" in result:
+ search_result.albums = [
+ await parse_album(self.provider, item) for item in result["albums"].get("items", [])
+ ]
+ if "tracks" in result:
+ search_result.tracks = [
+ await parse_track(self.provider, item) for item in result["tracks"].get("items", [])
+ ]
+ if "playlists" in result:
+ search_result.playlists = [
+ await parse_playlist(self.provider, item)
+ for item in result["playlists"].get("items", [])
+ ]
+
+ return search_result
+
+ async def get_artist(self, prov_artist_id: str) -> Artist:
+ """Get full artist details by id."""
+ query = """
+ query Catalog($id: ID!, $imageSize: Int = 512) {
+ catalog {
+ artist(id: $id) {
+ id
+ title
+ cover(size: $imageSize)
+ share
+ }
+ }
+ }
+ """
+ variables = {"id": prov_artist_id, "imageSize": IMAGE_SIZE}
+
+ result = await self.api.post_graphql(query, variables)
+ if not result or not result.get("data", {}).get("catalog", {}).get("artist"):
+ raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
+ return parse_artist(self.provider, result["data"]["catalog"]["artist"])
+
+ async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
+ """Get a list of all albums for the given artist."""
+ query = """
+ query Catalog($id: ID!, $imageSize: Int = 512, $first: Int = 50, $after: String) {
+ catalog {
+ artist(id: $id) {
+ id
+ albums(first: $first, after: $after) {
+ totalCount
+ pageInfo {
+ hasNextPage
+ endCursor
+ }
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ }
+ }
+ }
+ }
+ """
+
+ albums = []
+ variables = {
+ "id": prov_artist_id,
+ "imageSize": IMAGE_SIZE,
+ }
+
+ async for item in self.api.paginate_graphql(
+ query,
+ variables,
+ ["data", "catalog", "artist", "albums"],
+ ):
+ albums.append(await parse_album(self.provider, item))
+
+ return albums
+
+ async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
+ """Get a list of most popular tracks for the given artist."""
+ query = """
+ query Catalog($id: ID!, $imageSize: Int = 512, $first: Int = 25) {
+ catalog {
+ artist(id: $id) {
+ id
+ title
+ cover(size: $imageSize)
+ share
+ tracks(first: $first, after: null, orderBy: POPULARITY) {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ isrc
+ duration
+ label
+ artist {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ featuredArtists {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ }
+ share
+ genre
+ }
+ }
+ }
+ }
+ }
+ """
+
+ variables = {
+ "id": prov_artist_id,
+ "imageSize": IMAGE_SIZE,
+ "first": GET_POPULAR_TRACKS_LIMIT,
+ }
+
+ result = await self.api.post_graphql(query, variables)
+
+ if not result or not result.get("data", {}).get("catalog", {}).get("artist"):
+ raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
+ tracks = []
+
+ for item in result["data"]["catalog"]["artist"]["tracks"]["items"]:
+ tracks.append(await parse_track(self.provider, item))
+
+ return tracks
+
+ async def get_album(self, prov_album_id: str) -> Album:
+ """Get full album details by id."""
+ query = """
+ query Catalog($id: ID!, $imageSize: Int = 512) {
+ catalog {
+ album(id: $id) {
+ id
+ title
+ tracksCount
+ genre
+ label
+ releaseDate
+ available
+ upc
+ type
+ share
+ cover(size: $imageSize)
+ artist {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ featuredArtists {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ }
+ }
+ }
+ }
+ """
+ variables = {"id": prov_album_id, "imageSize": IMAGE_SIZE}
+
+ result = await self.api.post_graphql(query, variables)
+ if not result or not result.get("data", {}).get("catalog", {}).get("album"):
+ raise MediaNotFoundError(f"Album {prov_album_id} not found")
+ return await parse_album(self.provider, result["data"]["catalog"]["album"])
+
+ async def _get_lyrics(self, prov_track_id: str) -> list[JsonLike]:
+ """Attempt to retrieve lyrics for the given track id."""
+ query = """
+ query Lyric($id: ID!, $first: Int = 50, $after: String) {
+ catalog {
+ track(id: $id) {
+ lyrics {
+ lrc(first: $first, after: $after) {
+ pageInfo {
+ hasNextPage
+ endCursor
+ }
+ items {
+ startInMs
+ durationInMs
+ line
+ }
+ }
+ }
+ }
+ }
+ }
+ """
+ variables = {"id": prov_track_id}
+
+ lines = []
+
+ async for line in self.api.paginate_graphql(
+ query, variables, ["data", "catalog", "track", "lyrics", "lrc"]
+ ):
+ lines.append(line)
+
+ return lines
+
+ async def get_track(self, prov_track_id: str) -> Track:
+ """Get full track details by id."""
+ query = """
+ query getTrack($id: ID!, $imageSize: Int = 512) {
+ catalog {
+ track(id: $id) {
+ id
+ title
+ duration
+ genre
+ label
+ releaseDate
+ availableToStream
+ isrc
+ share
+ cover(size: $imageSize)
+ lyrics {
+ id
+ }
+ album {
+ id
+ title
+ }
+ artist {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ featuredArtists {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ }
+ }
+ }
+ }
+ """
+ variables = {"id": prov_track_id, "imageSize": IMAGE_SIZE}
+
+ result = await self.api.post_graphql(query, variables)
+ if not result or not result.get("data", {}).get("catalog", {}).get("track"):
+ raise MediaNotFoundError(f"Track {prov_track_id} not found")
+
+ track = await parse_track(self.provider, result["data"]["catalog"]["track"])
+
+ if result["data"]["catalog"]["track"].get("lyrics"):
+ lyrics = await self._get_lyrics(prov_track_id)
+ parsed_lyrics, parsed_lrc_lyrics = await parse_lyrics(lyrics)
+
+ if parsed_lyrics:
+ self.logger.debug("Attached lyrics to track")
+ track.metadata.lyrics = parsed_lyrics
+ if parsed_lrc_lyrics:
+ self.logger.debug("Attached LRC lyrics to track")
+ track.metadata.lrc_lyrics = parsed_lrc_lyrics
+
+ return track
+
+ async def get_playlist(self, prov_playlist_id: str) -> Playlist:
+ """Get full playlist details by id."""
+ query = """
+ query getPlaylist($id: ID!, $imageSize: Int = 512) {
+ playlists {
+ playlist(id: $id) {
+ id
+ title
+ description
+ tracksCount
+ createdAt
+ isOwned
+ share
+ cover(size: $imageSize)
+ }
+ }
+ }
+ """
+ variables = {"id": prov_playlist_id, "imageSize": IMAGE_SIZE}
+
+ result = await self.api.post_graphql(query, variables)
+ if not result or not result.get("data", {}).get("playlists", {}).get("playlist"):
+ raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found")
+
+ return await parse_playlist(self.provider, result["data"]["playlists"]["playlist"])
+
+ async def get_album_tracks(
+ self,
+ prov_album_id: str,
+ ) -> list[Track]:
+ """Get album tracks for given album id."""
+ query = """
+ query GetAlbum($id: ID!, $imageSize: Int = 512, $first: Int = 50, $after: String) {
+ catalog {
+ album(id: $id) {
+ id
+ tracks(first: $first, after: $after) {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ isrc
+ duration
+ label
+ artist {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ featuredArtists {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ }
+ share
+ genre
+ }
+ pageInfo {
+ hasNextPage
+ endCursor
+ }
+ }
+ }
+ }
+ }
+ """
+ tracks = []
+ variables = {
+ "id": prov_album_id,
+ "imageSize": IMAGE_SIZE,
+ }
+
+ i = 1
+ async for item in self.api.paginate_graphql(
+ query,
+ variables,
+ ["data", "catalog", "album", "tracks"],
+ ):
+ track = await parse_track(self.provider, item)
+ track.position = i
+ tracks.append(track)
+ i += 1
+
+ return tracks
+
+ async def get_playlist_tracks(
+ self,
+ prov_playlist_id: str,
+ page: int = 0,
+ ) -> list[Track]:
+ """Get all playlist tracks for given playlist id."""
+ query = """
+ query getPlaylist($id: ID!, $imageSize: Int = 512, $first: Int = 50, $after: String) {
+ playlists {
+ playlist(id: $id) {
+ id
+ tracks(first: $first, after: $after) {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ isrc
+ duration
+ label
+ artist {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ featuredArtists {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ }
+ share
+ genre
+ }
+ pageInfo {
+ hasNextPage
+ endCursor
+ }
+ }
+ }
+ }
+ }
+ """
+ tracks: list[Track] = []
+
+ if page > 0:
+ # paging not supported, we always return the whole list at once
+ return []
+ # TODO: access the underlying paging on the yousee api (if possible))
+
+ variables = {
+ "id": prov_playlist_id,
+ "imageSize": IMAGE_SIZE,
+ }
+
+ i = 1
+ async for item in self.api.paginate_graphql(
+ query, variables, ["data", "playlists", "playlist", "tracks"]
+ ):
+ track = await parse_track(self.provider, item)
+ track.position = i
+ tracks.append(track)
+ i += 1
+
+ return tracks
+
+ async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
+ """Retrieve a dynamic list of similar tracks based on the provided track."""
+ query = """
+ query similarTracks($id: ID!, $first: Int = 25, $imageSize: Int = 512) {
+ catalog {
+ track(id: $id) {
+ id
+ similarTracks(first: $first) {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ isrc
+ duration
+ label
+ artist {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ featuredArtists {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ }
+ share
+ genre
+ }
+ }
+ }
+ }
+ }
+ """
+
+ variables = {
+ "id": prov_track_id,
+ "first": limit,
+ "imageSize": IMAGE_SIZE,
+ }
+ result = await self.api.post_graphql(query, variables)
+ if not result or not result.get("data", {}).get("catalog", {}).get("track"):
+ raise MediaNotFoundError(f"Track {prov_track_id} not found")
+
+ return [
+ await parse_track(self.provider, item)
+ for item in result["data"]["catalog"]["track"]["similarTracks"]["items"]
+ ]
--- /dev/null
+"""Parsers for YouSee Musik API responses."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from music_assistant_models.enums import AlbumType, ContentType, ExternalID, ImageType
+from music_assistant_models.media_items import (
+ Album,
+ Artist,
+ AudioFormat,
+ MediaItemImage,
+ Playlist,
+ ProviderMapping,
+ Track,
+)
+
+from music_assistant.constants import (
+ VARIOUS_ARTISTS_MBID,
+ VARIOUS_ARTISTS_NAME,
+)
+from music_assistant.helpers.util import infer_album_type, parse_title_and_version, try_parse_int
+from music_assistant.providers.yousee.constants import (
+ CONF_QUALITY,
+ VARIOUS_ARTISTS_ID,
+)
+
+if TYPE_CHECKING:
+ from music_assistant.providers.yousee.api_client import JsonLike
+ from music_assistant.providers.yousee.provider import YouSeeMusikProvider
+
+
+async def parse_track(provider: YouSeeMusikProvider, track_obj: JsonLike) -> Track:
+ """Parse track data from YouSee API response."""
+ track = Track(
+ item_id=track_obj["id"],
+ provider=provider.instance_id,
+ name=track_obj["title"],
+ duration=track_obj.get("duration", 0),
+ provider_mappings={
+ ProviderMapping(
+ item_id=str(track_obj["id"]),
+ provider_domain=provider.domain,
+ provider_instance=provider.instance_id,
+ available=track_obj.get("availableToStream", True),
+ audio_format=AudioFormat(
+ content_type=ContentType.MP4,
+ bit_rate=try_parse_int(provider.config.get_value(CONF_QUALITY)),
+ ),
+ url=track_obj.get("share"),
+ )
+ },
+ )
+
+ if isrc := track_obj.get("isrc"):
+ track.external_ids.add((ExternalID.ISRC, isrc))
+
+ if "artist" in track_obj:
+ artist = parse_artist(provider, track_obj["artist"])
+ track.artists.append(artist)
+
+ for feat_artist_obj in track_obj.get("featuredArtists", {}).get("items", []):
+ feat_artist = parse_artist(provider, feat_artist_obj)
+ track.artists.append(feat_artist)
+
+ if "album" in track_obj:
+ album = await parse_album(provider, track_obj["album"])
+ track.album = album
+
+ if track_genre := track_obj.get("genre"):
+ track.metadata.genres = set(track_genre)
+
+ if track_label := track_obj.get("label"):
+ track.metadata.label = track_label
+
+ if track_obj.get("cover"):
+ track.metadata.add_image(
+ MediaItemImage(
+ type=ImageType.THUMB,
+ path=track_obj["cover"],
+ remotely_accessible=True,
+ provider=provider.instance_id,
+ )
+ )
+
+ return track
+
+
+def parse_artist(provider: YouSeeMusikProvider, artist_obj: JsonLike) -> Artist:
+ """Parse artist data from YouSee API response."""
+ artist = Artist(
+ item_id=artist_obj["id"],
+ provider=provider.instance_id,
+ name=artist_obj["title"],
+ uri=artist_obj.get("share"),
+ provider_mappings={
+ ProviderMapping(
+ item_id=str(artist_obj["id"]),
+ provider_domain=provider.domain,
+ provider_instance=provider.instance_id,
+ )
+ },
+ )
+
+ if artist.item_id == VARIOUS_ARTISTS_ID:
+ artist.mbid = VARIOUS_ARTISTS_MBID
+ artist.name = VARIOUS_ARTISTS_NAME
+
+ if artist_obj.get("cover"):
+ artist.metadata.add_image(
+ MediaItemImage(
+ type=ImageType.THUMB,
+ path=artist_obj["cover"],
+ remotely_accessible=True,
+ provider=provider.instance_id,
+ )
+ )
+
+ return artist
+
+
+async def parse_album(provider: YouSeeMusikProvider, album_obj: JsonLike) -> Album:
+ """Parse album data from YouSee API response."""
+ if "artist" not in album_obj:
+ return await provider.get_album(str(album_obj["id"]))
+
+ name, version = parse_title_and_version(album_obj["title"])
+ album = Album(
+ item_id=album_obj["id"],
+ provider=provider.instance_id,
+ name=name,
+ version=version,
+ provider_mappings={
+ ProviderMapping(
+ item_id=str(album_obj["id"]),
+ provider_domain=provider.domain,
+ provider_instance=provider.instance_id,
+ audio_format=AudioFormat(
+ content_type=ContentType.MP4,
+ bit_rate=try_parse_int(provider.config.get_value(CONF_QUALITY)),
+ ),
+ url=album_obj.get("share"),
+ )
+ },
+ is_playable=album_obj.get("available", True),
+ )
+
+ if album_upc := album_obj.get("upc"):
+ album.external_ids.add((ExternalID.BARCODE, album_upc))
+
+ album.artists.append(parse_artist(provider, album_obj["artist"]))
+
+ for feat_artist_obj in album_obj.get("featuredArtists", {}).get("items", []):
+ feat_artist = parse_artist(provider, feat_artist_obj)
+ album.artists.append(feat_artist)
+
+ if album_genre := album_obj.get("genre"):
+ album.metadata.genres = set(album_genre)
+
+ if album_obj.get("type") == "COMPILATION":
+ album.album_type = AlbumType.COMPILATION
+ elif album_obj.get("type") == "SINGLE":
+ album.album_type = AlbumType.SINGLE
+ elif album_obj.get("type") == "REGULAR":
+ album.album_type = AlbumType.ALBUM
+
+ inferred_type = infer_album_type(name, version)
+ if inferred_type in (AlbumType.SOUNDTRACK, AlbumType.LIVE):
+ album.album_type = inferred_type
+
+ if album_obj.get("cover"):
+ album.metadata.add_image(
+ MediaItemImage(
+ type=ImageType.THUMB,
+ path=album_obj["cover"],
+ remotely_accessible=True,
+ provider=provider.instance_id,
+ )
+ )
+
+ if album_label := album_obj.get("label"):
+ album.metadata.label = album_label
+
+ if album_obj.get("releaseDate"):
+ album.year = try_parse_int(album_obj["releaseDate"][:4])
+
+ return album
+
+
+async def parse_playlist(provider: YouSeeMusikProvider, playlist_obj: JsonLike) -> Playlist:
+ """Parse playlist data from YouSee API response."""
+ playlist = Playlist(
+ item_id=str(playlist_obj["id"]),
+ provider=provider.instance_id,
+ name=playlist_obj["title"],
+ is_editable=playlist_obj["isOwned"],
+ provider_mappings={
+ ProviderMapping(
+ item_id=str(playlist_obj["id"]),
+ provider_domain=provider.domain,
+ provider_instance=provider.instance_id,
+ url=playlist_obj["share"],
+ is_unique=playlist_obj["isOwned"],
+ )
+ },
+ )
+
+ if playlist_obj.get("description"):
+ playlist.metadata.description = playlist_obj["description"]
+
+ if playlist_obj.get("cover"):
+ playlist.metadata.add_image(
+ MediaItemImage(
+ type=ImageType.THUMB,
+ path=playlist_obj["cover"],
+ remotely_accessible=True,
+ provider=provider.instance_id,
+ )
+ )
+
+ return playlist
+
+
+async def parse_lyrics(lyrics: list[JsonLike]) -> tuple[str | None, str | None]:
+ """Parse the YouSee lyrics payload and extract the lyric text in two formats if possible.
+
+ Returns:
+ Tuple[str | None, str | None]: lyrics (plain) and lyrics_lrc, if present.
+ """
+ if not lyrics:
+ return None, None
+
+ plain = ""
+ lrc = ""
+
+ for item in lyrics:
+ line = item.get("line", "")
+ if (start_ms := item.get("startInMs")) is not None:
+ minutes = start_ms // 60000
+ seconds = (start_ms % 60000) // 1000
+ milliseconds = start_ms % 1000
+ lrc += f"[{minutes:02}:{seconds:02}.{milliseconds:02}] {line}\n"
+
+ plain += line + "\n"
+
+ plain = plain.strip()
+ lrc = lrc.strip()
+
+ return plain if plain else None, lrc if lrc else None
--- /dev/null
+"""YouSee Musik playlist manager."""
+
+from typing import TYPE_CHECKING
+
+from music_assistant_models.errors import MediaNotFoundError
+
+from music_assistant.providers.yousee.constants import IMAGE_SIZE
+from music_assistant.providers.yousee.parsers import parse_playlist
+
+if TYPE_CHECKING:
+ from music_assistant_models.media_items import Playlist
+
+ from music_assistant.providers.yousee.provider import YouSeeMusikProvider
+
+
+class YouSeePlaylistManager:
+ """Manages YouSee Musik playlist operations."""
+
+ def __init__(self, provider: "YouSeeMusikProvider"):
+ """Initialize playlist manager."""
+ self.provider = provider
+ self.api = provider.api
+ self.auth = provider.auth
+ self.logger = provider.logger
+
+ async def create(self, name: str) -> "Playlist":
+ """Create a new playlist on provider with given name."""
+ query = """
+ mutation createPlaylist($title: String!, $imageSize: Int = 512) {
+ playlists {
+ create(playlist: {title: $title}) {
+ playlist {
+ id
+ title
+ description
+ tracksCount
+ createdAt
+ isOwned
+ share
+ cover(size: $imageSize)
+ }
+ }
+ }
+ }
+ """
+ variables = {"title": name, "imageSize": IMAGE_SIZE}
+ result = await self.api.post_graphql(query, variables)
+ if not result or not result.get("data", {}).get("playlists", {}).get("create", {}).get(
+ "playlist"
+ ):
+ raise MediaNotFoundError(f"Could not create playlist {name}")
+
+ return await parse_playlist(
+ self.provider, result["data"]["playlists"]["create"]["playlist"]
+ )
+
+ async def add_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
+ """Add track(s) to playlist."""
+ query = """
+ mutation addToLibrary( $id: ID!, $trackIds: [ID]!) {
+ playlists {
+ addTracks(id: $id, duplicatesHandling: SKIP_DUPLICATES, trackIds: $trackIds) {
+ ok
+ }
+ }
+ }
+ """
+ variables = {"id": prov_playlist_id, "trackIds": prov_track_ids}
+ result = await self.api.post_graphql(query, variables)
+
+ if not result or not result.get("data", {}).get("playlists", {}).get("addTracks", {}).get(
+ "ok"
+ ):
+ raise MediaNotFoundError(
+ f"Could not add tracks to playlist {prov_playlist_id}: {prov_track_ids}"
+ )
+
+ async def remove_tracks(
+ self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
+ ) -> None:
+ """Remove track(s) from playlist."""
+ query = """
+ mutation addToLibrary($id: ID!, $mods: [ModifyPlaylistTrackInput!]!) {
+ playlists {
+ modifyTracks(id: $id, modifications: $mods) {
+ ok
+ }
+ }
+ }
+
+ """
+
+ mods = [
+ {"positionFrom": pos - 1, "type": "REMOVE"}
+ for pos in sorted(positions_to_remove, reverse=True)
+ ]
+
+ variables = {"id": prov_playlist_id, "mods": mods}
+
+ result = await self.api.post_graphql(query, variables)
+
+ if not result or not result.get("data", {}).get("playlists", {}).get(
+ "modifyTracks", {}
+ ).get("ok"):
+ raise MediaNotFoundError(
+ f"Could not remove tracks from playlist {prov_playlist_id}: {positions_to_remove}"
+ )
--- /dev/null
+"""YouSee Musik musicprovider support for MusicAssistant."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+from typing import TYPE_CHECKING
+
+from music_assistant_models.errors import (
+ LoginFailed,
+)
+from music_assistant_models.media_items import (
+ Album,
+ Artist,
+ MediaItemType,
+ Playlist,
+ RecommendationFolder,
+ SearchResults,
+ Track,
+)
+
+from music_assistant.constants import (
+ CONF_PASSWORD,
+ CONF_USERNAME,
+)
+from music_assistant.controllers.cache import use_cache
+from music_assistant.models.music_provider import MusicProvider
+from music_assistant.providers.yousee.api_client import YouSeeAPIClient
+from music_assistant.providers.yousee.auth_manager import YouSeeAuthManager
+from music_assistant.providers.yousee.library import YouSeeLibraryManager
+from music_assistant.providers.yousee.media import YouSeeMediaManager
+from music_assistant.providers.yousee.playlist import YouSeePlaylistManager
+from music_assistant.providers.yousee.recommendations import YouSeeRecommendationsManager
+from music_assistant.providers.yousee.streaming import YouSeeStreamingManager
+
+if TYPE_CHECKING:
+ from music_assistant_models.enums import (
+ MediaType,
+ )
+ from music_assistant_models.media_items import (
+ Album,
+ Artist,
+ MediaItemType,
+ Playlist,
+ RecommendationFolder,
+ SearchResults,
+ Track,
+ )
+ from music_assistant_models.streamdetails import StreamDetails
+
+
+class YouSeeMusikProvider(MusicProvider):
+ """Provider implementation for YouSee Musik."""
+
+ auth: YouSeeAuthManager
+
+ async def handle_async_init(self) -> None:
+ """Handle async initialization of the provider."""
+ if not self.config.get_value(CONF_USERNAME) or not self.config.get_value(CONF_PASSWORD):
+ msg = "Invalid login credentials"
+ raise LoginFailed(msg)
+ # try to get a token, raise if that fails
+ self.auth = YouSeeAuthManager(self)
+ self.api = YouSeeAPIClient(self)
+ self.library = YouSeeLibraryManager(self)
+ self.media = YouSeeMediaManager(self)
+ self.playlist = YouSeePlaylistManager(self)
+ self.streaming = YouSeeStreamingManager(self)
+ self.recommendations_manager = YouSeeRecommendationsManager(self)
+
+ token = await self.auth.auth_token()
+ if not token:
+ msg = f"Login failed for user {self.config.get_value(CONF_USERNAME)}"
+ raise LoginFailed(msg)
+
+ async def search(
+ self,
+ search_query: str,
+ media_types: list[MediaType],
+ limit: int = 5,
+ ) -> SearchResults:
+ """Perform search on musicprovider.
+
+ :param search_query: Search query.
+ :param media_types: A list of media_types to include.
+ :param limit: Number of items to return in the search (per type).
+ """
+ return await self.media.search(search_query, media_types, limit)
+
+ async def get_library_artists(self) -> AsyncGenerator[Artist, None]:
+ """Retrieve library artists from the provider."""
+ async for artist in self.library.get_artists():
+ yield artist
+
+ async def get_library_albums(self) -> AsyncGenerator[Album, None]:
+ """Retrieve library albums from the provider."""
+ async for album in self.library.get_albums():
+ yield album
+
+ async def get_library_tracks(self) -> AsyncGenerator[Track, None]:
+ """Retrieve library tracks from the provider."""
+ async for track in self.library.get_tracks():
+ yield track
+
+ async def get_library_playlists(self) -> AsyncGenerator[Playlist, None]:
+ """Retrieve library/subscribed playlists from the provider."""
+ async for playlist in self.library.get_playlists():
+ yield playlist
+
+ @use_cache(3600 * 24 * 30) # Cache for 30 days
+ async def get_artist(self, prov_artist_id: str) -> Artist:
+ """Get full artist details by id."""
+ return await self.media.get_artist(prov_artist_id)
+
+ @use_cache(3600 * 24 * 14) # Cache for 14 days
+ async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
+ """Get a list of all albums for the given artist."""
+ return await self.media.get_artist_albums(prov_artist_id)
+
+ @use_cache(3600 * 24 * 14) # Cache for 14 days
+ async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
+ """Get a list of most popular tracks for the given artist."""
+ return await self.media.get_artist_toptracks(prov_artist_id)
+
+ @use_cache(3600 * 24 * 30) # Cache for 30 days
+ async def get_album(self, prov_album_id: str) -> Album:
+ """Get full album details by id."""
+ return await self.media.get_album(prov_album_id)
+
+ @use_cache(3600 * 24 * 30) # Cache for 30 days
+ async def get_track(self, prov_track_id: str) -> Track:
+ """Get full track details by id."""
+ return await self.media.get_track(prov_track_id)
+
+ @use_cache(3600 * 24 * 30) # Cache for 30 days
+ async def get_playlist(self, prov_playlist_id: str) -> Playlist:
+ """Get full playlist details by id."""
+ return await self.media.get_playlist(prov_playlist_id)
+
+ @use_cache(3600 * 24 * 30) # Cache for 30 days
+ async def get_album_tracks(
+ self,
+ prov_album_id: str,
+ ) -> list[Track]:
+ """Get album tracks for given album id."""
+ return await self.media.get_album_tracks(prov_album_id)
+
+ @use_cache(3600 * 3) # Cache for 3 hours
+ async def get_playlist_tracks(
+ self,
+ prov_playlist_id: str,
+ page: int = 0,
+ ) -> list[Track]:
+ """Get all playlist tracks for given playlist id."""
+ return await self.media.get_playlist_tracks(prov_playlist_id, page)
+
+ async def library_add(self, item: MediaItemType) -> bool:
+ """Add item to provider's library. Return true on success."""
+ return await self.library.add_item(item)
+
+ async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
+ """Remove item from provider's library. Return true on success."""
+ return await self.library.remove_item(prov_item_id, media_type)
+
+ async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
+ """Add track(s) to playlist."""
+ return await self.playlist.add_tracks(prov_playlist_id, prov_track_ids)
+
+ async def remove_playlist_tracks(
+ self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
+ ) -> None:
+ """Remove track(s) from playlist."""
+ return await self.playlist.remove_tracks(prov_playlist_id, positions_to_remove)
+
+ async def create_playlist(self, name: str) -> Playlist:
+ """Create a new playlist on provider with given name."""
+ return await self.playlist.create(name)
+
+ @use_cache(3600 * 24) # Cache for 24 hours
+ async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
+ """Retrieve a dynamic list of similar tracks based on the provided track."""
+ return await self.media.get_similar_tracks(prov_track_id, limit)
+
+ async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
+ """Get streamdetails for a track."""
+ return await self.streaming.get_stream_details(item_id, media_type)
+
+ async def on_streamed(
+ self,
+ streamdetails: StreamDetails,
+ ) -> None:
+ """
+ Handle callback when given streamdetails completed streaming.
+
+ To get the number of seconds streamed, see streamdetails.seconds_streamed.
+ To get the number of seconds seeked/skipped, see streamdetails.seek_position.
+ Note that seconds_streamed is the total streamed seconds, so without seeked time.
+
+ NOTE: Due to internal and player buffering,
+ this may be called in advance of the actual completion.
+ """
+ await self.streaming.report_playback(
+ streamdetails,
+ )
+
+ @use_cache(3600 * 24) # Cache for 1 day
+ async def recommendations(self) -> list[RecommendationFolder]:
+ """
+ Get this provider's recommendations.
+
+ Returns an actual (and often personalised) list of recommendations
+ from this provider for the user/account.
+ """
+ return await self.recommendations_manager.get_recommendations()
--- /dev/null
+"""Recommendation logic for YouSee Musik."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from music_assistant_models.enums import MediaType
+from music_assistant_models.media_items import (
+ RecommendationFolder,
+ UniqueList,
+)
+
+from music_assistant.providers.yousee.constants import IMAGE_SIZE, PAGE_SIZE
+from music_assistant.providers.yousee.parsers import parse_album, parse_track
+
+if TYPE_CHECKING:
+ from music_assistant.providers.yousee.provider import YouSeeMusikProvider
+
+
+class YouSeeRecommendationsManager:
+ """Manages YouSee Musik recommendations."""
+
+ def __init__(self, provider: YouSeeMusikProvider):
+ """Initialize recommendation manager."""
+ self.provider = provider
+ self.api = provider.api
+ self.auth = provider.auth
+ self.logger = provider.logger
+ self.mass = provider.mass
+
+ async def get_recommendations(self) -> list[RecommendationFolder]:
+ """Get recommendations from YouSee Musik."""
+ query = """
+ query Recommendations($imageSize: Int = 512, $first: Int = 50) {
+ me {
+ recommendations {
+ albumRecommendations: recommendation(id: "discoveralbums") {
+ id
+ title
+ subtitle
+ description
+ cover(size: $imageSize)
+ ... on AlbumsRecommendation {
+ albums(first: $first) {
+ items {
+ id
+ title
+ tracksCount
+ genre
+ label
+ releaseDate
+ available
+ upc
+ type
+ share
+ cover(size: $imageSize)
+ artist {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ featuredArtists {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ }
+ }
+ }
+ }
+ }
+ trackRecommendations: recommendation(id: "discovertracks") {
+ ...RecommendationTracks
+ }
+ weeklyDiscoveries: recommendation(id: "weeklyDiscoveries") {
+ ...RecommendationTracks
+ }
+ trackRecommendationsFirstMostPlayed: recommendation(
+ id: "tracksbasedonfirstmostplayedartist"
+ ) {
+ ...RecommendationTracks
+ }
+ trackRecommendationsSecondMostPlayed: recommendation(
+ id: "tracksbasedonSecondmostplayedartist"
+ ) {
+ ...RecommendationTracks
+ }
+ historyTopTracks: recommendation(
+ id: "toptracks"
+ ) {
+ ...RecommendationTracks
+ }
+ historyRecentTracks: recommendation(
+ id: "recenttracks"
+ ) {
+ ...RecommendationTracks
+ }
+ yourmix1: recommendation(
+ id: "yourmix"
+ ) {
+ ...RecommendationTracks
+ }
+ yourmix2: recommendation(
+ id: "yourmix2"
+ ) {
+ ...RecommendationTracks
+ }
+ yourmix3: recommendation(
+ id: "yourmix3"
+ ) {
+ ...RecommendationTracks
+ }
+ }
+ }
+ }
+ fragment RecommendationTracks on Recommendation {
+ id
+ title
+ subtitle
+ description
+ cover(size: $imageSize)
+ ... on TracksRecommendation {
+ tracks(first: $first) {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ isrc
+ duration
+ label
+ artist {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ featuredArtists {
+ items {
+ id
+ title
+ cover(size: $imageSize)
+ }
+ }
+ share
+ genre
+ }
+ }
+ }
+ }
+ """
+
+ variables = {
+ "imageSize": IMAGE_SIZE,
+ "first": PAGE_SIZE,
+ }
+
+ result = await self.api.post_graphql(query, variables)
+
+ if not result or not result.get("data", {}).get("me", {}).get("recommendations"):
+ return []
+
+ recommendations: list[RecommendationFolder] = []
+
+ album_keys = ["albumRecommendations"]
+ track_keys = [
+ "trackRecommendations",
+ "weeklyDiscoveries",
+ "trackRecommendationsFirstMostPlayed",
+ "trackRecommendationsSecondMostPlayed",
+ "historyTopTracks",
+ "historyRecentTracks",
+ "yourmix1",
+ "yourmix2",
+ "yourmix3",
+ ]
+
+ for key in album_keys:
+ rec_data = result["data"]["me"]["recommendations"].get(key)
+ if rec_data:
+ folder = RecommendationFolder(
+ name=rec_data.get("title"),
+ subtitle=rec_data.get("subtitle"),
+ provider=self.provider.instance_id,
+ item_id=rec_data["id"],
+ media_type=MediaType.ALBUM,
+ items=UniqueList(
+ [
+ await parse_album(self.provider, item)
+ for item in rec_data.get("albums", {}).get("items", [])
+ ]
+ ),
+ )
+ recommendations.append(folder)
+ for key in track_keys:
+ rec_data = result["data"]["me"]["recommendations"].get(key)
+ if rec_data:
+ folder = RecommendationFolder(
+ name=rec_data.get("title"),
+ subtitle=rec_data.get("subtitle"),
+ provider=self.provider.instance_id,
+ item_id=rec_data["id"],
+ media_type=MediaType.TRACK,
+ items=UniqueList(
+ [
+ await parse_track(self.provider, item)
+ for item in rec_data.get("tracks", {}).get("items", [])
+ ]
+ ),
+ )
+ recommendations.append(folder)
+
+ return recommendations
--- /dev/null
+"""Streaming operations for YouSee Musik."""
+
+from __future__ import annotations
+
+import re
+from base64 import b64encode
+from typing import TYPE_CHECKING
+
+from music_assistant_models.enums import ContentType, MediaType, StreamType
+from music_assistant_models.errors import MediaNotFoundError, ResourceTemporarilyUnavailable
+from music_assistant_models.media_items import AudioFormat
+from music_assistant_models.streamdetails import StreamDetails
+
+from music_assistant.helpers.datetime import iso_from_utc_timestamp, utc_timestamp
+from music_assistant.providers.yousee.constants import CONF_QUALITY
+
+if TYPE_CHECKING:
+ from music_assistant.providers.yousee.provider import YouSeeMusikProvider
+
+
+class YouSeeStreamingManager:
+ """Manages YouSee Musik streaming operations."""
+
+ def __init__(self, provider: YouSeeMusikProvider):
+ """Initialize streaming manager."""
+ self.provider = provider
+ self.api = provider.api
+ self.mass = provider.mass
+ self.logger = provider.logger
+
+ async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
+ """Get streamdetails for a track."""
+ query = """
+ query playbackFull($id: ID!, $quality: StreamQuality!) {
+ playback(trackId: $id) {
+ full(quality: $quality)
+ }
+ }
+ """
+
+ if media_type != MediaType.TRACK:
+ raise MediaNotFoundError(f"Streaming of media type {media_type} is not supported")
+
+ variables = {
+ "id": item_id,
+ "quality": f"KBPS_{self.provider.config.get_value(CONF_QUALITY)}",
+ }
+
+ result = await self.api.post_graphql(query, variables)
+
+ playback_url = result.get("data", {}).get("playback", {}).get("full")
+ if not playback_url:
+ raise ResourceTemporarilyUnavailable(f"Track {item_id} is not available for streaming")
+
+ matches = re.search(r"mp4-(\d+)kbps", playback_url)
+ returned_playback_quality = int(matches.group(1)) if matches else None
+
+ return StreamDetails(
+ provider=self.provider.instance_id,
+ item_id=item_id,
+ audio_format=AudioFormat(
+ content_type=ContentType.MP4,
+ bit_rate=returned_playback_quality,
+ ),
+ media_type=MediaType.TRACK,
+ stream_type=StreamType.HLS,
+ allow_seek=True,
+ can_seek=True,
+ path=playback_url,
+ data={"start_ts": utc_timestamp()},
+ )
+
+ async def report_playback(
+ self,
+ streamdetails: StreamDetails,
+ ) -> None:
+ """Handle callback when given streamdetails completed streaming."""
+ mutation = """
+ mutation reportPlayback($report: ReportPlaybackInput!) {
+ reportPlayback(report: $report) {
+ ok
+ }
+ }
+ """
+
+ seconds_streamed = min(
+ utc_timestamp() - streamdetails.data["start_ts"],
+ streamdetails.seconds_streamed,
+ )
+
+ variables = {
+ "playbackUrl": streamdetails.path,
+ "playbackContext": b64encode(
+ f"catalog:track;{streamdetails.item_id}".encode()
+ ).decode(),
+ "playedSeconds": int(seconds_streamed),
+ "playedAt": iso_from_utc_timestamp(utc_timestamp()),
+ }
+
+ result = await self.api.post_graphql(mutation, {"report": variables})
+
+ if not result.get("data", {}).get("reportPlayback", {}).get("ok"):
+ self.logger.warning(
+ "Reporting playback for track %s failed with result %s",
+ streamdetails.item_id,
+ result,
+ )
projects / music-assistant-server.git / commitdiff
