Chad Auld
88f2f6ee27
Rename integration for consistent branding: - User-facing name: "Spotify Voice Assistant" - Integration name: "Spotify Voice Assistant Search" - Domain: spotify_voice_assistant Changes: - Rename custom_components/spotify_search/ to spotify_voice_assistant/ - Update domain in manifest.json and __init__.py - Update all service references in examples - Update all documentation references
273 lines
12 KiB
Python
273 lines
12 KiB
Python
"""Spotify Voice Assistant Search Integration for Home Assistant."""
|
|
import logging
|
|
import re
|
|
import voluptuous as vol
|
|
from homeassistant.core import HomeAssistant, ServiceCall
|
|
import homeassistant.helpers.config_validation as cv
|
|
from homeassistant.helpers.typing import ConfigType
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
DOMAIN = "spotify_voice_assistant"
|
|
VALID_SEARCH_TYPES = {"artist", "album", "track", "playlist"}
|
|
|
|
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
|
|
|
|
# Cache Spotify client to avoid repeated lookups
|
|
_spotify_cache = {
|
|
"client": None,
|
|
"entity_id": None,
|
|
"user_playlists": None,
|
|
}
|
|
|
|
|
|
def clean_query(query: str, search_type: str) -> str:
|
|
"""Remove common command words to improve search accuracy."""
|
|
query = query.lower().strip()
|
|
|
|
# Remove "play" from the start of any query
|
|
if query.startswith("play "):
|
|
query = query[5:]
|
|
|
|
# Remove type-specific filler words
|
|
if search_type == "artist":
|
|
# Remove "artist" prefix if LLM included it
|
|
query = query.replace("artist ", "").replace(
|
|
"group ", "").replace("band ", "")
|
|
elif search_type == "album":
|
|
# Remove "album" prefix
|
|
query = query.replace("album ", "")
|
|
elif search_type == "track":
|
|
# Remove "song" or "track" prefix
|
|
query = query.replace("song ", "").replace("track ", "")
|
|
|
|
return " ".join(query.split()).strip()
|
|
|
|
|
|
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
|
"""Set up the Spotify Search component."""
|
|
|
|
async def get_spotify_client():
|
|
"""Get Spotify client with caching and validation."""
|
|
if _spotify_cache["client"] is not None:
|
|
if _spotify_cache["entity_id"] in hass.states.async_entity_ids():
|
|
_LOGGER.debug("Using cached Spotify client")
|
|
return _spotify_cache["client"]
|
|
else:
|
|
_LOGGER.info(
|
|
"Cached Spotify entity no longer exists, invalidating cache")
|
|
_spotify_cache["client"] = None
|
|
_spotify_cache["entity_id"] = None
|
|
_spotify_cache["user_playlists"] = None
|
|
|
|
_LOGGER.debug("Cache miss, performing Spotify entity lookup")
|
|
|
|
spotify_entity_id = None
|
|
for state in hass.states.async_all("media_player"):
|
|
if "spotify" in state.entity_id.lower():
|
|
spotify_entity_id = state.entity_id
|
|
break
|
|
|
|
if not spotify_entity_id:
|
|
_LOGGER.error("No Spotify media player entity found")
|
|
raise LookupError("Spotify not configured")
|
|
|
|
entity_component = hass.data.get(
|
|
"entity_components", {}).get("media_player")
|
|
if not entity_component:
|
|
raise LookupError("Media player component not available")
|
|
|
|
spotify_entity = None
|
|
for entity in entity_component.entities:
|
|
if entity.entity_id == spotify_entity_id:
|
|
spotify_entity = entity
|
|
break
|
|
|
|
if not spotify_entity:
|
|
raise LookupError("Spotify entity not available")
|
|
|
|
if not hasattr(spotify_entity, "coordinator"):
|
|
raise AttributeError("Spotify coordinator not available")
|
|
|
|
coordinator = spotify_entity.coordinator
|
|
if not hasattr(coordinator, "client"):
|
|
raise AttributeError("Spotify client not available")
|
|
|
|
client = coordinator.client
|
|
_spotify_cache["client"] = client
|
|
_spotify_cache["entity_id"] = spotify_entity_id
|
|
_LOGGER.info("Cached Spotify client for entity: %s", spotify_entity_id)
|
|
|
|
return client
|
|
|
|
async def search_spotify(call: ServiceCall):
|
|
"""Search Spotify and return the first result's URI."""
|
|
raw_query = call.data.get("query")
|
|
search_type = call.data.get("type", "artist")
|
|
|
|
if not raw_query:
|
|
_LOGGER.error("No query provided to spotify_voice_assistant")
|
|
return {"error": "No query provided"}
|
|
|
|
if search_type not in VALID_SEARCH_TYPES:
|
|
return {"error": f"Invalid type. Must be one of: {', '.join(VALID_SEARCH_TYPES)}"}
|
|
|
|
# --- UPDATE: Clean the query before sending to Spotify ---
|
|
query = clean_query(raw_query, search_type)
|
|
_LOGGER.info("Searching Spotify (%s) for cleaned query: '%s' (raw: '%s')",
|
|
search_type, query, raw_query)
|
|
|
|
try:
|
|
client = await get_spotify_client()
|
|
except (LookupError, AttributeError) as err:
|
|
_LOGGER.error("Failed to get Spotify client: %s", err)
|
|
return {"error": str(err)}
|
|
|
|
try:
|
|
if search_type == "artist":
|
|
results = await client.search(query, ["artist"], limit=10)
|
|
items_list = results.artists
|
|
if items_list and len(items_list) > 0:
|
|
exact_match = None
|
|
query_lower = query.lower()
|
|
for artist in items_list:
|
|
if hasattr(artist, "name") and artist.name.lower() == query_lower:
|
|
exact_match = artist
|
|
break
|
|
|
|
selected_artist = exact_match if exact_match else items_list[0]
|
|
|
|
if not hasattr(selected_artist, "uri"):
|
|
return {"error": "Invalid artist data from Spotify"}
|
|
|
|
uri = selected_artist.uri
|
|
name = selected_artist.name
|
|
match_type = "exact match" if exact_match else "first result"
|
|
result = {"uri": uri, "name": name, "type": "artist"}
|
|
_LOGGER.info("✅ SEARCH RESULT (%s): %s",
|
|
match_type, result)
|
|
return result
|
|
else:
|
|
error_result = {"error": f"No artist found for: {query}"}
|
|
_LOGGER.error("❌ SEARCH ERROR: %s", error_result)
|
|
return error_result
|
|
|
|
elif search_type == "playlist":
|
|
# Cleaning is already handled by clean_query logic above,
|
|
# but we keep the specific 'playlist' word removal for safety
|
|
query_cleaned = query.lower().replace(
|
|
"playlist", "").replace("playlists", "").strip()
|
|
|
|
# 1. Search User Library
|
|
try:
|
|
if _spotify_cache["user_playlists"] is None:
|
|
user_playlists_response = await client.get_playlists_for_current_user()
|
|
if user_playlists_response and hasattr(user_playlists_response, "items"):
|
|
_spotify_cache["user_playlists"] = user_playlists_response.items
|
|
else:
|
|
_spotify_cache["user_playlists"] = []
|
|
|
|
user_playlists = _spotify_cache["user_playlists"]
|
|
|
|
# Exact match in library
|
|
for playlist in user_playlists:
|
|
if hasattr(playlist, "name") and playlist.name.lower() == query_cleaned:
|
|
result = {"uri": playlist.uri,
|
|
"name": playlist.name, "type": "playlist"}
|
|
_LOGGER.info(
|
|
"✅ SEARCH RESULT (user library - exact match): %s", result)
|
|
return result
|
|
|
|
# Partial match in library
|
|
for playlist in user_playlists:
|
|
if hasattr(playlist, "name") and query_cleaned in playlist.name.lower():
|
|
result = {"uri": playlist.uri,
|
|
"name": playlist.name, "type": "playlist"}
|
|
_LOGGER.info(
|
|
"✅ SEARCH RESULT (user library - partial match): %s", result)
|
|
return result
|
|
|
|
except Exception as err:
|
|
_LOGGER.warning("Error searching user playlists: %s", err)
|
|
|
|
# 2. Fallback to Public Search
|
|
results = await client.search(query_cleaned, ["playlist"], limit=10)
|
|
items_list = results.playlists
|
|
if items_list and len(items_list) > 0:
|
|
selected_playlist = items_list[0]
|
|
result = {"uri": selected_playlist.uri,
|
|
"name": selected_playlist.name, "type": "playlist"}
|
|
_LOGGER.info(
|
|
"✅ SEARCH RESULT (public playlist - first result): %s", result)
|
|
return result
|
|
else:
|
|
error_result = {"error": f"No playlist found for: {query}"}
|
|
_LOGGER.error("❌ SEARCH ERROR: %s", error_result)
|
|
return error_result
|
|
|
|
else:
|
|
# Handle Album and Track
|
|
results = await client.search(query, [search_type], limit=10)
|
|
items_list = getattr(results, f"{search_type}s", None)
|
|
|
|
if items_list and len(items_list) > 0:
|
|
exact_match = None
|
|
query_lower = query.lower()
|
|
for item in items_list:
|
|
if hasattr(item, "name") and item.name.lower() == query_lower:
|
|
exact_match = item
|
|
break
|
|
|
|
# Fallback logic for Albums
|
|
if not exact_match and search_type == "album" and len(query.split()) >= 2:
|
|
_LOGGER.info(
|
|
"No exact album match for '%s', trying track search", query)
|
|
try:
|
|
track_results = await client.search(query, ["track"], limit=10)
|
|
track_items = getattr(
|
|
track_results, "tracks", None)
|
|
if track_items and len(track_items) > 0:
|
|
first_track = track_items[0]
|
|
result = {"uri": first_track.uri,
|
|
"name": first_track.name, "type": "track"}
|
|
_LOGGER.info(
|
|
"✅ SEARCH RESULT (album→track fallback): %s", result)
|
|
return result
|
|
except Exception:
|
|
pass
|
|
|
|
selected_item = exact_match if exact_match else items_list[0]
|
|
match_type = "exact match" if exact_match else "first result"
|
|
result = {"uri": selected_item.uri,
|
|
"name": selected_item.name, "type": search_type}
|
|
_LOGGER.info("✅ SEARCH RESULT (%s - %s): %s",
|
|
search_type, match_type, result)
|
|
return result
|
|
else:
|
|
error_result = {
|
|
"error": f"No {search_type} found for: {query}"}
|
|
_LOGGER.error("❌ SEARCH ERROR: %s", error_result)
|
|
return error_result
|
|
|
|
except Exception as err:
|
|
_LOGGER.exception("Unexpected error searching Spotify")
|
|
return {"error": "Search failed"}
|
|
|
|
async def clear_cache(call: ServiceCall):
|
|
"""Clear Spotify client and user playlists cache."""
|
|
if _spotify_cache["client"] is not None or _spotify_cache["user_playlists"] is not None:
|
|
_spotify_cache["client"] = None
|
|
_spotify_cache["entity_id"] = None
|
|
_spotify_cache["user_playlists"] = None
|
|
return {"success": True, "message": "Cache cleared"}
|
|
else:
|
|
return {"success": False, "message": "Cache was already empty"}
|
|
|
|
hass.services.async_register(
|
|
DOMAIN, "search", search_spotify, supports_response="only"
|
|
)
|
|
hass.services.async_register(
|
|
DOMAIN, "clear_cache", clear_cache, supports_response="only"
|
|
)
|
|
return True
|