1337 lines
52 KiB
Python
1337 lines
52 KiB
Python
"""HTTP API views for My Music Library (proxy to Music Assistant)."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import dataclasses
|
|
import enum
|
|
import logging
|
|
from http import HTTPStatus
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
import aiohttp
|
|
from aiohttp import web
|
|
|
|
from homeassistant.components.http import HomeAssistantView
|
|
from homeassistant.config_entries import ConfigEntryState
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.helpers import entity_registry as er
|
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|
|
|
from . import outputs
|
|
from .const import CONF_MA_URL, DOMAIN
|
|
from .entity import UNIQUE_ID_PREFIX
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
# ── Serialization ────────────────────────────────────────────────────────────
|
|
|
|
def _to_json_safe(obj: Any) -> Any:
|
|
"""Recursively convert MA model objects to JSON-serialisable dicts."""
|
|
if obj is None or isinstance(obj, (bool, int, float, str)):
|
|
return obj
|
|
if isinstance(obj, enum.Enum):
|
|
return obj.value
|
|
if dataclasses.is_dataclass(obj) and not isinstance(obj, type):
|
|
d = {f.name: _to_json_safe(getattr(obj, f.name)) for f in dataclasses.fields(obj)}
|
|
if "image" not in d:
|
|
img = getattr(obj, "image", None)
|
|
if img is not None:
|
|
d["image"] = _to_json_safe(img)
|
|
return d
|
|
if isinstance(obj, dict):
|
|
return {k: _to_json_safe(v) for k, v in obj.items()}
|
|
if isinstance(obj, (list, tuple, set, frozenset)):
|
|
return [_to_json_safe(i) for i in obj]
|
|
return str(obj)
|
|
|
|
|
|
def _extract_thumbnail(item: dict) -> str:
|
|
"""Extract thumbnail URL from a serialised MA item dict.
|
|
|
|
Looks in multiple locations where MA stores image paths,
|
|
returns the first non-empty path found.
|
|
"""
|
|
thumb = item.get("thumbnail") or ""
|
|
if thumb:
|
|
return thumb
|
|
|
|
image = item.get("image")
|
|
if isinstance(image, dict):
|
|
thumb = image.get("path") or image.get("url") or ""
|
|
elif isinstance(image, str) and image:
|
|
thumb = image
|
|
if thumb:
|
|
return thumb
|
|
|
|
metadata = item.get("metadata") or {}
|
|
for img in metadata.get("images") or []:
|
|
path = img.get("path", "") if isinstance(img, dict) else ""
|
|
if path:
|
|
return path
|
|
|
|
return ""
|
|
|
|
|
|
def _make_thumb_url(raw_path: str) -> str:
|
|
"""Wrap a raw image path in our thumbnail proxy endpoint.
|
|
|
|
Instead of exposing raw image URLs (which may require provider auth
|
|
or be unreachable from the browser), route everything through our
|
|
server-side proxy that resolves via the MA server.
|
|
"""
|
|
if not raw_path:
|
|
return ""
|
|
return f"/my_music_library/thumb?path={quote(raw_path, safe='')}"
|
|
|
|
|
|
def _serialize_search_results(results: Any) -> dict:
|
|
"""Convert a MA SearchResults object (or dict) to a plain dict."""
|
|
if results is None:
|
|
return {"tracks": [], "artists": [], "albums": [], "playlists": []}
|
|
safe = _to_json_safe(results)
|
|
if isinstance(safe, dict):
|
|
for key in ("tracks", "artists", "albums", "playlists", "radios"):
|
|
for item in safe.get(key) or []:
|
|
if isinstance(item, dict):
|
|
raw = item.get("thumbnail") or _extract_thumbnail(item)
|
|
item["thumbnail"] = _make_thumb_url(raw)
|
|
return safe
|
|
return {"tracks": [], "artists": [], "albums": [], "playlists": [], "raw": str(safe)}
|
|
|
|
|
|
# ── MA client / URL resolution ────────────────────────────────────────────────
|
|
#
|
|
# my_music_library owns its Music Assistant connection (see mass_connection.py) —
|
|
# there is exactly one config entry, and it only reaches ConfigEntryState.LOADED
|
|
# once that connection succeeded. No more scanning other integrations' domains.
|
|
|
|
|
|
def _get_mass_client(hass: HomeAssistant) -> Any | None:
|
|
"""Return the native MusicAssistantClient for our config entry, if loaded."""
|
|
for entry in hass.config_entries.async_entries(DOMAIN):
|
|
if entry.state == ConfigEntryState.LOADED:
|
|
return entry.runtime_data.mass
|
|
_LOGGER.warning(
|
|
"My Music Library config entry is not loaded — check its connection "
|
|
"to the Music Assistant server (filter: my_music_library)."
|
|
)
|
|
return None
|
|
|
|
|
|
def _get_mass_url(hass: HomeAssistant) -> str | None:
|
|
"""Return the configured Music Assistant server URL."""
|
|
for entry in hass.config_entries.async_entries(DOMAIN):
|
|
if url := entry.data.get(CONF_MA_URL):
|
|
return url.rstrip("/")
|
|
return None
|
|
|
|
|
|
# ── Search ────────────────────────────────────────────────────────────────────
|
|
|
|
async def _search_via_ma_client(
|
|
hass: HomeAssistant,
|
|
query: str,
|
|
limit: int,
|
|
*,
|
|
library_only: bool = False,
|
|
) -> dict | None:
|
|
"""Search Music Assistant via the native client."""
|
|
mass = _get_mass_client(hass)
|
|
if mass is None:
|
|
return None
|
|
try:
|
|
results = await mass.music.search(
|
|
search_query=query, limit=limit, library_only=library_only
|
|
)
|
|
return _serialize_search_results(results)
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.warning("MA search(%r) failed: %s", query, err)
|
|
return None
|
|
|
|
|
|
# ── Library ───────────────────────────────────────────────────────────────────
|
|
|
|
def _normalize_library_item(item: dict) -> dict:
|
|
"""Map MA Python client fields to the browse_media-compatible fields the card expects."""
|
|
title = item.get("name") or item.get("title") or ""
|
|
uri = item.get("uri") or item.get("media_content_id") or ""
|
|
|
|
media_type = item.get("media_type", "")
|
|
if isinstance(media_type, dict):
|
|
media_type = media_type.get("value", "")
|
|
|
|
thumbnail = _make_thumb_url(_extract_thumbnail(item))
|
|
|
|
artist = item.get("media_artist") or ""
|
|
if not artist:
|
|
artists = item.get("artists") or []
|
|
if artists and isinstance(artists[0], dict):
|
|
artist = artists[0].get("name", "")
|
|
|
|
album_type = item.get("album_type", "album")
|
|
if isinstance(album_type, dict):
|
|
album_type = album_type.get("value", "album")
|
|
|
|
track_number = item.get("track_number") or item.get("position") or 0
|
|
duration = item.get("duration") or 0
|
|
|
|
provider_mappings = item.get("provider_mappings") or []
|
|
providers: set[str] = set()
|
|
provider_instances: set[str] = set()
|
|
for m in provider_mappings:
|
|
if isinstance(m, dict):
|
|
dom = m.get("provider_domain", "")
|
|
inst = (
|
|
m.get("provider_instance_id_or_domain")
|
|
or m.get("provider_instance")
|
|
or m.get("provider_instance_id")
|
|
or dom
|
|
)
|
|
if dom:
|
|
providers.add(dom)
|
|
if inst:
|
|
provider_instances.add(inst)
|
|
elif isinstance(m, str):
|
|
provider_instances.add(m)
|
|
else:
|
|
dom = str(getattr(m, "provider_domain", "") or "")
|
|
inst = str(
|
|
getattr(m, "provider_instance", "")
|
|
or getattr(m, "provider_instance_id_or_domain", "")
|
|
or dom
|
|
)
|
|
if dom:
|
|
providers.add(dom)
|
|
if inst:
|
|
provider_instances.add(inst)
|
|
provider_instances.discard("")
|
|
provider_instances.discard("builtin")
|
|
providers.discard("")
|
|
providers.discard("builtin")
|
|
return {
|
|
"title": title,
|
|
"media_content_id": uri,
|
|
"media_content_type": media_type,
|
|
"thumbnail": thumbnail,
|
|
"media_artist": artist,
|
|
"album_type": str(album_type).lower() if album_type else "album",
|
|
"track_number": int(track_number) if track_number else 0,
|
|
"duration": float(duration) if duration else 0,
|
|
"providers": sorted(providers),
|
|
"provider_instances": sorted(provider_instances),
|
|
}
|
|
|
|
|
|
_LIBRARY_METHODS: dict[str, str] = {
|
|
"artists": "get_library_artists",
|
|
"albums": "get_library_albums",
|
|
"tracks": "get_library_tracks",
|
|
"playlists": "get_library_playlists",
|
|
"radios": "get_library_radios",
|
|
}
|
|
|
|
|
|
async def _get_library_via_ma_client(
|
|
hass: HomeAssistant,
|
|
media_type: str,
|
|
limit: int,
|
|
favorite: bool,
|
|
offset: int = 0,
|
|
provider_instance: str | None = None,
|
|
) -> list | None:
|
|
"""Fetch library items via the native MA client."""
|
|
mass = _get_mass_client(hass)
|
|
if mass is None:
|
|
return None
|
|
|
|
method_name = _LIBRARY_METHODS.get(media_type)
|
|
if method_name is None:
|
|
return None
|
|
fn = getattr(mass.music, method_name)
|
|
|
|
try:
|
|
items = await fn(favorite=favorite, limit=limit, offset=offset, provider=provider_instance)
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.warning("Library fetch for %s failed: %s", media_type, err)
|
|
return None
|
|
|
|
normalized = [_normalize_library_item(_to_json_safe(i)) for i in items[:limit]]
|
|
_LOGGER.debug("Library %s: %d items", media_type, len(normalized))
|
|
return normalized
|
|
|
|
|
|
# ── Recommendations ──────────────────────────────────────────────────────────
|
|
|
|
def _normalize_recommendation_item(item: dict) -> dict:
|
|
"""Normalize a single item from a MA recommendation folder."""
|
|
media_type = item.get("media_type", "")
|
|
if isinstance(media_type, dict):
|
|
media_type = media_type.get("value", "")
|
|
media_type = str(media_type).lower() if media_type else ""
|
|
|
|
if media_type in ("track", "album", "artist", "playlist", "radio"):
|
|
return _normalize_library_item(item)
|
|
return _normalize_browse_item(item)
|
|
|
|
|
|
async def _get_recommendation_folder_items(
|
|
mass: Any,
|
|
provider: str,
|
|
item_id: str,
|
|
) -> list[Any]:
|
|
"""Fetch the concrete items of one recommendation folder.
|
|
|
|
``music.recommendations()`` only returns the folder metadata (name,
|
|
provider, item_id) — MA's API works in two steps, and the actual content
|
|
of a folder requires a second call to "music/recommendations/items"
|
|
(server-side: RecommendationsController.get_recommendation_items).
|
|
There is no Python-client wrapper for it yet, so we call it directly via
|
|
``send_command``.
|
|
"""
|
|
result = await asyncio.wait_for(
|
|
mass.send_command("music/recommendations/items", provider=provider, item_id=item_id),
|
|
timeout=10,
|
|
)
|
|
return list(result) if not isinstance(result, list) else result
|
|
|
|
|
|
async def _get_recommendations_via_ma_client(
|
|
hass: HomeAssistant,
|
|
) -> list[dict] | None:
|
|
"""Fetch recommendations via the native MA client."""
|
|
mass = _get_mass_client(hass)
|
|
if mass is None:
|
|
return None
|
|
|
|
try:
|
|
folders = await asyncio.wait_for(mass.music.recommendations(), timeout=10)
|
|
except asyncio.TimeoutError:
|
|
_LOGGER.warning("Recommendations fetch timed out")
|
|
return None
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.warning("Recommendations fetch failed: %s", err)
|
|
return None
|
|
|
|
_LOGGER.debug("Recommendations: %d folders", len(folders))
|
|
safe_folders = [_to_json_safe(folder) for folder in folders]
|
|
|
|
# A folder from music.recommendations() only carries metadata —
|
|
# its actual content needs one extra call per folder. Fetch them
|
|
# all in parallel, isolating failures so one slow/broken tiroir
|
|
# doesn't blank the others.
|
|
fetch_tasks = [
|
|
_get_recommendation_folder_items(
|
|
mass, str(f.get("provider") or ""), str(f.get("item_id") or ""),
|
|
)
|
|
for f in safe_folders
|
|
]
|
|
items_results = await asyncio.gather(*fetch_tasks, return_exceptions=True)
|
|
|
|
normalized: list[dict] = []
|
|
for f, items_result in zip(safe_folders, items_results):
|
|
try:
|
|
if isinstance(items_result, BaseException):
|
|
_LOGGER.warning(
|
|
"Recommendations: items fetch failed for folder %r (provider=%s, item_id=%s): %s",
|
|
f.get("name"), f.get("provider"), f.get("item_id"), items_result,
|
|
)
|
|
items_raw = f.get("items") or []
|
|
else:
|
|
items_raw = items_result or f.get("items") or []
|
|
|
|
folder_domain = str(f.get("provider_domain") or f.get("provider", "") or "")
|
|
folder_instance = str(
|
|
f.get("provider_instance_id_or_domain")
|
|
or f.get("provider_instance")
|
|
or f.get("provider_instance_id")
|
|
or folder_domain
|
|
)
|
|
is_library_folder = folder_domain in ("library", "builtin", "")
|
|
items_norm = []
|
|
for i in items_raw:
|
|
i_safe = _to_json_safe(i)
|
|
if is_library_folder:
|
|
pm = i_safe.get("provider_mappings") or []
|
|
if pm:
|
|
i_safe["provider_mappings"] = [
|
|
m for m in pm
|
|
if (m.get("in_library") is True if isinstance(m, dict) else getattr(m, "in_library", True))
|
|
] or pm
|
|
item = _normalize_recommendation_item(i_safe)
|
|
if not is_library_folder:
|
|
if folder_instance and not item.get("provider_instances"):
|
|
item["provider_instances"] = [folder_instance]
|
|
if folder_domain and not item.get("providers"):
|
|
item["providers"] = [folder_domain]
|
|
items_norm.append(item)
|
|
|
|
# Infer folder provider from items when MA doesn't provide it
|
|
inferred_domain = folder_domain
|
|
inferred_instance = folder_instance
|
|
if is_library_folder and items_norm:
|
|
instance_counts: dict[str, int] = {}
|
|
for itm in items_norm:
|
|
for pi in itm.get("provider_instances") or []:
|
|
if pi and pi not in ("builtin", "library"):
|
|
instance_counts[pi] = instance_counts.get(pi, 0) + 1
|
|
if instance_counts:
|
|
top_inst = max(instance_counts, key=instance_counts.get) # type: ignore[arg-type]
|
|
if instance_counts[top_inst] == len(items_norm):
|
|
inferred_instance = top_inst
|
|
inferred_domain = top_inst.split("--")[0] if "--" in top_inst else top_inst
|
|
|
|
normalized.append({
|
|
"folder_id": f.get("item_id") or f.get("path") or "",
|
|
"name": f.get("name") or f.get("label") or "",
|
|
"icon": f.get("icon") or "",
|
|
"provider_domain": inferred_domain,
|
|
"provider_instance": inferred_instance,
|
|
"items": items_norm,
|
|
})
|
|
except Exception as folder_err: # noqa: BLE001
|
|
_LOGGER.warning(
|
|
"Recommendations: skipping folder %r due to processing error: %s",
|
|
f.get("name"), folder_err,
|
|
)
|
|
continue
|
|
return normalized
|
|
|
|
|
|
# ── Browse ────────────────────────────────────────────────────────────────────
|
|
|
|
def _normalize_browse_item(item: dict) -> dict:
|
|
"""Normalise a MA browse result item (BrowseFolder or MediaItem) for the card."""
|
|
media_type = item.get("media_type", "")
|
|
if isinstance(media_type, dict):
|
|
media_type = media_type.get("value", "")
|
|
media_type = str(media_type).lower() if media_type else ""
|
|
|
|
is_folder = (
|
|
"path" in item and media_type in ("", "folder", "directory")
|
|
) or media_type in ("folder", "directory")
|
|
|
|
title = (
|
|
item.get("name") or item.get("display_name") or item.get("label")
|
|
or item.get("title") or ""
|
|
)
|
|
if not title:
|
|
title = item.get("translation_key") or item.get("item_id") or ""
|
|
if title:
|
|
title = title.replace("_", " ").capitalize()
|
|
# For BrowseFolder items, prefer `path` (the navigation path MA expects)
|
|
# over `uri` (an auto-generated semantic identifier).
|
|
uri = item.get("path") or item.get("uri") or item.get("media_content_id") or ""
|
|
|
|
# MA sometimes returns URIs like "provider://folder/Deftones" where "folder/"
|
|
# is an internal prefix, not a real directory. Strip it so navigation works.
|
|
if "://" in uri:
|
|
_scheme, _path = uri.split("://", 1)
|
|
if _path.startswith("folder/"):
|
|
uri = f"{_scheme}://{_path[len('folder/'):]}"
|
|
|
|
thumbnail = _make_thumb_url(_extract_thumbnail(item))
|
|
|
|
artist = item.get("media_artist") or ""
|
|
if not artist:
|
|
artists = item.get("artists") or []
|
|
if artists and isinstance(artists[0], dict):
|
|
artist = artists[0].get("name", "")
|
|
|
|
duration = item.get("duration") or 0
|
|
|
|
return {
|
|
"title": title,
|
|
"uri": uri,
|
|
"media_content_type": media_type or ("folder" if is_folder else "music"),
|
|
"thumbnail": thumbnail,
|
|
"subtitle": artist,
|
|
"duration": float(duration) if duration else 0,
|
|
"is_folder": is_folder,
|
|
}
|
|
|
|
|
|
def _is_ma_back_item(item: dict) -> bool:
|
|
"""Return True for MA virtual 'back' navigation items."""
|
|
name = item.get("title", "") or item.get("name", "")
|
|
if name == "..":
|
|
return True
|
|
uri = item.get("uri", "")
|
|
if "://" in uri:
|
|
tail = uri.split("://", 1)[1].lower().rstrip("/")
|
|
if tail in ("back", "..", "root"):
|
|
return True
|
|
return False
|
|
|
|
|
|
async def _browse_via_ma_client(
|
|
hass: HomeAssistant,
|
|
uri: str | None,
|
|
limit: int = 200,
|
|
) -> list | None:
|
|
"""Browse a MA path via the native MA client."""
|
|
mass = _get_mass_client(hass)
|
|
if mass is None:
|
|
return None
|
|
|
|
try:
|
|
# Raw command, not mass.music.browse(): the client turns every browse
|
|
# entry without provider_mappings (all folders) into an ItemMapping,
|
|
# which drops BrowseFolder.path — the navigation path some providers
|
|
# set explicitly (e.g. Deezer "…://Made For You" vs item_id
|
|
# "made_for_me"). The raw dicts keep it.
|
|
items = await mass.send_command("music/browse", path=uri)
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.warning("browse(%r) failed: %s", uri, err)
|
|
return None
|
|
|
|
normalized = []
|
|
for raw_item in items:
|
|
safe = _to_json_safe(raw_item)
|
|
n = _normalize_browse_item(safe)
|
|
if _is_ma_back_item(n):
|
|
n["is_back"] = True
|
|
elif n.get("is_folder") and uri:
|
|
n["uri"] = _browse_folder_uri(safe, uri, n["uri"])
|
|
normalized.append(n)
|
|
return normalized[:limit]
|
|
|
|
|
|
def _browse_folder_uri(raw: dict, parent_uri: str, normalized_uri: str) -> str:
|
|
"""Return the navigation URI of a browse sub-folder.
|
|
|
|
MA's `path` is the navigation path. When a provider sets it explicitly,
|
|
use it as is: Deezer's "Made For You" folder has `path` "…://Made For You"
|
|
but `item_id` "made_for_me", so rebuilding from `item_id` breaks it (502).
|
|
When a provider doesn't set it, `BrowseFolder` defaults `path` to
|
|
`<provider>://<item_id>`, which loses the parent folder: only then rebuild
|
|
it from the parent URI — without prepending the parent when `item_id` is
|
|
already relative to the provider root, as for the filesystem provider
|
|
("Disco/ABBA Gold" must not become "Disco/Disco/ABBA Gold", issue #17).
|
|
"""
|
|
item_id = str(raw.get("item_id") or "")
|
|
path = str(raw.get("path") or "")
|
|
if path and path != f"{raw.get('provider') or ''}://{item_id}":
|
|
return normalized_uri
|
|
if not item_id or "://" not in parent_uri:
|
|
return normalized_uri
|
|
scheme, parent_sub = parent_uri.split("://", 1)
|
|
parent_sub = parent_sub.rstrip("/")
|
|
if not parent_sub or item_id == parent_sub or item_id.startswith(f"{parent_sub}/"):
|
|
return f"{scheme}://{item_id}"
|
|
return f"{scheme}://{parent_sub}/{item_id}"
|
|
|
|
|
|
# ── Subitems (artist albums, album tracks, playlist tracks) ───────────────────
|
|
|
|
_SUBITEM_METHODS: dict[str, str] = {
|
|
"artist_albums": "get_artist_albums",
|
|
"album_tracks": "get_album_tracks",
|
|
"playlist_tracks": "get_playlist_tracks",
|
|
}
|
|
|
|
|
|
def _parse_ma_uri(uri: str) -> tuple[str, str]:
|
|
"""Return (item_id, provider_domain) from a MA URI string.
|
|
|
|
Examples
|
|
--------
|
|
``spotify://artist/12345`` → ("12345", "spotify")
|
|
``music_assistant://abc`` → ("abc", "music_assistant")
|
|
``qobuz://album/xyz/789`` → ("789", "qobuz")
|
|
"""
|
|
if "://" not in uri:
|
|
return uri, "library"
|
|
scheme, rest = uri.split("://", 1)
|
|
parts = [p for p in rest.split("/") if p]
|
|
item_id = parts[-1] if parts else rest
|
|
return item_id, scheme
|
|
|
|
|
|
async def _get_subitems(
|
|
hass: HomeAssistant,
|
|
action: str,
|
|
uri: str,
|
|
limit: int = 50,
|
|
) -> list | None:
|
|
"""Fetch sub-items of a MA library item via the native MA client."""
|
|
mass = _get_mass_client(hass)
|
|
if mass is None:
|
|
return None
|
|
|
|
method_name = _SUBITEM_METHODS.get(action)
|
|
if method_name is None:
|
|
return None
|
|
fn = getattr(mass.music, method_name)
|
|
|
|
item_id, provider = _parse_ma_uri(uri)
|
|
if action == "artist_albums" and provider != "library":
|
|
# A provider URI (search result, track played from a streaming provider) would
|
|
# return that provider's whole catalog: this action is library-scoped, so use the
|
|
# library version of the artist (MA returns it when the artist is in the library).
|
|
try:
|
|
artist = await mass.music.get_item_by_uri(uri)
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.warning("Subitems artist_albums: could not resolve %r: %s", uri, err)
|
|
return None
|
|
if artist.provider != "library":
|
|
return []
|
|
item_id, provider = artist.item_id, "library"
|
|
try:
|
|
items = await fn(item_id, provider)
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.warning("Subitems %s(%r, %r) failed: %s", method_name, item_id, provider, err)
|
|
return None
|
|
|
|
return [_normalize_library_item(_to_json_safe(i)) for i in items[:limit]]
|
|
|
|
|
|
async def _get_artist_all_albums(
|
|
hass: HomeAssistant,
|
|
uri: str,
|
|
limit: int = 200,
|
|
) -> list | None:
|
|
"""Fetch an artist's full album catalog (not just library/favorites).
|
|
|
|
``artist_albums`` (above) resolves the URI's own scheme for its provider,
|
|
which for a ``library://...`` URI is the internal "library" pseudo-provider
|
|
— that can only ever return what's already synced (i.e. favorites), it has
|
|
no external catalog behind it. To get everything, resolve the artist's real
|
|
provider mapping(s) and query those directly with ``in_library_only=False``.
|
|
"""
|
|
mass = _get_mass_client(hass)
|
|
if mass is None:
|
|
return None
|
|
|
|
try:
|
|
artist = await mass.music.get_item_by_uri(uri)
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.warning("artist_albums_all: could not resolve artist %r: %s", uri, err)
|
|
return None
|
|
|
|
provider_mappings = list(getattr(artist, "provider_mappings", None) or [])
|
|
if not provider_mappings:
|
|
# Fallback: whatever provider the URI itself points to.
|
|
item_id, provider = _parse_ma_uri(uri)
|
|
provider_mappings = [
|
|
SimpleNamespace(item_id=item_id, provider_instance=provider, provider_domain=provider)
|
|
]
|
|
|
|
candidates = [m for m in provider_mappings if m.provider_domain not in ("library", "builtin")] or provider_mappings
|
|
|
|
seen_instances: set[str] = set()
|
|
all_albums: list = []
|
|
for mapping in candidates:
|
|
instance = mapping.provider_instance or mapping.provider_domain
|
|
if instance in seen_instances:
|
|
continue
|
|
seen_instances.add(instance)
|
|
try:
|
|
albums = await mass.music.get_artist_albums(mapping.item_id, instance, in_library_only=False)
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.debug("artist_albums_all: provider %s failed: %s", instance, err)
|
|
continue
|
|
all_albums.extend(albums)
|
|
|
|
# Multiple providers can return the same album — dedupe by (name, year).
|
|
deduped: dict[tuple[str, int | None], Any] = {}
|
|
for album in all_albums:
|
|
key = (album.name.strip().lower(), getattr(album, "year", None))
|
|
deduped.setdefault(key, album)
|
|
|
|
normalized = []
|
|
for album in deduped.values():
|
|
item = _normalize_library_item(_to_json_safe(album))
|
|
item["year"] = getattr(album, "year", None)
|
|
normalized.append(item)
|
|
return normalized[:limit]
|
|
|
|
|
|
async def _get_track_artist(hass: HomeAssistant, uri: str) -> list | None:
|
|
"""Return the main artist of a track, as a one-item list (empty if it has none).
|
|
|
|
Used by the player tab's "artist" button: the media_player entity only
|
|
exposes the artist *name*, the card needs a URI to open the artist page.
|
|
Re-fetching the artist by its URI returns the library version when the
|
|
artist is in the library (MA prefers it), so the page's favorites section
|
|
keeps working.
|
|
"""
|
|
mass = _get_mass_client(hass)
|
|
if mass is None:
|
|
return None
|
|
|
|
try:
|
|
track = await mass.music.get_item_by_uri(uri)
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.warning("track_artist: could not resolve %r: %s", uri, err)
|
|
return None
|
|
|
|
artists = list(getattr(track, "artists", None) or [])
|
|
if not artists:
|
|
return []
|
|
artist = artists[0]
|
|
try:
|
|
artist = await mass.music.get_item_by_uri(artist.uri)
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.debug("track_artist: keeping the track's artist mapping for %r: %s", uri, err)
|
|
return [_normalize_library_item(_to_json_safe(artist))]
|
|
|
|
|
|
# ── Providers ─────────────────────────────────────────────────────────────────
|
|
|
|
async def _get_providers_via_ma_client(hass: HomeAssistant) -> list | None:
|
|
"""Return available MA providers (domain + name) via the MA Python client.
|
|
|
|
mass.providers returns list[ProviderInstance] in music_assistant_client.
|
|
We access attributes directly to avoid _to_json_safe conversion issues
|
|
(e.g. StrEnum not being a dataclass in certain model versions).
|
|
"""
|
|
mass = _get_mass_client(hass)
|
|
if mass is None:
|
|
return None
|
|
|
|
try:
|
|
raw = getattr(mass, "providers", None)
|
|
if raw is None:
|
|
_LOGGER.debug("mass.providers is None — MA provider list unavailable")
|
|
return None
|
|
|
|
# mass.providers is list[ProviderInstance]; guard against dict-like wrappers too
|
|
items: Any = raw.values() if (isinstance(raw, dict) or hasattr(raw, "items")) else raw
|
|
|
|
seen: set[str] = set()
|
|
result = []
|
|
for p in items:
|
|
try:
|
|
# Access attributes directly — avoids _to_json_safe conversion issues
|
|
provider_type = str(getattr(p, "type", "") or "").lower()
|
|
if provider_type and provider_type != "music":
|
|
continue
|
|
domain = str(getattr(p, "domain", "") or "")
|
|
if domain == "builtin":
|
|
continue
|
|
instance_id = str(getattr(p, "instance_id", "") or domain)
|
|
name = str(getattr(p, "name", "") or domain)
|
|
available = bool(getattr(p, "available", True))
|
|
if not instance_id or instance_id in seen:
|
|
continue
|
|
seen.add(instance_id)
|
|
_LOGGER.debug("MA provider: instance_id=%r domain=%r name=%r available=%s", instance_id, domain, name, available)
|
|
result.append({
|
|
"domain": domain,
|
|
"instance_id": instance_id,
|
|
"name": name,
|
|
"available": available,
|
|
})
|
|
except Exception as item_err: # noqa: BLE001
|
|
_LOGGER.debug("Skipping MA provider item: %s", item_err)
|
|
continue
|
|
|
|
_LOGGER.debug(
|
|
"MA providers: %d total, %d available",
|
|
len(result), sum(1 for r in result if r["available"]),
|
|
)
|
|
return [r for r in result if r["available"]]
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.warning("Failed to get MA providers: %s", err)
|
|
return None
|
|
|
|
|
|
# ── MA queue helpers ──────────────────────────────────────────────────────────
|
|
|
|
def _resolve_ma_player_id(hass: HomeAssistant, entity_id: str) -> str | None:
|
|
"""Map a HA media_player entity_id to its Music Assistant player_id.
|
|
|
|
Our media_player entities use unique_id = f"{UNIQUE_ID_PREFIX}{player_id}"
|
|
(see entity.py).
|
|
"""
|
|
ent_reg = er.async_get(hass)
|
|
entry = ent_reg.async_get(entity_id)
|
|
if entry is None or not entry.unique_id:
|
|
return None
|
|
return entry.unique_id.removeprefix(UNIQUE_ID_PREFIX)
|
|
|
|
|
|
async def _resolve_queue_id(hass: HomeAssistant, entity_id: str) -> tuple[Any | None, str | None]:
|
|
"""Return (mass_client, queue_id) for a HA media_player entity."""
|
|
mass = _get_mass_client(hass)
|
|
if mass is None:
|
|
return None, None
|
|
|
|
ma_player_id = _resolve_ma_player_id(hass, entity_id)
|
|
if not ma_player_id:
|
|
return mass, None
|
|
|
|
queue = await mass.player_queues.get_active_queue(ma_player_id)
|
|
queue_id = queue.queue_id if queue is not None else ma_player_id
|
|
return mass, queue_id
|
|
|
|
|
|
def _normalize_queue_item(item: dict) -> dict:
|
|
"""Normalize a MA QueueItem dict for the frontend."""
|
|
name = item.get("name") or ""
|
|
queue_item_id = item.get("queue_item_id") or ""
|
|
duration = item.get("duration") or 0
|
|
|
|
raw_thumb = _extract_thumbnail(item)
|
|
if not raw_thumb:
|
|
raw_thumb = _extract_thumbnail(item.get("media_item") or {})
|
|
thumbnail = _make_thumb_url(raw_thumb)
|
|
|
|
media_item = item.get("media_item") or {}
|
|
uri = media_item.get("uri") or ""
|
|
artist = ""
|
|
artists = media_item.get("artists") or []
|
|
if artists and isinstance(artists[0], dict):
|
|
artist = artists[0].get("name", "")
|
|
|
|
media_type = media_item.get("media_type", "track")
|
|
if isinstance(media_type, dict):
|
|
media_type = media_type.get("value", "track")
|
|
|
|
return {
|
|
"queue_item_id": queue_item_id,
|
|
"title": name or media_item.get("name", ""),
|
|
"media_content_id": uri,
|
|
"media_content_type": str(media_type),
|
|
"media_artist": artist,
|
|
"duration": float(duration) if duration else 0,
|
|
"thumbnail": thumbnail,
|
|
}
|
|
|
|
|
|
async def _queue_jump_via_ma_client(
|
|
hass: HomeAssistant, entity_id: str, index: int,
|
|
) -> bool:
|
|
"""Ask MA to jump to ``index`` within the player's existing queue."""
|
|
mass, queue_id = await _resolve_queue_id(hass, entity_id)
|
|
if mass is None or queue_id is None:
|
|
_LOGGER.warning("queue_jump: cannot resolve queue for %s", entity_id)
|
|
return False
|
|
|
|
try:
|
|
await mass.player_queues.play_index(queue_id, index)
|
|
_LOGGER.debug("queue_jump: queue=%s index=%d ok", queue_id, index)
|
|
return True
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.warning("queue_jump: play_index(%s, %d) failed: %s", queue_id, index, err)
|
|
return False
|
|
|
|
|
|
# ── HA Views ──────────────────────────────────────────────────────────────────
|
|
|
|
class MusicAssistantSearchView(HomeAssistantView):
|
|
"""Proxy search requests to Music Assistant.
|
|
|
|
GET /my_music_library/search?query=<q>&limit=<n>&library_only=true
|
|
"""
|
|
|
|
url = "/my_music_library/search"
|
|
name = "my_music_library:search"
|
|
requires_auth = True
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
"""Handle search request."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
|
|
query = request.query.get("query", "").strip()
|
|
if not query:
|
|
return self.json_message("Missing 'query' parameter.", HTTPStatus.BAD_REQUEST)
|
|
|
|
limit = min(int(request.query.get("limit", 25)), 100)
|
|
library_only = request.query.get("library_only", "").lower() in ("1", "true", "yes")
|
|
_LOGGER.debug("Search request: query=%r limit=%d library_only=%s", query, limit, library_only)
|
|
|
|
result = await _search_via_ma_client(hass, query, limit, library_only=library_only)
|
|
if result is not None:
|
|
return web.json_response(result)
|
|
|
|
_LOGGER.error("Search failed for query=%r", query)
|
|
return self.json_message(
|
|
"Could not search Music Assistant. Check HA logs (filter: my_music_library).",
|
|
HTTPStatus.BAD_GATEWAY,
|
|
)
|
|
|
|
|
|
class MusicAssistantLibraryView(HomeAssistantView):
|
|
"""Return library items from Music Assistant.
|
|
|
|
GET /my_music_library/library?type=artists|albums|tracks|playlists&limit=25&favorite=true
|
|
"""
|
|
|
|
url = "/my_music_library/library"
|
|
name = "my_music_library:library"
|
|
requires_auth = True
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
"""Handle library request."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
|
|
media_type = request.query.get("type", "").strip()
|
|
if media_type not in ("artists", "albums", "tracks", "playlists", "radios"):
|
|
return self.json_message("Invalid 'type' parameter.", HTTPStatus.BAD_REQUEST)
|
|
|
|
limit = min(int(request.query.get("limit", 25)), 100)
|
|
favorite = request.query.get("favorite", "true").lower() != "false"
|
|
offset = max(0, int(request.query.get("offset", 0)))
|
|
provider_instance = request.query.get("provider", "").strip() or None
|
|
|
|
_LOGGER.debug("Library request: type=%s limit=%d offset=%d favorite=%s provider=%s", media_type, limit, offset, favorite, provider_instance)
|
|
|
|
items = await _get_library_via_ma_client(hass, media_type, limit, favorite, offset, provider_instance)
|
|
if items is None:
|
|
_LOGGER.error("Library fetch failed for type=%s", media_type)
|
|
return self.json_message(
|
|
"Could not get library from Music Assistant. Check HA logs (filter: my_music_library).",
|
|
HTTPStatus.BAD_GATEWAY,
|
|
)
|
|
|
|
return web.json_response({"type": media_type, "items": items})
|
|
|
|
|
|
class MusicAssistantBrowseView(HomeAssistantView):
|
|
"""Browse the MA filesystem tree.
|
|
|
|
GET /my_music_library/browse?uri=<uri>
|
|
uri is optional — omit for root.
|
|
"""
|
|
|
|
url = "/my_music_library/browse"
|
|
name = "my_music_library:browse"
|
|
requires_auth = True
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
"""Handle browse request."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
uri = request.query.get("uri", "").strip() or None
|
|
limit = min(int(request.query.get("limit", 200)), 500)
|
|
|
|
_LOGGER.debug("Browse request: uri=%r limit=%d", uri, limit)
|
|
items = await _browse_via_ma_client(hass, uri, limit)
|
|
if items is None:
|
|
return self.json_message(
|
|
"Could not browse Music Assistant. Check HA logs.",
|
|
HTTPStatus.BAD_GATEWAY,
|
|
)
|
|
return web.json_response({"uri": uri or "", "items": items})
|
|
|
|
|
|
class MusicAssistantRecommendationsView(HomeAssistantView):
|
|
"""Return recommendation folders from Music Assistant.
|
|
|
|
GET /my_music_library/recommendations
|
|
→ {"folders": [{folder_id, name, icon, items: [...]}]}
|
|
"""
|
|
|
|
url = "/my_music_library/recommendations"
|
|
name = "my_music_library:recommendations"
|
|
requires_auth = True
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
"""Handle recommendations request."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
folders = await _get_recommendations_via_ma_client(hass)
|
|
if folders is None:
|
|
return self.json_message(
|
|
"Could not get recommendations from Music Assistant. Check HA logs.",
|
|
HTTPStatus.BAD_GATEWAY,
|
|
)
|
|
return web.json_response({"folders": folders})
|
|
|
|
|
|
class MusicAssistantSubitemsView(HomeAssistantView):
|
|
"""Return sub-items of a MA library item.
|
|
|
|
GET /my_music_library/subitems?action=artist_albums|artist_albums_all|album_tracks|playlist_tracks|track_artist&uri=<uri>
|
|
|
|
``artist_albums`` is library/favorites-scoped; ``artist_albums_all`` resolves
|
|
the artist's real provider mapping(s) to return its full catalog.
|
|
``track_artist`` returns the main artist of the track ``uri`` (zero or one item).
|
|
"""
|
|
|
|
url = "/my_music_library/subitems"
|
|
name = "my_music_library:subitems"
|
|
requires_auth = True
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
"""Handle subitems request."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
|
|
action = request.query.get("action", "").strip()
|
|
uri = request.query.get("uri", "").strip()
|
|
limit = min(int(request.query.get("limit", 50)), 200)
|
|
|
|
if action not in (
|
|
"artist_albums", "artist_albums_all", "album_tracks", "playlist_tracks", "track_artist",
|
|
):
|
|
return self.json_message("Invalid 'action' parameter.", HTTPStatus.BAD_REQUEST)
|
|
if not uri:
|
|
return self.json_message("Missing 'uri' parameter.", HTTPStatus.BAD_REQUEST)
|
|
|
|
_LOGGER.debug("Subitems request: action=%s uri=%s limit=%d", action, uri, limit)
|
|
if action == "artist_albums_all":
|
|
items = await _get_artist_all_albums(hass, uri, limit)
|
|
elif action == "track_artist":
|
|
items = await _get_track_artist(hass, uri)
|
|
else:
|
|
items = await _get_subitems(hass, action, uri, limit)
|
|
if items is None:
|
|
return self.json_message(
|
|
"Could not get subitems from Music Assistant. Check HA logs.",
|
|
HTTPStatus.BAD_GATEWAY,
|
|
)
|
|
|
|
return web.json_response({"action": action, "items": items})
|
|
|
|
|
|
class MusicAssistantProvidersView(HomeAssistantView):
|
|
"""Return the list of available Music Assistant providers.
|
|
|
|
GET /my_music_library/providers
|
|
"""
|
|
|
|
url = "/my_music_library/providers"
|
|
name = "my_music_library:providers"
|
|
requires_auth = True
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
"""Handle providers request."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
providers = await _get_providers_via_ma_client(hass)
|
|
return web.json_response({"providers": providers or []})
|
|
|
|
|
|
class PlayerQueueView(HomeAssistantView):
|
|
"""Per-player queue storage — accessible from any browser/device.
|
|
|
|
GET /my_music_library/queue?player=<entity_id>
|
|
→ {"queue": [...], "source": "<uri or null>"}
|
|
|
|
POST /my_music_library/queue
|
|
body: {"player": "<entity_id>", "queue": [...], "source": "<uri or null>"}
|
|
→ {"ok": true}
|
|
"""
|
|
|
|
url = "/my_music_library/queue"
|
|
name = "my_music_library:queue"
|
|
requires_auth = True
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
"""Return the stored queue for a player."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
player = request.query.get("player", "").strip()
|
|
if not player:
|
|
return self.json_message("Missing 'player' parameter.", HTTPStatus.BAD_REQUEST)
|
|
queues: dict = hass.data.get(DOMAIN, {}).get("queues", {})
|
|
data = queues.get(player, {"queue": [], "source": None})
|
|
return web.json_response(data)
|
|
|
|
async def post(self, request: web.Request) -> web.Response:
|
|
"""Save the queue for a player."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
try:
|
|
body: dict = await request.json()
|
|
except Exception: # noqa: BLE001
|
|
return self.json_message("Invalid JSON body.", HTTPStatus.BAD_REQUEST)
|
|
|
|
player = (body.get("player") or "").strip()
|
|
if not player:
|
|
return self.json_message("Missing 'player' field.", HTTPStatus.BAD_REQUEST)
|
|
|
|
entry = {
|
|
"queue": body.get("queue") or [],
|
|
"source": body.get("source") or None,
|
|
}
|
|
domain_data = hass.data.setdefault(DOMAIN, {})
|
|
domain_data.setdefault("queues", {})[player] = entry
|
|
|
|
await async_save_store(hass)
|
|
|
|
return web.json_response({"ok": True})
|
|
|
|
|
|
class OutputsView(HomeAssistantView):
|
|
"""Audio outputs for the card's output panel (see outputs.py).
|
|
|
|
GET /my_music_library/outputs
|
|
→ {"outputs": [...], "presets": [...]}
|
|
|
|
POST /my_music_library/outputs
|
|
body: {"action": "transfer", "source": "<entity_id>", "targets": ["<entity_id>", ...]}
|
|
| {"action": "set_members", "leader": "<entity_id>", "add": [...], "remove": [...]}
|
|
| {"action": "group_volume", "leader": "<entity_id>", "volume": 0-100}
|
|
| {"action": "power", "player": "<entity_id>", "powered": true|false}
|
|
| {"action": "save_preset", "name": "...", "leader": "<entity_id>", "members": [...], "id": "<optional>"}
|
|
| {"action": "delete_preset", "id": "..."}
|
|
→ {"ok": true} (+ "preset" for save_preset)
|
|
"""
|
|
|
|
url = "/my_music_library/outputs"
|
|
name = "my_music_library:outputs"
|
|
requires_auth = True
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
"""Return the outputs and the group presets."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
mass = outputs.get_mass(hass)
|
|
return web.json_response({
|
|
"outputs": outputs.describe_outputs(hass, mass) if mass else [],
|
|
"presets": outputs.get_presets(hass),
|
|
})
|
|
|
|
async def post(self, request: web.Request) -> web.Response:
|
|
"""Run an output action."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
try:
|
|
body: dict = await request.json()
|
|
except Exception: # noqa: BLE001
|
|
return self.json_message("Invalid JSON body.", HTTPStatus.BAD_REQUEST)
|
|
|
|
def _list(key: str) -> list[str]:
|
|
return [v for v in body.get(key) or [] if isinstance(v, str) and v]
|
|
|
|
action = body.get("action")
|
|
result: dict[str, Any] = {"ok": True}
|
|
try:
|
|
if action == "transfer":
|
|
await outputs.async_transfer(hass, body.get("source") or "", _list("targets"))
|
|
elif action == "set_members":
|
|
await outputs.async_set_members(hass, body.get("leader") or "", _list("add"), _list("remove"))
|
|
elif action == "group_volume":
|
|
await outputs.async_group_volume(hass, body.get("leader") or "", int(body.get("volume", 0)))
|
|
elif action == "power":
|
|
await outputs.async_power(hass, body.get("player") or "", bool(body.get("powered")))
|
|
elif action == "save_preset":
|
|
result["preset"] = await outputs.async_save_preset(
|
|
hass, str(body.get("name") or ""), body.get("leader") or "", _list("members"), body.get("id"),
|
|
)
|
|
elif action == "delete_preset":
|
|
await outputs.async_delete_preset(hass, str(body.get("id") or ""))
|
|
else:
|
|
return self.json_message("Invalid 'action'.", HTTPStatus.BAD_REQUEST)
|
|
except HomeAssistantError as err:
|
|
_LOGGER.warning("outputs %s failed: %s", action, err)
|
|
return self.json_message(str(err), HTTPStatus.BAD_GATEWAY)
|
|
except (TypeError, ValueError) as err:
|
|
return self.json_message(f"Invalid parameters: {err}", HTTPStatus.BAD_REQUEST)
|
|
return web.json_response(result)
|
|
|
|
|
|
class MAQueueView(HomeAssistantView):
|
|
"""Proxy to Music Assistant's native player queue.
|
|
|
|
GET /my_music_library/ma_queue?player=<entity_id>&limit=50&offset=0
|
|
→ {"items": [...], "queue_id": "..."}
|
|
|
|
POST /my_music_library/ma_queue
|
|
body: {"player": "<entity_id>", "action": "delete_item", "item_id": "..."}
|
|
→ {"ok": true}
|
|
"""
|
|
|
|
url = "/my_music_library/ma_queue"
|
|
name = "my_music_library:ma_queue"
|
|
requires_auth = True
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
"""Return the MA queue items for a player."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
player = request.query.get("player", "").strip()
|
|
if not player:
|
|
return self.json_message("Missing 'player' parameter.", HTTPStatus.BAD_REQUEST)
|
|
|
|
mass, queue_id = await _resolve_queue_id(hass, player)
|
|
if mass is None or queue_id is None:
|
|
_LOGGER.warning("ma_queue GET: cannot resolve queue for %s", player)
|
|
return self.json_message(
|
|
"Cannot resolve MA queue. Check HA logs.", HTTPStatus.BAD_GATEWAY,
|
|
)
|
|
|
|
limit = min(int(request.query.get("limit", 50)), 500)
|
|
offset = max(0, int(request.query.get("offset", 0)))
|
|
|
|
try:
|
|
items = await mass.player_queues.get_queue_items(queue_id, limit=limit, offset=offset)
|
|
normalized = [_normalize_queue_item(_to_json_safe(i)) for i in items]
|
|
_LOGGER.debug("ma_queue GET: queue=%s → %d items", queue_id, len(normalized))
|
|
return web.json_response({"items": normalized, "queue_id": queue_id})
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.warning("ma_queue GET failed: %s", err)
|
|
return self.json_message(
|
|
f"Failed to get queue items: {err}", HTTPStatus.BAD_GATEWAY,
|
|
)
|
|
|
|
async def post(self, request: web.Request) -> web.Response:
|
|
"""Execute a queue action (delete_item)."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
try:
|
|
body: dict = await request.json()
|
|
except Exception: # noqa: BLE001
|
|
return self.json_message("Invalid JSON body.", HTTPStatus.BAD_REQUEST)
|
|
|
|
player = (body.get("player") or "").strip()
|
|
action = (body.get("action") or "").strip()
|
|
if not player or not action:
|
|
return self.json_message("Missing 'player' or 'action'.", HTTPStatus.BAD_REQUEST)
|
|
|
|
mass, queue_id = await _resolve_queue_id(hass, player)
|
|
if mass is None or queue_id is None:
|
|
return self.json_message("Cannot resolve MA queue.", HTTPStatus.BAD_GATEWAY)
|
|
|
|
if action == "delete_item":
|
|
item_id = body.get("item_id")
|
|
if item_id is None:
|
|
return self.json_message("Missing 'item_id'.", HTTPStatus.BAD_REQUEST)
|
|
try:
|
|
await mass.player_queues.delete_item(queue_id, item_id)
|
|
return web.json_response({"ok": True})
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.warning("ma_queue delete_item failed: %s", err)
|
|
return self.json_message(f"delete_item failed: {err}", HTTPStatus.BAD_GATEWAY)
|
|
|
|
if action == "clear":
|
|
try:
|
|
await mass.player_queues.clear(queue_id)
|
|
return web.json_response({"ok": True})
|
|
except Exception as err: # noqa: BLE001
|
|
_LOGGER.warning("ma_queue clear failed: %s", err)
|
|
return self.json_message(f"clear failed: {err}", HTTPStatus.BAD_GATEWAY)
|
|
|
|
return self.json_message(f"Unknown action: {action}", HTTPStatus.BAD_REQUEST)
|
|
|
|
|
|
class PlayerQueueJumpView(HomeAssistantView):
|
|
"""Jump to a specific index in the player's existing MA queue.
|
|
|
|
POST /my_music_library/queue_jump
|
|
body: {"player": "<entity_id>", "index": <int>}
|
|
→ {"ok": true} on success, 502 otherwise
|
|
"""
|
|
|
|
url = "/my_music_library/queue_jump"
|
|
name = "my_music_library:queue_jump"
|
|
requires_auth = True
|
|
|
|
async def post(self, request: web.Request) -> web.Response:
|
|
"""Handle queue jump request."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
try:
|
|
body: dict = await request.json()
|
|
except Exception: # noqa: BLE001
|
|
return self.json_message("Invalid JSON body.", HTTPStatus.BAD_REQUEST)
|
|
|
|
player = (body.get("player") or "").strip()
|
|
try:
|
|
index = int(body.get("index"))
|
|
except (TypeError, ValueError):
|
|
return self.json_message("Missing or invalid 'index'.", HTTPStatus.BAD_REQUEST)
|
|
if not player or index < 0:
|
|
return self.json_message("Missing 'player' or invalid 'index'.", HTTPStatus.BAD_REQUEST)
|
|
|
|
ok = await _queue_jump_via_ma_client(hass, player, index)
|
|
if not ok:
|
|
return self.json_message(
|
|
"Queue jump failed. Check HA logs (filter: my_music_library).",
|
|
HTTPStatus.BAD_GATEWAY,
|
|
)
|
|
return web.json_response({"ok": True})
|
|
|
|
|
|
class ImageProxyView(HomeAssistantView):
|
|
"""Proxy an image URL through HA to avoid mixed-content / CORS issues.
|
|
|
|
GET /my_music_library/image_proxy?url=<encoded_url>
|
|
→ binary image response
|
|
"""
|
|
|
|
url = "/my_music_library/image_proxy"
|
|
name = "my_music_library:image_proxy"
|
|
requires_auth = False
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
"""Fetch an image URL server-side and return it."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
image_url = request.query.get("url", "").strip()
|
|
if not image_url or not image_url.startswith("http"):
|
|
return web.Response(status=HTTPStatus.BAD_REQUEST)
|
|
|
|
session = async_get_clientsession(hass)
|
|
try:
|
|
async with session.get(
|
|
image_url,
|
|
timeout=aiohttp.ClientTimeout(total=10),
|
|
) as resp:
|
|
if resp.status != 200:
|
|
return web.Response(status=resp.status)
|
|
body = await resp.read()
|
|
return web.Response(
|
|
body=body,
|
|
content_type=resp.content_type or "image/jpeg",
|
|
headers={"Cache-Control": "public, max-age=300"},
|
|
)
|
|
except Exception: # noqa: BLE001
|
|
return web.Response(status=HTTPStatus.BAD_GATEWAY)
|
|
|
|
|
|
class MAThumbnailView(HomeAssistantView):
|
|
"""Proxy thumbnail requests through the MA server.
|
|
|
|
GET /my_music_library/thumb?path=<raw_image_path>
|
|
|
|
Resolves image paths server-side via the Music Assistant server,
|
|
handling provider-specific auth (Plex tokens, etc.) and internal
|
|
MA references that browsers cannot access directly.
|
|
"""
|
|
|
|
url = "/my_music_library/thumb"
|
|
name = "my_music_library:thumb"
|
|
requires_auth = False
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
"""Resolve an image path via MA and return the image bytes."""
|
|
hass: HomeAssistant = request.app["hass"]
|
|
raw_path = request.query.get("path", "").strip()
|
|
if not raw_path:
|
|
return web.Response(status=HTTPStatus.BAD_REQUEST)
|
|
|
|
ma_url = _get_mass_url(hass)
|
|
session = async_get_clientsession(hass)
|
|
|
|
urls_to_try: list[str] = []
|
|
if raw_path.startswith("http"):
|
|
urls_to_try.append(raw_path)
|
|
if ma_url:
|
|
urls_to_try.append(
|
|
f"{ma_url}/api/image/{quote(raw_path, safe='')}"
|
|
)
|
|
elif ma_url:
|
|
urls_to_try.append(
|
|
f"{ma_url}/api/image/{quote(raw_path, safe='')}"
|
|
)
|
|
if raw_path.startswith("/"):
|
|
urls_to_try.append(f"{ma_url}{raw_path}")
|
|
|
|
for image_url in urls_to_try:
|
|
try:
|
|
async with session.get(
|
|
image_url,
|
|
timeout=aiohttp.ClientTimeout(total=10),
|
|
) as resp:
|
|
if resp.status != 200:
|
|
continue
|
|
ct = resp.content_type or "image/jpeg"
|
|
if not ct.startswith("image"):
|
|
continue
|
|
body = await resp.read()
|
|
return web.Response(
|
|
body=body,
|
|
content_type=ct,
|
|
headers={"Cache-Control": "public, max-age=3600"},
|
|
)
|
|
except Exception: # noqa: BLE001
|
|
continue
|
|
|
|
return web.Response(status=HTTPStatus.NOT_FOUND)
|