"""Map Spotify response data into application-owned models.""" from collections.abc import Mapping from dataclasses import dataclass from typing import cast from app.domain.matching import track_key from app.domain.models import CreatedPlaylist, Track @dataclass(frozen=True) class CurrentUser: """The stable identity fields returned for the current Spotify user.""" account_id: str display_name: str def parse_search_tracks(payload: object) -> list[Track]: """Map valid Spotify search items and discard malformed entries.""" root = _as_mapping(payload) tracks = _as_mapping(root.get("tracks")) if root is not None else None items = tracks.get("items") if tracks is not None else None if not isinstance(items, list): return [] parsed_tracks: list[Track] = [] seen_ids: set[str] = set() seen_keys: set[str] = set() for item in items: parsed_track = _parse_track(item) if parsed_track is None: continue normalized_key = track_key(parsed_track) if parsed_track.id in seen_ids or normalized_key in seen_keys: continue seen_ids.add(parsed_track.id) seen_keys.add(normalized_key) parsed_tracks.append(parsed_track) return parsed_tracks def parse_top_artists(payload: object) -> list[str]: """Map a top-artists page to valid artist names.""" root = _as_mapping(payload) items = root.get("items") if root is not None else None if not isinstance(items, list): return [] return [ name for item in items if (name := _required_string(_as_mapping(item), "name")) is not None ] def parse_track_page(payload: object) -> list[Track]: """Map a direct Spotify track page to domain tracks.""" root = _as_mapping(payload) items = root.get("items") if root is not None else None return _parse_track_items(items) def parse_saved_track_page(payload: object) -> list[Track]: """Map a saved-track wrapper page to domain tracks.""" root = _as_mapping(payload) items = root.get("items") if root is not None else None if not isinstance(items, list): return [] return _parse_track_items( [wrapper.get("track") for item in items if (wrapper := _as_mapping(item)) is not None] ) def parse_current_user(payload: object) -> CurrentUser: """Map a Spotify current-user response into stable identity fields.""" root = _as_mapping(payload) account_id = _required_string(root, "account_id") or _required_string(root, "id") display_name = _required_string(root, "display_name") if account_id is None or display_name is None: raise ValueError("Spotify returned an invalid current-user response") return CurrentUser(account_id=account_id, display_name=display_name) def parse_created_playlist(payload: object) -> CreatedPlaylist: """Map a Spotify playlist creation response.""" root = _as_mapping(payload) playlist_id = _required_string(root, "id") external_urls = _as_mapping(root.get("external_urls")) if root is not None else None url = _required_string(external_urls, "spotify") if playlist_id is None or url is None: raise ValueError("Spotify returned an invalid playlist response") return CreatedPlaylist(id=playlist_id, url=url) def _parse_track(payload: object) -> Track | None: item = _as_mapping(payload) track_id = _required_string(item, "id") uri = _required_string(item, "uri") title = _required_string(item, "name") album = _as_mapping(item.get("album")) if item is not None else None album_name = _required_string(album, "name") artists = _parse_artists(item.get("artists")) if item is not None else None if ( item is None or track_id is None or uri is None or title is None or album_name is None or artists is None ): return None return Track( id=track_id, uri=uri, title=title, artists=artists, album_name=album_name, album_art_url=_parse_album_art_url(album), external_url=_parse_external_url(item), ) def _parse_track_items(payload: object) -> list[Track]: if not isinstance(payload, list): return [] return [track for item in payload if (track := _parse_track(item)) is not None] def _parse_artists(payload: object) -> tuple[str, ...] | None: if not isinstance(payload, list) or not payload: return None names: list[str] = [] for artist_payload in payload: name = _required_string(_as_mapping(artist_payload), "name") if name is None: return None names.append(name) return tuple(names) def _parse_album_art_url(album: Mapping[str, object] | None) -> str | None: images = album.get("images") if album is not None else None if not isinstance(images, list) or not images: return None return _optional_string(_as_mapping(images[0]), "url") def _parse_external_url(item: Mapping[str, object]) -> str | None: return _optional_string(_as_mapping(item.get("external_urls")), "spotify") def _as_mapping(payload: object) -> Mapping[str, object] | None: if not isinstance(payload, Mapping): return None return cast(Mapping[str, object], payload) def _required_string(payload: Mapping[str, object] | None, key: str) -> str | None: value = _optional_string(payload, key) return value if value else None def _optional_string(payload: Mapping[str, object] | None, key: str) -> str | None: if payload is None: return None value = payload.get(key) return value if isinstance(value, str) else None