82 lines
2.4 KiB
Python
82 lines
2.4 KiB
Python
"""Structural ports implemented by external service adapters."""
|
|
|
|
from collections.abc import AsyncIterator
|
|
from typing import Literal, Protocol
|
|
|
|
from app.domain.models import (
|
|
ConversationTurn,
|
|
CreatedPlaylist,
|
|
Intent,
|
|
PreviousRecommendation,
|
|
RerankSelection,
|
|
Track,
|
|
)
|
|
|
|
TimeRange = Literal["short_term", "long_term"]
|
|
|
|
|
|
class CatalogQuotaExhaustedError(Exception):
|
|
"""A catalog quota stopped further resolution attempts."""
|
|
|
|
|
|
class RecommenderOutputError(Exception):
|
|
"""The recommender returned unusable structured output."""
|
|
|
|
|
|
class MusicCatalog(Protocol):
|
|
"""Read the restricted Spotify surface used by discovery."""
|
|
|
|
async def search_tracks(self, query: str, limit: int = 10) -> list[Track]:
|
|
"""Search tracks by fielded or bare-text query."""
|
|
...
|
|
|
|
async def fetch_top_artists(self, time_range: TimeRange, limit: int) -> list[str]:
|
|
"""Fetch bounded top artist names for one time range."""
|
|
...
|
|
|
|
async def fetch_top_tracks(self, time_range: TimeRange, limit: int) -> list[Track]:
|
|
"""Fetch bounded top tracks for one time range."""
|
|
...
|
|
|
|
async def fetch_saved_tracks(self, limit: int) -> list[Track]:
|
|
"""Fetch a bounded sample of saved tracks."""
|
|
...
|
|
|
|
|
|
class Recommender(Protocol):
|
|
"""Interpret discovery intent and stream a grounded reranking."""
|
|
|
|
async def create_intent(
|
|
self,
|
|
query: str,
|
|
history: tuple[ConversationTurn, ...],
|
|
previous_recommendations: tuple[PreviousRecommendation, ...],
|
|
taste_summary: str,
|
|
candidate_count: int,
|
|
) -> Intent:
|
|
"""Interpret a query and propose a bounded candidate set."""
|
|
...
|
|
|
|
def stream_rerank(
|
|
self,
|
|
intent: Intent,
|
|
grounded_tracks: tuple[Track, ...],
|
|
taste_summary: str,
|
|
history: tuple[ConversationTurn, ...],
|
|
selection_count: int,
|
|
correction: str | None = None,
|
|
) -> AsyncIterator[RerankSelection]:
|
|
"""Stream validated selections from the grounded pool."""
|
|
...
|
|
|
|
|
|
class PlaylistWriter(Protocol):
|
|
"""Write a Spotify playlist without retrying ambiguous mutations."""
|
|
|
|
async def create_playlist(self, name: str, description: str) -> CreatedPlaylist:
|
|
"""Create a private playlist."""
|
|
...
|
|
|
|
async def add_tracks_to_playlist(self, playlist_id: str, track_uris: list[str]) -> None:
|
|
"""Add ordered Spotify track URIs to a playlist."""
|
|
...
|