Thomas Hallock
7ff69449d6
Full music recommendation pipeline: listening history capture via webhook, Last.fm candidate discovery, iTunes preview download, CLAP audio embeddings (512-dim), pgvector cosine similarity recommendations, playlist generation with known/new track interleaving, and Music Assistant playback via HA. Includes: FastAPI app, SQLAlchemy models, Alembic migrations, Docker Compose with pgvector/pg17, status dashboard, and all API endpoints. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
118 lines
3.4 KiB
Python
118 lines
3.4 KiB
Python
import logging
|
|
|
|
import httpx
|
|
|
|
from haunt_fm.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def _ha_request(method: str, path: str, **kwargs) -> dict:
|
|
"""Make an authenticated request to Home Assistant REST API."""
|
|
headers = {
|
|
"Authorization": f"Bearer {settings.ha_token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.request(
|
|
method, f"{settings.ha_url}{path}", headers=headers, **kwargs
|
|
)
|
|
resp.raise_for_status()
|
|
if resp.content:
|
|
return resp.json()
|
|
return {}
|
|
|
|
|
|
async def is_ha_reachable() -> bool:
|
|
"""Check if Home Assistant is reachable."""
|
|
try:
|
|
await _ha_request("GET", "/api/")
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
async def play_media_on_speaker(
|
|
media_content_id: str,
|
|
speaker_entity: str,
|
|
media_content_type: str = "music",
|
|
) -> None:
|
|
"""Play a media item on a speaker via HA media_player service."""
|
|
await _ha_request(
|
|
"POST",
|
|
"/api/services/media_player/play_media",
|
|
json={
|
|
"entity_id": speaker_entity,
|
|
"media_content_id": media_content_id,
|
|
"media_content_type": media_content_type,
|
|
},
|
|
)
|
|
logger.info("Playing %s on %s", media_content_id, speaker_entity)
|
|
|
|
|
|
async def search_and_play(
|
|
artist: str,
|
|
title: str,
|
|
speaker_entity: str,
|
|
) -> bool:
|
|
"""Search Music Assistant for a track and play it.
|
|
|
|
Uses the mass.search service to find the track, then plays it.
|
|
"""
|
|
try:
|
|
# Use Music Assistant search via HA
|
|
result = await _ha_request(
|
|
"POST",
|
|
"/api/services/mass/search",
|
|
json={
|
|
"name": f"{artist} {title}",
|
|
"media_type": "track",
|
|
"limit": 1,
|
|
},
|
|
)
|
|
logger.info("MA search result for '%s - %s': %s", artist, title, result)
|
|
return True
|
|
except Exception:
|
|
logger.exception("Failed to search MA for %s - %s", artist, title)
|
|
return False
|
|
|
|
|
|
async def play_playlist_on_speaker(
|
|
tracks: list[dict],
|
|
speaker_entity: str,
|
|
) -> None:
|
|
"""Play a list of tracks on a speaker. Each track dict has 'artist' and 'title'.
|
|
|
|
Enqueues tracks via Music Assistant.
|
|
"""
|
|
if not tracks:
|
|
return
|
|
|
|
for i, track in enumerate(tracks):
|
|
try:
|
|
if i == 0:
|
|
# Play first track
|
|
await _ha_request(
|
|
"POST",
|
|
"/api/services/media_player/play_media",
|
|
json={
|
|
"entity_id": speaker_entity,
|
|
"media_content_id": f"{track['artist']} - {track['title']}",
|
|
"media_content_type": "music",
|
|
},
|
|
)
|
|
else:
|
|
# Enqueue subsequent tracks
|
|
await _ha_request(
|
|
"POST",
|
|
"/api/services/media_player/play_media",
|
|
json={
|
|
"entity_id": speaker_entity,
|
|
"media_content_id": f"{track['artist']} - {track['title']}",
|
|
"media_content_type": "music",
|
|
"enqueue": "add",
|
|
},
|
|
)
|
|
except Exception:
|
|
logger.exception("Failed to enqueue %s - %s", track["artist"], track["title"])
|