From 85984d4174345bbd8a500b493efd013082f9d181 Mon Sep 17 00:00:00 2001 From: Broque Thomas <26755000+Nezreka@users.noreply.github.com> Date: Sat, 16 May 2026 07:46:41 -0700 Subject: [PATCH] Amazon Music provider: metadata client + download source (T2Tunes) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit core/amazon_client.py — T2Tunes-backed metadata client following the DeezerClient/iTunesClient contract. Exposes search_tracks, search_artists, search_albums, get_track_details, get_album, get_album_tracks, get_artist, get_artist_albums, get_track_features. T2TunesStreamInfo dataclass captures the hex decryption key returned by the proxy (CENC/AES-128). Handles the "stremeable" API typo. 0.5 s rate-limit guard + api_call_tracker. core/amazon_download_client.py — DownloadSourcePlugin backed by the above client. Codec waterfall: FLAC → Opus → EAC3. Downloads the encrypted MP4 container, decrypts with ffmpeg -decryption_key, yields the native audio file (.flac / .opus / .eac3). Not yet wired into the app source registry — validated in isolation only; see tests/tools/. tools/t2tunes_probe.py + tools/t2tunes_media_plan.py — standalone CLI tools used for live API exploration during development. tests/tools/test_amazon_client.py — 72 unit tests (all mocked). tests/tools/test_amazon_download_client.py — 52 unit tests (all mocked). 124 tests pass. --- core/amazon_client.py | 604 ++++++++++++++ core/amazon_download_client.py | 432 ++++++++++ tests/tools/__init__.py | 0 tests/tools/test_amazon_client.py | 915 +++++++++++++++++++++ tests/tools/test_amazon_download_client.py | 783 ++++++++++++++++++ tools/t2tunes_media_plan.py | 199 +++++ tools/t2tunes_probe.py | 369 +++++++++ webui/static/helper.js | 4 + 8 files changed, 3306 insertions(+) create mode 100644 core/amazon_client.py create mode 100644 core/amazon_download_client.py create mode 100644 tests/tools/__init__.py create mode 100644 tests/tools/test_amazon_client.py create mode 100644 tests/tools/test_amazon_download_client.py create mode 100644 tools/t2tunes_media_plan.py create mode 100644 tools/t2tunes_probe.py diff --git a/core/amazon_client.py b/core/amazon_client.py new file mode 100644 index 00000000..421781ae --- /dev/null +++ b/core/amazon_client.py @@ -0,0 +1,604 @@ +"""Amazon Music metadata client backed by a T2Tunes proxy instance. + +T2Tunes exposes the Amazon Music catalog (search, album metadata, stream info) +through a simple REST API. This module wraps those endpoints and normalises the +responses into the same Track / Artist / Album dataclass shape used by +DeezerClient and iTunesClient. + +Endpoints used: + GET /api/status + GET /api/amazon-music/search + GET /api/amazon-music/metadata + GET /api/amazon-music/media-from-asin + +Config keys (all optional — fall back to public defaults): + amazon.base_url T2Tunes instance URL (default: https://t2tunes.site) + amazon.country ISO-3166 country code (default: US) + amazon.preferred_codec Preferred audio codec (default: flac) +""" + +from __future__ import annotations + +import threading +import time +from dataclasses import dataclass +from typing import Any, Dict, Iterator, List, Optional +from urllib.parse import urljoin + +import requests + +from config.settings import config_manager +from core.api_call_tracker import api_call_tracker +from utils.logging_config import get_logger + +logger = get_logger("amazon_client") + +DEFAULT_BASE_URL = "https://t2tunes.site" +DEFAULT_COUNTRY = "US" +DEFAULT_CODEC = "flac" +MIN_API_INTERVAL = 0.5 # seconds — T2Tunes has no published rate limit + +_last_api_call: float = 0.0 +_api_call_lock = threading.Lock() + + +class AmazonClientError(RuntimeError): + """Raised on unrecoverable T2Tunes API errors.""" + + +# --------------------------------------------------------------------------- +# Dataclasses — field layout matches DeezerClient / iTunesClient exactly +# --------------------------------------------------------------------------- + +@dataclass +class Track: + id: str # Amazon track ASIN + name: str + artists: List[str] + album: str + duration_ms: int + popularity: int + preview_url: Optional[str] = None + external_urls: Optional[Dict[str, str]] = None + image_url: Optional[str] = None + release_date: Optional[str] = None + track_number: Optional[int] = None + disc_number: Optional[int] = None + album_type: Optional[str] = None + total_tracks: Optional[int] = None + isrc: Optional[str] = None + + @classmethod + def from_search_hit(cls, doc: Dict[str, Any]) -> "Track": + return cls( + id=str(doc.get("asin") or ""), + name=str(doc.get("title") or ""), + artists=[str(doc.get("artistName") or "Unknown Artist")], + album=str(doc.get("albumName") or ""), + duration_ms=int(doc.get("duration") or 0) * 1000, + popularity=0, + isrc=str(doc.get("isrc") or "") or None, + ) + + @classmethod + def from_stream_info( + cls, + stream: "T2TunesStreamInfo", + album_meta: Optional[Dict[str, Any]] = None, + ) -> "Track": + album = album_meta or {} + return cls( + id=stream.asin, + name=stream.title, + artists=[stream.artist] if stream.artist else ["Unknown Artist"], + album=stream.album, + duration_ms=0, + popularity=0, + image_url=album.get("image"), + release_date=album.get("release_date"), + total_tracks=album.get("trackCount"), + isrc=stream.isrc or None, + ) + + +@dataclass +class Artist: + id: str # Slugified artist name — T2Tunes exposes no artist IDs + name: str + popularity: int + genres: List[str] + followers: int + image_url: Optional[str] = None + external_urls: Optional[Dict[str, str]] = None + + @classmethod + def from_name(cls, name: str) -> "Artist": + slug = name.lower().replace(" ", "_") + return cls(id=slug, name=name, popularity=0, genres=[], followers=0) + + +@dataclass +class Album: + id: str # Amazon album ASIN + name: str + artists: List[str] + release_date: str + total_tracks: int + album_type: str + image_url: Optional[str] = None + external_urls: Optional[Dict[str, str]] = None + explicit: Optional[bool] = None + + @classmethod + def from_search_hit(cls, doc: Dict[str, Any]) -> "Album": + return cls( + id=str(doc.get("albumAsin") or doc.get("asin") or ""), + name=str(doc.get("albumName") or doc.get("title") or ""), + artists=[str(doc.get("artistName") or "Unknown Artist")], + release_date="", + total_tracks=0, + album_type="album", + ) + + @classmethod + def from_metadata(cls, album_meta: Dict[str, Any], asin: str = "") -> "Album": + return cls( + id=str(album_meta.get("asin") or asin or ""), + name=str(album_meta.get("title") or ""), + artists=[str(album_meta.get("artistName") or "Unknown Artist")], + release_date=str(album_meta.get("release_date") or ""), + total_tracks=int(album_meta.get("trackCount") or 0), + album_type="album", + image_url=album_meta.get("image"), + explicit=album_meta.get("explicit"), + ) + + +# --------------------------------------------------------------------------- +# Internal dataclasses for raw T2Tunes response parsing +# --------------------------------------------------------------------------- + +@dataclass(frozen=True) +class T2TunesSearchItem: + asin: str + title: str + artist_name: str + item_type: str + album_name: str = "" + album_asin: str = "" + duration_seconds: int = 0 + isrc: str = "" + + @property + def is_album(self) -> bool: + return "album" in self.item_type.lower() + + @property + def is_track(self) -> bool: + return "track" in self.item_type.lower() + + +@dataclass(frozen=True) +class T2TunesStreamInfo: + asin: str + streamable: bool + codec: str + format: str + sample_rate: Optional[int] + stream_url: str + decryption_key: Optional[str] # hex-encoded AES key; None when stream is clear + title: str = "" + artist: str = "" + album: str = "" + isrc: str = "" + cover_url: str = "" + track_number: Optional[int] = None + disc_number: Optional[int] = None + genre: str = "" + label: str = "" + date: str = "" + + @property + def has_decryption_key(self) -> bool: + return bool(self.decryption_key) + + +# --------------------------------------------------------------------------- +# Rate-limit enforcement +# --------------------------------------------------------------------------- + +def _rate_limit() -> None: + global _last_api_call + with _api_call_lock: + elapsed = time.monotonic() - _last_api_call + if elapsed < MIN_API_INTERVAL: + time.sleep(MIN_API_INTERVAL - elapsed) + _last_api_call = time.monotonic() + api_call_tracker.record_call("amazon") + + +# --------------------------------------------------------------------------- +# Main client +# --------------------------------------------------------------------------- + +class AmazonClient: + """T2Tunes-backed Amazon Music metadata and stream-info client.""" + + def __init__( + self, + base_url: Optional[str] = None, + country: Optional[str] = None, + preferred_codec: Optional[str] = None, + timeout: int = 30, + session: Optional[Any] = None, + ) -> None: + self.base_url = (base_url or config_manager.get("amazon.base_url", DEFAULT_BASE_URL)).rstrip("/") + self.country = (country or config_manager.get("amazon.country", DEFAULT_COUNTRY)).upper() + self.preferred_codec = ( + preferred_codec or config_manager.get("amazon.preferred_codec", DEFAULT_CODEC) + ).lower() + self.timeout = timeout + self.session: Any = session or requests.Session() + if isinstance(self.session, requests.Session): + self.session.headers.update({ + "Accept": "application/json", + "User-Agent": "SoulSync/1.0", + "Referer": self.base_url, + }) + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + def reload_config(self) -> None: + self.base_url = config_manager.get("amazon.base_url", DEFAULT_BASE_URL).rstrip("/") + self.country = config_manager.get("amazon.country", DEFAULT_COUNTRY).upper() + self.preferred_codec = config_manager.get("amazon.preferred_codec", DEFAULT_CODEC).lower() + + def is_authenticated(self) -> bool: + """Return True when the T2Tunes instance reports Amazon Music as up.""" + try: + return str(self.status().get("amazonMusic", "")).lower() == "up" + except AmazonClientError: + return False + + # ------------------------------------------------------------------ + # Low-level API wrappers + # ------------------------------------------------------------------ + + def status(self) -> Dict[str, Any]: + return self._get_json("/api/status") + + def search_raw(self, query: str, *, types: str = "track,album") -> List[T2TunesSearchItem]: + data = self._get_json( + "/api/amazon-music/search", + params={"query": query, "types": types, "country": self.country}, + ) + return list(self._iter_search_items(data)) + + def album_metadata(self, asin: str) -> Dict[str, Any]: + return self._get_json( + "/api/amazon-music/metadata", + params={"asin": asin, "country": self.country}, + ) + + def media_from_asin(self, asin: str, codec: Optional[str] = None) -> List[T2TunesStreamInfo]: + effective_codec = (codec or self.preferred_codec).lower() + data = self._get_json( + "/api/amazon-music/media-from-asin", + params={"asin": asin, "country": self.country, "codec": effective_codec}, + ) + if isinstance(data, list): + return [self._parse_stream_info(item) for item in data if isinstance(item, dict)] + if isinstance(data, dict): + return [self._parse_stream_info(data)] + raise AmazonClientError(f"Unexpected media-from-asin response type: {type(data).__name__}") + + # ------------------------------------------------------------------ + # Metadata interface — mirrors DeezerClient / iTunesClient signatures + # ------------------------------------------------------------------ + + def search_tracks(self, query: str, limit: int = 20) -> List[Track]: + _rate_limit() + items = self.search_raw(query, types="track") + tracks: List[Track] = [] + for item in items: + if not item.is_track: + continue + tracks.append(Track.from_search_hit({ + "asin": item.asin, + "title": item.title, + "artistName": item.artist_name, + "albumName": item.album_name, + "albumAsin": item.album_asin, + "duration": item.duration_seconds, + "isrc": item.isrc, + })) + if len(tracks) >= limit: + break + return tracks + + def search_artists(self, query: str, limit: int = 20) -> List[Artist]: + _rate_limit() + items = self.search_raw(query, types="track") + seen: Dict[str, Artist] = {} + for item in items: + name = item.artist_name + if name and name not in seen: + seen[name] = Artist.from_name(name) + if len(seen) >= limit: + break + return list(seen.values()) + + def search_albums(self, query: str, limit: int = 20) -> List[Album]: + _rate_limit() + items = self.search_raw(query, types="album") + albums: List[Album] = [] + seen_asins: set = set() + for item in items: + if not item.is_album: + continue + album_asin = item.album_asin or item.asin + if album_asin in seen_asins: + continue + seen_asins.add(album_asin) + albums.append(Album.from_search_hit({ + "albumAsin": album_asin, + "albumName": item.album_name or item.title, + "artistName": item.artist_name, + })) + if len(albums) >= limit: + break + return albums + + def get_track_details(self, asin: str) -> Optional[Dict[str, Any]]: + """Return a Spotify-compatible dict for a single track ASIN.""" + _rate_limit() + try: + streams = self.media_from_asin(asin) + except AmazonClientError: + return None + if not streams: + return None + s = streams[0] + + album_data: Dict[str, Any] = {} + try: + meta = self.album_metadata(asin) + albums = meta.get("albumList") + if isinstance(albums, list) and albums and isinstance(albums[0], dict): + album_data = albums[0] + except AmazonClientError: + pass + + return { + "id": s.asin, + "name": s.title, + "artists": [{"name": s.artist, "id": ""}], + "album": { + "id": album_data.get("asin", ""), + "name": s.album, + "images": [{"url": album_data["image"]}] if album_data.get("image") else [], + "release_date": album_data.get("release_date", ""), + "total_tracks": album_data.get("trackCount", 0), + }, + "duration_ms": 0, + "popularity": 0, + "external_urls": {"amazon": f"https://music.amazon.com/albums/{asin}"}, + "track_number": None, + "disc_number": None, + "isrc": s.isrc, + "is_album_track": True, + "raw_data": { + "codec": s.codec, + "format": s.format, + "sample_rate": s.sample_rate, + "streamable": s.streamable, + "has_decryption_key": s.has_decryption_key, + }, + } + + def get_album(self, asin: str, include_tracks: bool = True) -> Optional[Dict[str, Any]]: + """Return a Spotify-compatible album dict.""" + _rate_limit() + try: + meta = self.album_metadata(asin) + except AmazonClientError: + return None + + albums = meta.get("albumList") + if not isinstance(albums, list) or not albums: + return None + album = albums[0] if isinstance(albums[0], dict) else {} + + result: Dict[str, Any] = { + "id": asin, + "name": album.get("title", ""), + "artists": [{"name": album.get("artistName", ""), "id": ""}], + "release_date": album.get("release_date", ""), + "total_tracks": album.get("trackCount", 0), + "album_type": "album", + "images": [{"url": album["image"]}] if album.get("image") else [], + "external_urls": {"amazon": f"https://music.amazon.com/albums/{asin}"}, + "label": album.get("label", ""), + } + if include_tracks: + result["tracks"] = self.get_album_tracks(asin) or { + "items": [], + "total": 0, + "limit": 50, + "next": None, + } + return result + + def get_album_tracks(self, asin: str) -> Optional[Dict[str, Any]]: + """Return album tracks in Spotify pagination format.""" + _rate_limit() + try: + streams = self.media_from_asin(asin) + except AmazonClientError: + return None + items = [ + { + "id": s.asin, + "name": s.title, + "artists": [{"name": s.artist, "id": ""}], + "duration_ms": 0, + "track_number": None, + "disc_number": None, + "isrc": s.isrc, + } + for s in streams + ] + return {"items": items, "total": len(items), "limit": 50, "next": None} + + def get_artist(self, artist_name: str) -> Optional[Dict[str, Any]]: + """Return a Spotify-compatible artist dict inferred from search results.""" + _rate_limit() + try: + items = self.search_raw(artist_name, types="track") + except AmazonClientError: + return None + name_lower = artist_name.lower() + match = next( + (i for i in items if i.artist_name.lower() == name_lower), + next((i for i in items if name_lower in i.artist_name.lower()), None), + ) + if not match: + return None + return { + "id": match.artist_name.lower().replace(" ", "_"), + "name": match.artist_name, + "genres": [], + "popularity": 0, + "followers": {"total": 0}, + "images": [], + "external_urls": {}, + } + + def get_artist_albums( + self, + artist_name: str, + album_type: str = "album,single", + limit: int = 200, + ) -> List[Album]: + """Return albums for an artist inferred from search results.""" + _rate_limit() + try: + items = self.search_raw(f"{artist_name} album", types="album") + except AmazonClientError: + return [] + albums: List[Album] = [] + seen_asins: set = set() + name_lower = artist_name.lower() + for item in items: + if item.artist_name.lower() != name_lower: + continue + album_asin = item.album_asin or item.asin + if album_asin in seen_asins: + continue + seen_asins.add(album_asin) + albums.append(Album.from_search_hit({ + "albumAsin": album_asin, + "albumName": item.album_name or item.title, + "artistName": item.artist_name, + })) + if len(albums) >= limit: + break + return albums + + def get_track_features(self, track_id: str) -> Optional[Dict[str, Any]]: + """Not available from Amazon Music — returns None for compatibility.""" + return None + + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ + + def _get_json(self, path: str, params: Optional[Dict[str, Any]] = None) -> Any: + url = urljoin(f"{self.base_url}/", path.lstrip("/")) + try: + resp = self.session.get(url, params=params, timeout=self.timeout) + resp.raise_for_status() + except requests.HTTPError as exc: + raise AmazonClientError( + f"HTTP {exc.response.status_code} for {url}" + ) from exc + except requests.RequestException as exc: + raise AmazonClientError(f"Request failed for {url}: {exc}") from exc + try: + return resp.json() + except ValueError as exc: + preview = resp.text[:200].replace("\n", " ") + raise AmazonClientError(f"Response not JSON for {url}: {preview!r}") from exc + + @staticmethod + def _iter_search_items(response: Any) -> Iterator[T2TunesSearchItem]: + if not isinstance(response, dict): + raise AmazonClientError( + f"Unexpected search response type: {type(response).__name__}" + ) + for result in response.get("results") or []: + if not isinstance(result, dict): + continue + for hit in result.get("hits") or []: + if not isinstance(hit, dict): + continue + doc = hit.get("document") + if not isinstance(doc, dict): + continue + asin = str(doc.get("asin") or "") + if not asin: + continue + yield T2TunesSearchItem( + asin=asin, + title=str(doc.get("title") or ""), + artist_name=str(doc.get("artistName") or ""), + item_type=str(doc.get("__type") or ""), + album_name=str(doc.get("albumName") or ""), + album_asin=str(doc.get("albumAsin") or ""), + duration_seconds=int(doc.get("duration") or 0), + isrc=str(doc.get("isrc") or ""), + ) + + @staticmethod + def _parse_stream_info(item: Dict[str, Any]) -> T2TunesStreamInfo: + stream_info = item.get("streamInfo") if isinstance(item.get("streamInfo"), dict) else {} + tags = item.get("tags") if isinstance(item.get("tags"), dict) else {} + # T2Tunes API has a typo: "stremeable" in some responses + streamable = item.get("streamable") + if streamable is None: + streamable = item.get("stremeable") + raw_key = item.get("decryptionKey") + decryption_key = str(raw_key) if raw_key else None + + def _int_tag(key: str) -> Optional[int]: + v = tags.get(key) + try: + return int(v) if v is not None else None + except (TypeError, ValueError): + return None + + return T2TunesStreamInfo( + asin=str(item.get("asin") or ""), + streamable=bool(streamable), + codec=str(stream_info.get("codec") or ""), + format=str(stream_info.get("format") or ""), + sample_rate=( + stream_info.get("sampleRate") + if isinstance(stream_info.get("sampleRate"), int) + else None + ), + stream_url=str(stream_info.get("streamUrl") or ""), + decryption_key=decryption_key, + title=str(tags.get("title") or ""), + artist=str(tags.get("artist") or ""), + album=str(tags.get("album") or ""), + isrc=str(tags.get("isrc") or ""), + cover_url=str(item.get("coverUrl") or ""), + track_number=_int_tag("trackNumber"), + disc_number=_int_tag("discNumber"), + genre=str(tags.get("genre") or ""), + label=str(tags.get("label") or ""), + date=str(tags.get("date") or ""), + ) diff --git a/core/amazon_download_client.py b/core/amazon_download_client.py new file mode 100644 index 00000000..81477b13 --- /dev/null +++ b/core/amazon_download_client.py @@ -0,0 +1,432 @@ +"""Amazon Music download source plugin backed by a T2Tunes proxy. + +NOT yet wired into the app registry — validated in isolation only. +See tests/tools/test_amazon_download_client.py. + +Filename encoding (the DownloadSourcePlugin dispatch contract): + "{asin}||{display_name}" + e.g. "B09XYZ1234||Kendrick Lamar - Not Like Us" + +Codec preference order: FLAC → Opus → EAC3. + +Download flow (from Tubifarry reference implementation): + 1. GET stream_url → encrypted bytes on disk + 2. FFmpeg -decryption_key to decrypt in place + 3. Embed metadata tags (handled by the app's standard post-processing) +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +import threading +import time +import uuid +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +import requests as http_requests + +from config.settings import config_manager +from core.amazon_client import AmazonClient, AmazonClientError +from core.download_plugins.base import DownloadSourcePlugin +from core.download_plugins.types import AlbumResult, DownloadStatus, TrackResult +from utils.logging_config import get_logger + +logger = get_logger("amazon_download_client") + +# Quality / codec helpers +CODEC_PREFERENCE = ["flac", "opus", "eac3"] + +_CODEC_EXTENSIONS: Dict[str, str] = { + "flac": "flac", + "ogg_vorbis": "ogg", + "opus": "opus", + "eac3": "eac3", + "mp4": "m4a", + "aac": "m4a", + "mp3": "mp3", +} + +MIN_AUDIO_BYTES = 512 * 1024 # 512 KB — anything smaller is a broken stream + + +def _codec_key(codec: str) -> str: + return codec.lower().replace("-", "_").replace(" ", "_") + + +def _file_extension(codec: str) -> str: + return _CODEC_EXTENSIONS.get(_codec_key(codec), "bin") + + +def _quality_label(codec: str, sample_rate: Optional[int] = None) -> str: + ck = _codec_key(codec) + if ck == "flac": + if sample_rate and sample_rate > 48000: + return "Hi-Res" + return "Lossless" + return "Lossy" + + +class AmazonDownloadClient(DownloadSourcePlugin): + """DownloadSourcePlugin — Amazon Music via T2Tunes proxy.""" + + def __init__(self, download_path: Optional[str] = None) -> None: + if download_path is None: + download_path = config_manager.get("soulseek.download_path", "./downloads") + self.download_path = Path(download_path) + self.download_path.mkdir(parents=True, exist_ok=True) + + self._client = AmazonClient() + self.session = http_requests.Session() + self.session.headers.update({ + "User-Agent": "SoulSync/1.0", + "Accept": "*/*", + }) + + self._engine: Any = None + self.shutdown_check: Any = None + + # ------------------------------------------------------------------ + # DownloadSourcePlugin — lifecycle + # ------------------------------------------------------------------ + + def is_configured(self) -> bool: + # T2Tunes has a public default instance; no credentials required. + # Return True unconditionally so the source shows as available. + return True + + async def check_connection(self) -> bool: + try: + return self._client.is_authenticated() + except Exception: + return False + + # ------------------------------------------------------------------ + # DownloadSourcePlugin — search + # ------------------------------------------------------------------ + + async def search( + self, + query: str, + timeout: Optional[int] = None, + progress_callback: Any = None, + ) -> Tuple[List[TrackResult], List[AlbumResult]]: + try: + items = self._client.search_raw(query, types="track,album") + except AmazonClientError as exc: + logger.warning(f"Amazon search failed for {query!r}: {exc}") + return [], [] + + track_results: List[TrackResult] = [] + album_map: Dict[str, AlbumResult] = {} + album_order: List[str] = [] + preferred = self._client.preferred_codec + + for item in items: + quality = _quality_label(preferred) + if item.is_track: + track_results.append(TrackResult( + username="amazon", + filename=f"{item.asin}||{item.artist_name} - {item.title}", + size=0, + bitrate=None, + duration=item.duration_seconds * 1000 if item.duration_seconds else None, + quality=quality, + free_upload_slots=999, + upload_speed=999_999, + queue_length=0, + artist=item.artist_name, + title=item.title, + album=item.album_name, + _source_metadata={ + "asin": item.asin, + "album_asin": item.album_asin, + "isrc": item.isrc, + }, + )) + elif item.is_album: + album_asin = item.album_asin or item.asin + if album_asin not in album_map: + placeholder = TrackResult( + username="amazon", + filename=f"{item.asin}||{item.artist_name} - {item.title}", + size=0, + bitrate=None, + duration=None, + quality=quality, + free_upload_slots=999, + upload_speed=999_999, + queue_length=0, + artist=item.artist_name, + title=item.title, + album=item.album_name, + ) + album_map[album_asin] = AlbumResult( + username="amazon", + album_path=album_asin, + album_title=item.album_name or item.title, + artist=item.artist_name, + track_count=0, + total_size=0, + tracks=[placeholder], + dominant_quality=quality, + ) + album_order.append(album_asin) + + return track_results, [album_map[k] for k in album_order] + + # ------------------------------------------------------------------ + # DownloadSourcePlugin — download dispatch + # ------------------------------------------------------------------ + + async def download( + self, + username: str, + filename: str, + file_size: int = 0, + ) -> Optional[str]: + if "||" not in filename: + logger.error(f"Invalid Amazon filename format (no '||'): {filename!r}") + return None + if self._engine is None: + raise RuntimeError( + "AmazonDownloadClient._engine is not set — " + "client not connected to download infrastructure" + ) + asin, display_name = filename.split("||", 1) + asin = asin.strip() + display_name = display_name.strip() + return self._engine.worker.dispatch( + source_name="amazon", + target_id=asin, + display_name=display_name, + original_filename=filename, + impl_callable=self._download_sync, + ) + + def _download_sync( + self, + download_id: str, + target_id: str, + display_name: str, + ) -> Optional[str]: + asin = str(target_id) + for codec in CODEC_PREFERENCE: + try: + streams = self._client.media_from_asin(asin, codec=codec) + except AmazonClientError as exc: + logger.warning(f"media_from_asin({asin!r}, {codec}) failed: {exc}") + continue + + stream = next( + (s for s in streams if s.streamable and s.stream_url), + None, + ) + if not stream: + logger.debug(f"No streamable result for {asin} at codec={codec}") + continue + + ext = _file_extension(stream.codec or codec) + safe = "".join( + ch if ch.isalnum() or ch in " -_." else "_" + for ch in display_name + )[:80] + # T2Tunes always serves audio in an encrypted MP4 container. + # The codec extension (.flac, .opus, .eac3) is only for the + # final decrypted output. + enc_ext = "mp4" if stream.decryption_key else ext + enc_path = self._unique_path(self.download_path / f"{safe}.enc.{enc_ext}") + out_path = self._unique_path(self.download_path / f"{safe}.{ext}") + + if self._engine is not None: + self._engine.update_record( + "amazon", download_id, {"state": "downloading", "progress": 0.0} + ) + try: + downloaded = self._stream_to_file(stream.stream_url, enc_path, download_id) + except Exception as exc: + logger.warning(f"Stream download failed for {asin} ({codec}): {exc}") + enc_path.unlink(missing_ok=True) + continue + + if downloaded < MIN_AUDIO_BYTES: + logger.warning( + f"File too small ({downloaded} B) for {asin} at {codec} — trying next" + ) + enc_path.unlink(missing_ok=True) + continue + + if stream.decryption_key: + if self._engine is not None: + self._engine.update_record( + "amazon", download_id, {"state": "decrypting", "progress": 1.0} + ) + try: + self._decrypt_with_ffmpeg(enc_path, out_path, stream.decryption_key) + enc_path.unlink(missing_ok=True) + except Exception as exc: + logger.error(f"Decryption failed for {asin} ({codec}): {exc}") + enc_path.unlink(missing_ok=True) + out_path.unlink(missing_ok=True) + continue + else: + enc_path.rename(out_path) + + logger.info( + f"Amazon download complete ({codec}): {out_path} " + f"({out_path.stat().st_size / (1024 * 1024):.1f} MB)" + ) + return str(out_path) + + logger.error(f"All codec tiers exhausted for '{display_name}' ({asin})") + return None + + def _decrypt_with_ffmpeg( + self, enc_path: Path, out_path: Path, hex_key: str + ) -> None: + """Decrypt a T2Tunes encrypted audio file using FFmpeg -decryption_key. + + T2Tunes uses CENC (Common Encryption) for DRM-protected tracks. + FFmpeg supports decryption via the -decryption_key flag when the + key is provided in hex. + """ + ffmpeg = shutil.which("ffmpeg") + if not ffmpeg: + tools_dir = Path(__file__).parent.parent / "tools" + candidate = tools_dir / ("ffmpeg.exe" if os.name == "nt" else "ffmpeg") + if candidate.exists(): + ffmpeg = str(candidate) + else: + raise RuntimeError( + "ffmpeg is required for Amazon Music decryption. " + "Install ffmpeg and ensure it is on your PATH." + ) + cmd = [ + ffmpeg, + "-y", + "-hide_banner", + "-loglevel", "error", + "-decryption_key", hex_key, + "-i", str(enc_path), + "-map", "0:a:0", # extract first audio stream (FLAC/Opus/EAC3 inside MP4) + "-c", "copy", + str(out_path), + ] + logger.debug(f"Decrypting {enc_path.name} → {out_path.name}") + result = subprocess.run(cmd, capture_output=True) + if result.returncode != 0: + stderr = result.stderr.decode("utf-8", errors="replace").strip() + raise RuntimeError(f"FFmpeg decryption failed (exit {result.returncode}): {stderr}") + + def _stream_to_file(self, url: str, out_path: Path, download_id: str) -> int: + resp = self.session.get(url, stream=True, timeout=60) + resp.raise_for_status() + + total = int(resp.headers.get("content-length", 0)) + downloaded = 0 + last_report = time.monotonic() + shutdown_triggered = False + + with out_path.open("wb") as fh: + for chunk in resp.iter_content(chunk_size=64 * 1024): + if not chunk: + continue + if self.shutdown_check and self.shutdown_check(): + shutdown_triggered = True + break + fh.write(chunk) + downloaded += len(chunk) + now = time.monotonic() + if self._engine and now - last_report >= 0.5: + self._engine.update_record( + "amazon", + download_id, + { + "transferred": downloaded, + "size": total, + "progress": downloaded / total if total else 0.0, + }, + ) + last_report = now + + if shutdown_triggered: + out_path.unlink(missing_ok=True) + raise RuntimeError("Shutdown requested mid-download") + + return downloaded + + # ------------------------------------------------------------------ + # DownloadSourcePlugin — status interface + # ------------------------------------------------------------------ + + async def get_all_downloads(self) -> List[DownloadStatus]: + if self._engine is None: + return [] + try: + records = self._engine.get_all_records("amazon") + return [self._record_to_status(dl_id, rec) for dl_id, rec in records.items()] + except Exception: + return [] + + async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]: + if self._engine is None: + return None + try: + rec = self._engine.get_record("amazon", download_id) + return self._record_to_status(download_id, rec) if rec is not None else None + except Exception: + return None + + async def cancel_download( + self, + download_id: str, + username: Optional[str] = None, + remove: bool = False, + ) -> bool: + if self._engine is None: + return False + try: + return self._engine.cancel_record("amazon", download_id, remove=remove) + except Exception: + return False + + async def clear_all_completed_downloads(self) -> bool: + if self._engine is None: + return False + try: + self._engine.clear_completed("amazon") + return True + except Exception: + return False + + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ + + @staticmethod + def _unique_path(path: Path) -> Path: + if not path.exists(): + return path + stem, suffix = path.stem, path.suffix + for i in range(1, 100): + candidate = path.with_name(f"{stem} ({i}){suffix}") + if not candidate.exists(): + return candidate + return path.with_name(f"{stem}_{uuid.uuid4().hex[:8]}{suffix}") + + @staticmethod + def _record_to_status(download_id: str, rec: Dict[str, Any]) -> DownloadStatus: + return DownloadStatus( + id=download_id, + filename=str(rec.get("original_filename", "")), + username="amazon", + state=str(rec.get("state", "queued")), + progress=float(rec.get("progress", 0.0)), + size=int(rec.get("size", 0)), + transferred=int(rec.get("transferred", 0)), + speed=int(rec.get("speed", 0)), + time_remaining=rec.get("time_remaining"), + file_path=rec.get("file_path"), + ) diff --git a/tests/tools/__init__.py b/tests/tools/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/tools/test_amazon_client.py b/tests/tools/test_amazon_client.py new file mode 100644 index 00000000..bdceea74 --- /dev/null +++ b/tests/tools/test_amazon_client.py @@ -0,0 +1,915 @@ +"""Unit tests for core/amazon_client.py. + +All network I/O is mocked via a fake session — no real T2Tunes instance needed. + +Run from project root: + python -m pytest tests/tools/test_amazon_client.py -v +""" + +from __future__ import annotations + +import json +import sys +import threading +from pathlib import Path +from typing import Any, Dict, Optional +from unittest.mock import MagicMock, patch + +import pytest + +# Make sure project root is importable when running tests directly. +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from core.amazon_client import ( + Album, + AmazonClient, + AmazonClientError, + Artist, + T2TunesSearchItem, + T2TunesStreamInfo, + Track, + _rate_limit, +) + + +# --------------------------------------------------------------------------- +# Shared fixtures +# --------------------------------------------------------------------------- + +TRACK_DOC = { + "asin": "B09XYZ1234", + "title": "Not Like Us", + "artistName": "Kendrick Lamar", + "__type": "track", + "albumName": "GNX", + "albumAsin": "B0ABCDE123", + "duration": 217, + "isrc": "USRC12345678", +} + +ALBUM_DOC = { + "asin": "B0ABCDE123", + "albumAsin": "B0ABCDE123", + "title": "GNX", + "albumName": "GNX", + "artistName": "Kendrick Lamar", + "__type": "album", + "duration": 0, +} + +SEARCH_RESPONSE_TRACKS = { + "results": [ + { + "hits": [ + {"document": TRACK_DOC}, + { + "document": { + "asin": "B09XYZ5678", + "title": "euphoria", + "artistName": "Kendrick Lamar", + "__type": "track", + "albumName": "euphoria", + "albumAsin": "B0ABCDE456", + "duration": 480, + "isrc": "USRC87654321", + } + }, + ] + } + ] +} + +SEARCH_RESPONSE_ALBUMS = { + "results": [{"hits": [{"document": ALBUM_DOC}]}] +} + +SEARCH_RESPONSE_MIXED = { + "results": [ + { + "hits": [ + {"document": TRACK_DOC}, + {"document": ALBUM_DOC}, + ] + } + ] +} + +ALBUM_METADATA_RESPONSE = { + "albumList": [ + { + "asin": "B0ABCDE123", + "title": "GNX", + "image": "https://example.com/cover.jpg", + "trackCount": 12, + "label": "pgLang/Interscope", + "artistName": "Kendrick Lamar", + } + ] +} + +MEDIA_RESPONSE_FLAC = { + "asin": "B09XYZ1234", + "streamable": True, + "decryptionKey": None, + "streamInfo": { + "codec": "FLAC", + "format": "FLAC", + "sampleRate": 44100, + "streamUrl": "https://cdn.example.com/track.flac", + }, + "tags": { + "title": "Not Like Us", + "artist": "Kendrick Lamar", + "album": "GNX", + "isrc": "USRC12345678", + }, +} + +MEDIA_RESPONSE_HIRES = { + "asin": "B09XYZ1234", + "streamable": True, + "decryptionKey": "somekey", + "streamInfo": { + "codec": "FLAC", + "format": "FLAC", + "sampleRate": 96000, + "streamUrl": "https://cdn.example.com/track-hires.flac", + }, + "tags": { + "title": "Not Like Us", + "artist": "Kendrick Lamar", + "album": "GNX", + "isrc": "USRC12345678", + }, +} + +STATUS_UP = {"amazonMusic": "up", "version": "1.0"} +STATUS_DOWN = {"amazonMusic": "down", "version": "1.0"} + + +def _mock_response(data: Any, status_code: int = 200) -> MagicMock: + resp = MagicMock() + resp.status_code = status_code + resp.ok = 200 <= status_code < 400 + resp.json.return_value = data + resp.text = json.dumps(data) + if status_code >= 400: + from requests import HTTPError + exc = HTTPError(response=resp) + resp.raise_for_status.side_effect = exc + else: + resp.raise_for_status.return_value = None + return resp + + +def _make_client(response_map: Optional[Dict[str, Any]] = None) -> AmazonClient: + """Build an AmazonClient with a fake session. + + response_map: path substring → response data (first match wins). + """ + session = MagicMock() + + def _get(url, params=None, timeout=None, **_): + if response_map: + for key, data in response_map.items(): + if key in url: + if isinstance(data, Exception): + raise data + return _mock_response(data) + return _mock_response({"error": "no mock for " + url}, 404) + + session.get.side_effect = _get + with patch("core.amazon_client._rate_limit"): + with patch("core.amazon_client.config_manager") as cfg: + cfg.get.return_value = "" + client = AmazonClient( + base_url="https://test.t2tunes.local", + country="US", + session=session, + ) + client.session = session + return client + + +# --------------------------------------------------------------------------- +# Dataclass construction +# --------------------------------------------------------------------------- + +class TestTrackDataclass: + def test_from_search_hit_basic(self): + t = Track.from_search_hit(TRACK_DOC) + assert t.id == "B09XYZ1234" + assert t.name == "Not Like Us" + assert t.artists == ["Kendrick Lamar"] + assert t.album == "GNX" + assert t.duration_ms == 217_000 + assert t.isrc == "USRC12345678" + assert t.popularity == 0 + + def test_from_search_hit_missing_fields(self): + t = Track.from_search_hit({}) + assert t.id == "" + assert t.name == "" + assert t.artists == ["Unknown Artist"] + assert t.duration_ms == 0 + assert t.isrc is None + + def test_from_stream_info(self): + stream = T2TunesStreamInfo( + asin="B09XYZ1234", + streamable=True, + codec="FLAC", + format="FLAC", + sample_rate=44100, + stream_url="https://cdn.example.com/track.flac", + decryption_key=None, + title="Not Like Us", + artist="Kendrick Lamar", + album="GNX", + isrc="USRC12345678", + ) + t = Track.from_stream_info(stream) + assert t.id == "B09XYZ1234" + assert t.name == "Not Like Us" + assert t.artists == ["Kendrick Lamar"] + assert t.isrc == "USRC12345678" + + def test_from_stream_info_empty_artist(self): + stream = T2TunesStreamInfo( + asin="B1", + streamable=True, + codec="FLAC", + format="FLAC", + sample_rate=44100, + stream_url="https://cdn.example.com/t.flac", + decryption_key=None, + ) + t = Track.from_stream_info(stream) + assert t.artists == ["Unknown Artist"] + + +class TestArtistDataclass: + def test_from_name(self): + a = Artist.from_name("Kendrick Lamar") + assert a.id == "kendrick_lamar" + assert a.name == "Kendrick Lamar" + assert a.genres == [] + assert a.followers == 0 + + def test_from_name_special_chars(self): + a = Artist.from_name("AC/DC") + assert "ac" in a.id + + +class TestAlbumDataclass: + def test_from_search_hit(self): + al = Album.from_search_hit(ALBUM_DOC) + assert al.id == "B0ABCDE123" + assert al.name == "GNX" + assert al.artists == ["Kendrick Lamar"] + assert al.album_type == "album" + + def test_from_search_hit_fallback_asin(self): + al = Album.from_search_hit({"asin": "B0001", "albumName": "Test", "artistName": "X"}) + assert al.id == "B0001" + + def test_from_metadata(self): + meta = ALBUM_METADATA_RESPONSE["albumList"][0] + al = Album.from_metadata(meta, asin="B0ABCDE123") + assert al.id == "B0ABCDE123" + assert al.name == "GNX" + assert al.total_tracks == 12 + assert al.image_url == "https://example.com/cover.jpg" + + +# --------------------------------------------------------------------------- +# T2TunesSearchItem helpers +# --------------------------------------------------------------------------- + +class TestT2TunesSearchItem: + def test_is_track(self): + item = T2TunesSearchItem( + asin="A1", title="T", artist_name="X", item_type="MusicTrack" + ) + assert item.is_track is True + assert item.is_album is False + + def test_is_album(self): + item = T2TunesSearchItem( + asin="A1", title="T", artist_name="X", item_type="MusicAlbum" + ) + assert item.is_album is True + assert item.is_track is False + + def test_ambiguous_type(self): + item = T2TunesSearchItem( + asin="A1", title="T", artist_name="X", item_type="Unknown" + ) + assert item.is_track is False + assert item.is_album is False + + +# --------------------------------------------------------------------------- +# _iter_search_items static method +# --------------------------------------------------------------------------- + +class TestIterSearchItems: + def test_parses_tracks(self): + items = list(AmazonClient._iter_search_items(SEARCH_RESPONSE_TRACKS)) + assert len(items) == 2 + assert items[0].asin == "B09XYZ1234" + assert items[0].title == "Not Like Us" + assert items[0].is_track + + def test_parses_albums(self): + items = list(AmazonClient._iter_search_items(SEARCH_RESPONSE_ALBUMS)) + assert len(items) == 1 + assert items[0].is_album + + def test_skips_missing_asin(self): + resp = {"results": [{"hits": [{"document": {"title": "No ASIN"}}]}]} + items = list(AmazonClient._iter_search_items(resp)) + assert items == [] + + def test_empty_results(self): + items = list(AmazonClient._iter_search_items({"results": []})) + assert items == [] + + def test_wrong_type_raises(self): + with pytest.raises(AmazonClientError): + list(AmazonClient._iter_search_items(["not", "a", "dict"])) + + def test_skips_malformed_hits(self): + resp = { + "results": [ + { + "hits": [ + "not_a_dict", + {"document": None}, + {"document": {"asin": "B1", "__type": "track", "title": "T", "artistName": "A"}}, + ] + } + ] + } + items = list(AmazonClient._iter_search_items(resp)) + assert len(items) == 1 + assert items[0].asin == "B1" + + +# --------------------------------------------------------------------------- +# _parse_stream_info static method +# --------------------------------------------------------------------------- + +class TestParseStreamInfo: + def test_flac_stream(self): + s = AmazonClient._parse_stream_info(MEDIA_RESPONSE_FLAC) + assert s.asin == "B09XYZ1234" + assert s.streamable is True + assert s.codec == "FLAC" + assert s.sample_rate == 44100 + assert s.stream_url == "https://cdn.example.com/track.flac" + assert s.has_decryption_key is False + assert s.title == "Not Like Us" + assert s.isrc == "USRC12345678" + + def test_hires_with_key(self): + s = AmazonClient._parse_stream_info(MEDIA_RESPONSE_HIRES) + assert s.sample_rate == 96000 + assert s.has_decryption_key is True + + def test_typo_stremeable(self): + data = { + "asin": "B1", + "stremeable": True, # typo variant + "streamInfo": {"codec": "OPUS", "format": "OPUS", "streamUrl": "https://x.com/t.opus"}, + "tags": {}, + } + s = AmazonClient._parse_stream_info(data) + assert s.streamable is True + assert s.codec == "OPUS" + + def test_missing_stream_info(self): + s = AmazonClient._parse_stream_info({"asin": "B1"}) + assert s.stream_url == "" + assert s.codec == "" + assert s.sample_rate is None + assert s.has_decryption_key is False + + +# --------------------------------------------------------------------------- +# AmazonClient — HTTP layer +# --------------------------------------------------------------------------- + +class TestStatus: + def test_success(self): + client = _make_client({"/api/status": STATUS_UP}) + with patch("core.amazon_client._rate_limit"): + result = client.status() + assert result["amazonMusic"] == "up" + + def test_http_error_raises(self): + client = _make_client() + client.session.get.side_effect = None + client.session.get.return_value = _mock_response({}, 503) + with pytest.raises(AmazonClientError, match="HTTP 503"): + client.status() + + def test_non_json_raises(self): + resp = MagicMock() + resp.raise_for_status.return_value = None + resp.json.side_effect = ValueError("not json") + resp.text = "error" + client = _make_client() + client.session.get.side_effect = None + client.session.get.return_value = resp + with pytest.raises(AmazonClientError, match="not JSON"): + client.status() + + +class TestIsAuthenticated: + def test_true_when_up(self): + client = _make_client({"/api/status": STATUS_UP}) + assert client.is_authenticated() is True + + def test_false_when_down(self): + client = _make_client({"/api/status": STATUS_DOWN}) + assert client.is_authenticated() is False + + def test_false_on_error(self): + from requests import RequestException + client = _make_client() + client.session.get.side_effect = RequestException("network error") + assert client.is_authenticated() is False + + +class TestReloadConfig: + def test_reloads_fields(self): + with patch("core.amazon_client.config_manager") as cfg: + cfg.get.side_effect = lambda key, default="": { + "amazon.base_url": "https://new.instance.local", + "amazon.country": "GB", + "amazon.preferred_codec": "opus", + }.get(key, default) + client = AmazonClient(session=MagicMock()) + client.reload_config() + assert "new.instance.local" in client.base_url + assert client.country == "GB" + assert client.preferred_codec == "opus" + + +# --------------------------------------------------------------------------- +# AmazonClient — search_raw +# --------------------------------------------------------------------------- + +class TestSearchRaw: + def test_returns_items(self): + client = _make_client({"amazon-music/search": SEARCH_RESPONSE_TRACKS}) + with patch("core.amazon_client._rate_limit"): + items = client.search_raw("Kendrick Lamar") + assert len(items) == 2 + + def test_passes_country(self): + client = _make_client({"amazon-music/search": SEARCH_RESPONSE_TRACKS}) + with patch("core.amazon_client._rate_limit"): + client.search_raw("test", types="track") + call_kwargs = client.session.get.call_args + assert "country" in str(call_kwargs) + + def test_network_error_raises(self): + from requests import RequestException + client = _make_client() + client.session.get.side_effect = RequestException("timeout") + with pytest.raises(AmazonClientError): + client.search_raw("test") + + +# --------------------------------------------------------------------------- +# AmazonClient — search_tracks / search_artists / search_albums +# --------------------------------------------------------------------------- + +class TestSearchTracks: + def test_returns_track_list(self): + client = _make_client({"amazon-music/search": SEARCH_RESPONSE_TRACKS}) + with patch("core.amazon_client._rate_limit"): + tracks = client.search_tracks("Kendrick Lamar") + assert len(tracks) == 2 + assert all(isinstance(t, Track) for t in tracks) + + def test_respects_limit(self): + client = _make_client({"amazon-music/search": SEARCH_RESPONSE_TRACKS}) + with patch("core.amazon_client._rate_limit"): + tracks = client.search_tracks("Kendrick Lamar", limit=1) + assert len(tracks) == 1 + + def test_ignores_album_hits(self): + client = _make_client({"amazon-music/search": SEARCH_RESPONSE_ALBUMS}) + with patch("core.amazon_client._rate_limit"): + tracks = client.search_tracks("GNX") + assert tracks == [] + + def test_track_fields(self): + client = _make_client({"amazon-music/search": SEARCH_RESPONSE_TRACKS}) + with patch("core.amazon_client._rate_limit"): + tracks = client.search_tracks("Kendrick") + t = tracks[0] + assert t.name == "Not Like Us" + assert t.artists == ["Kendrick Lamar"] + assert t.album == "GNX" + assert t.duration_ms == 217_000 + assert t.isrc == "USRC12345678" + + +class TestSearchArtists: + def test_returns_unique_artists(self): + client = _make_client({"amazon-music/search": SEARCH_RESPONSE_TRACKS}) + with patch("core.amazon_client._rate_limit"): + artists = client.search_artists("Kendrick") + # Both tracks are by Kendrick Lamar — should deduplicate + assert len(artists) == 1 + assert artists[0].name == "Kendrick Lamar" + + def test_returns_artist_dataclass(self): + client = _make_client({"amazon-music/search": SEARCH_RESPONSE_TRACKS}) + with patch("core.amazon_client._rate_limit"): + artists = client.search_artists("Kendrick") + assert isinstance(artists[0], Artist) + + def test_respects_limit(self): + resp = { + "results": [ + { + "hits": [ + { + "document": { + "asin": f"B{i}", + "title": f"Song {i}", + "artistName": f"Artist {i}", + "__type": "track", + } + } + for i in range(10) + ] + } + ] + } + client = _make_client({"amazon-music/search": resp}) + with patch("core.amazon_client._rate_limit"): + artists = client.search_artists("Various", limit=3) + assert len(artists) == 3 + + +class TestSearchAlbums: + def test_returns_albums(self): + client = _make_client({"amazon-music/search": SEARCH_RESPONSE_ALBUMS}) + with patch("core.amazon_client._rate_limit"): + albums = client.search_albums("GNX") + assert len(albums) == 1 + assert isinstance(albums[0], Album) + assert albums[0].id == "B0ABCDE123" + + def test_deduplicates_by_asin(self): + resp = { + "results": [ + { + "hits": [ + {"document": {**ALBUM_DOC}}, + {"document": {**ALBUM_DOC}}, # duplicate + ] + } + ] + } + client = _make_client({"amazon-music/search": resp}) + with patch("core.amazon_client._rate_limit"): + albums = client.search_albums("GNX") + assert len(albums) == 1 + + def test_ignores_track_hits(self): + client = _make_client({"amazon-music/search": SEARCH_RESPONSE_TRACKS}) + with patch("core.amazon_client._rate_limit"): + albums = client.search_albums("Kendrick") + assert albums == [] + + +# --------------------------------------------------------------------------- +# AmazonClient — album_metadata / media_from_asin +# --------------------------------------------------------------------------- + +class TestAlbumMetadata: + def test_returns_dict(self): + client = _make_client({"amazon-music/metadata": ALBUM_METADATA_RESPONSE}) + with patch("core.amazon_client._rate_limit"): + meta = client.album_metadata("B0ABCDE123") + assert "albumList" in meta + assert meta["albumList"][0]["title"] == "GNX" + + +class TestMediaFromAsin: + def test_list_response(self): + client = _make_client({"amazon-music/media-from-asin": [MEDIA_RESPONSE_FLAC]}) + with patch("core.amazon_client._rate_limit"): + streams = client.media_from_asin("B09XYZ1234") + assert len(streams) == 1 + assert isinstance(streams[0], T2TunesStreamInfo) + assert streams[0].codec == "FLAC" + + def test_single_dict_response(self): + client = _make_client({"amazon-music/media-from-asin": MEDIA_RESPONSE_FLAC}) + with patch("core.amazon_client._rate_limit"): + streams = client.media_from_asin("B09XYZ1234") + assert len(streams) == 1 + + def test_invalid_response_raises(self): + client = _make_client({"amazon-music/media-from-asin": "not a list or dict"}) + with pytest.raises(AmazonClientError, match="Unexpected media"): + client.media_from_asin("B09XYZ1234") + + def test_uses_preferred_codec(self): + client = _make_client({"amazon-music/media-from-asin": [MEDIA_RESPONSE_FLAC]}) + client.preferred_codec = "opus" + with patch("core.amazon_client._rate_limit"): + client.media_from_asin("B09XYZ1234") + call_kwargs = str(client.session.get.call_args) + assert "opus" in call_kwargs + + def test_codec_override(self): + client = _make_client({"amazon-music/media-from-asin": [MEDIA_RESPONSE_FLAC]}) + with patch("core.amazon_client._rate_limit"): + client.media_from_asin("B09XYZ1234", codec="eac3") + call_kwargs = str(client.session.get.call_args) + assert "eac3" in call_kwargs + + +# --------------------------------------------------------------------------- +# AmazonClient — higher-level get_* methods +# --------------------------------------------------------------------------- + +class TestGetTrackDetails: + def _client(self): + return _make_client({ + "amazon-music/media-from-asin": [MEDIA_RESPONSE_FLAC], + "amazon-music/metadata": ALBUM_METADATA_RESPONSE, + }) + + def test_returns_spotify_compat_dict(self): + client = self._client() + with patch("core.amazon_client._rate_limit"): + details = client.get_track_details("B09XYZ1234") + assert details is not None + assert details["id"] == "B09XYZ1234" + assert details["name"] == "Not Like Us" + assert "artists" in details + assert "album" in details + assert details["is_album_track"] is True + + def test_album_image_populated(self): + client = self._client() + with patch("core.amazon_client._rate_limit"): + details = client.get_track_details("B09XYZ1234") + assert details["album"]["images"][0]["url"] == "https://example.com/cover.jpg" + + def test_raw_data_present(self): + client = self._client() + with patch("core.amazon_client._rate_limit"): + details = client.get_track_details("B09XYZ1234") + assert "raw_data" in details + assert details["raw_data"]["codec"] == "FLAC" + assert details["raw_data"]["sample_rate"] == 44100 + + def test_returns_none_on_empty_streams(self): + client = _make_client({"amazon-music/media-from-asin": []}) + with patch("core.amazon_client._rate_limit"): + assert client.get_track_details("B09XYZ1234") is None + + def test_returns_none_on_api_error(self): + client = _make_client() + client.session.get.return_value = _mock_response({}, 500) + with patch("core.amazon_client._rate_limit"): + assert client.get_track_details("B09XYZ1234") is None + + def test_graceful_when_metadata_fails(self): + session = MagicMock() + call_count = {"n": 0} + + def _get(url, params=None, timeout=None, **_): + call_count["n"] += 1 + if "media-from-asin" in url: + return _mock_response([MEDIA_RESPONSE_FLAC]) + return _mock_response({}, 500) + + session.get.side_effect = _get + client = AmazonClient(base_url="https://test.local", session=session) + with patch("core.amazon_client._rate_limit"): + details = client.get_track_details("B09XYZ1234") + assert details is not None + assert details["album"]["images"] == [] + + +class TestGetAlbum: + def _client(self): + return _make_client({"amazon-music/metadata": ALBUM_METADATA_RESPONSE, + "amazon-music/media-from-asin": [MEDIA_RESPONSE_FLAC]}) + + def test_returns_album_dict(self): + client = self._client() + with patch("core.amazon_client._rate_limit"): + album = client.get_album("B0ABCDE123") + assert album is not None + assert album["id"] == "B0ABCDE123" + assert album["name"] == "GNX" + assert album["total_tracks"] == 12 + assert album["label"] == "pgLang/Interscope" + + def test_includes_tracks_by_default(self): + client = self._client() + with patch("core.amazon_client._rate_limit"): + album = client.get_album("B0ABCDE123") + assert "tracks" in album + assert isinstance(album["tracks"], dict) + assert "items" in album["tracks"] + + def test_excludes_tracks_when_flag_false(self): + client = _make_client({"amazon-music/metadata": ALBUM_METADATA_RESPONSE}) + with patch("core.amazon_client._rate_limit"): + album = client.get_album("B0ABCDE123", include_tracks=False) + assert "tracks" not in album + + def test_returns_none_on_empty_albumlist(self): + client = _make_client({"amazon-music/metadata": {"albumList": []}}) + with patch("core.amazon_client._rate_limit"): + assert client.get_album("B0ABCDE123") is None + + def test_returns_none_on_api_error(self): + client = _make_client() + client.session.get.return_value = _mock_response({}, 500) + with patch("core.amazon_client._rate_limit"): + assert client.get_album("B0ABCDE123") is None + + +class TestGetAlbumTracks: + def test_returns_spotify_pagination(self): + client = _make_client({"amazon-music/media-from-asin": [MEDIA_RESPONSE_FLAC]}) + with patch("core.amazon_client._rate_limit"): + result = client.get_album_tracks("B09XYZ1234") + assert result is not None + assert "items" in result + assert result["total"] == 1 + assert result["next"] is None + assert result["limit"] == 50 + + def test_item_fields(self): + client = _make_client({"amazon-music/media-from-asin": [MEDIA_RESPONSE_FLAC]}) + with patch("core.amazon_client._rate_limit"): + result = client.get_album_tracks("B09XYZ1234") + item = result["items"][0] + assert item["id"] == "B09XYZ1234" + assert item["name"] == "Not Like Us" + assert item["isrc"] == "USRC12345678" + + def test_returns_none_on_api_error(self): + client = _make_client() + client.session.get.return_value = _mock_response({}, 500) + with patch("core.amazon_client._rate_limit"): + assert client.get_album_tracks("B09XYZ1234") is None + + +class TestGetArtist: + def test_returns_artist_dict(self): + client = _make_client({"amazon-music/search": SEARCH_RESPONSE_TRACKS}) + with patch("core.amazon_client._rate_limit"): + artist = client.get_artist("Kendrick Lamar") + assert artist is not None + assert artist["name"] == "Kendrick Lamar" + assert "genres" in artist + assert "followers" in artist + + def test_exact_match_preferred(self): + resp = { + "results": [ + { + "hits": [ + {"document": {"asin": "A1", "title": "T1", "artistName": "Kendrick Lamar", "__type": "track"}}, + {"document": {"asin": "A2", "title": "T2", "artistName": "Kendrick Lamar Jr.", "__type": "track"}}, + ] + } + ] + } + client = _make_client({"amazon-music/search": resp}) + with patch("core.amazon_client._rate_limit"): + artist = client.get_artist("Kendrick Lamar") + assert artist["name"] == "Kendrick Lamar" + + def test_returns_none_when_no_match(self): + client = _make_client({"amazon-music/search": {"results": []}}) + with patch("core.amazon_client._rate_limit"): + assert client.get_artist("Nobody") is None + + def test_returns_none_on_error(self): + client = _make_client() + client.session.get.return_value = _mock_response({}, 500) + with patch("core.amazon_client._rate_limit"): + assert client.get_artist("Kendrick") is None + + +class TestGetArtistAlbums: + def test_returns_filtered_albums(self): + resp = { + "results": [ + { + "hits": [ + { + "document": { + "asin": "B0ABCDE123", + "albumAsin": "B0ABCDE123", + "title": "GNX", + "albumName": "GNX", + "artistName": "Kendrick Lamar", + "__type": "album", + } + }, + { + "document": { + "asin": "B0ZZZ", + "albumAsin": "B0ZZZ", + "title": "Other Album", + "albumName": "Other Album", + "artistName": "Another Artist", + "__type": "album", + } + }, + ] + } + ] + } + client = _make_client({"amazon-music/search": resp}) + with patch("core.amazon_client._rate_limit"): + albums = client.get_artist_albums("Kendrick Lamar") + assert len(albums) == 1 + assert albums[0].name == "GNX" + + def test_respects_limit(self): + hits = [ + { + "document": { + "asin": f"B{i}", + "albumAsin": f"B{i}", + "albumName": f"Album {i}", + "artistName": "Kendrick Lamar", + "__type": "album", + } + } + for i in range(20) + ] + client = _make_client({"amazon-music/search": {"results": [{"hits": hits}]}}) + with patch("core.amazon_client._rate_limit"): + albums = client.get_artist_albums("Kendrick Lamar", limit=5) + assert len(albums) == 5 + + def test_returns_empty_on_error(self): + client = _make_client() + client.session.get.return_value = _mock_response({}, 500) + with patch("core.amazon_client._rate_limit"): + assert client.get_artist_albums("Kendrick") == [] + + +class TestGetTrackFeatures: + def test_always_none(self): + client = AmazonClient(session=MagicMock()) + assert client.get_track_features("B09XYZ1234") is None + + +# --------------------------------------------------------------------------- +# Rate-limit enforcement +# --------------------------------------------------------------------------- + +class TestRateLimit: + def test_enforces_min_interval(self): + import core.amazon_client as mod + original = mod._last_api_call + sleeps = [] + + def fake_sleep(t): + sleeps.append(t) + + with patch("core.amazon_client.time") as mock_time: + mock_time.monotonic.return_value = mod._last_api_call + 0.1 + mock_time.sleep = fake_sleep + with patch("core.amazon_client.api_call_tracker"): + _rate_limit() + # Should have slept since interval not elapsed + assert len(sleeps) > 0 + mod._last_api_call = original + + def test_no_sleep_when_interval_elapsed(self): + import core.amazon_client as mod + original = mod._last_api_call + sleeps = [] + + with patch("core.amazon_client.time") as mock_time: + mock_time.monotonic.return_value = mod._last_api_call + 10.0 + mock_time.sleep = lambda t: sleeps.append(t) + with patch("core.amazon_client.api_call_tracker"): + _rate_limit() + assert sleeps == [] + mod._last_api_call = original diff --git a/tests/tools/test_amazon_download_client.py b/tests/tools/test_amazon_download_client.py new file mode 100644 index 00000000..c591f7fb --- /dev/null +++ b/tests/tools/test_amazon_download_client.py @@ -0,0 +1,783 @@ +"""Unit tests for core/amazon_download_client.py. + +All network I/O and subprocess calls are mocked. +No real T2Tunes instance or ffmpeg binary required. + +Run from project root: + python -m pytest tests/tools/test_amazon_download_client.py -v +""" + +from __future__ import annotations + +import asyncio +import io +import sys +import tempfile +from pathlib import Path +from typing import Any, Dict, Iterator, List, Optional +from unittest.mock import AsyncMock, MagicMock, call, patch + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from core.amazon_client import AmazonClientError, T2TunesStreamInfo +from core.amazon_download_client import ( + AmazonDownloadClient, + MIN_AUDIO_BYTES, + _codec_key, + _file_extension, + _quality_label, +) +from core.download_plugins.types import AlbumResult, DownloadStatus, TrackResult + + +# --------------------------------------------------------------------------- +# Helpers / fixtures +# --------------------------------------------------------------------------- + +def run(coro): + return asyncio.get_event_loop().run_until_complete(coro) + + +def _stream_info( + *, + asin: str = "B09XYZ1234", + streamable: bool = True, + codec: str = "FLAC", + sample_rate: int = 44100, + stream_url: str = "https://cdn.example.com/track.enc.flac", + decryption_key: Optional[str] = "deadbeef1234", +) -> T2TunesStreamInfo: + return T2TunesStreamInfo( + asin=asin, + streamable=streamable, + codec=codec, + format=codec, + sample_rate=sample_rate, + stream_url=stream_url, + decryption_key=decryption_key, + title="Not Like Us", + artist="Kendrick Lamar", + album="GNX", + isrc="USRC12345678", + ) + + +def _search_items(n_tracks: int = 2, n_albums: int = 1): + from core.amazon_client import T2TunesSearchItem + items = [ + T2TunesSearchItem( + asin=f"B0TRACK{i}", + title=f"Track {i}", + artist_name="Kendrick Lamar", + item_type="MusicTrack", + album_name="GNX", + album_asin="B0ALBUM1", + duration_seconds=200 + i * 10, + isrc=f"USRC{i:08d}", + ) + for i in range(n_tracks) + ] + items += [ + T2TunesSearchItem( + asin=f"B0ALBUM{j}", + title=f"Album {j}", + artist_name="Kendrick Lamar", + item_type="MusicAlbum", + album_name=f"Album {j}", + album_asin=f"B0ALBUM{j}", + duration_seconds=0, + ) + for j in range(n_albums) + ] + return items + + +def _make_client(tmp_path: Path) -> AmazonDownloadClient: + with patch("core.amazon_download_client.config_manager") as cfg: + cfg.get.return_value = str(tmp_path) + with patch("core.amazon_client.config_manager") as cfg2: + cfg2.get.return_value = "" + client = AmazonDownloadClient(download_path=str(tmp_path)) + return client + + +def _fake_chunked_response(data: bytes, status_code: int = 200) -> MagicMock: + resp = MagicMock() + resp.status_code = status_code + resp.ok = status_code < 400 + resp.headers = {"content-length": str(len(data))} + + chunk_size = 4096 + chunks = [data[i : i + chunk_size] for i in range(0, len(data), chunk_size)] or [b""] + + def iter_content(chunk_size=None): + yield from chunks + + resp.iter_content = iter_content + if status_code >= 400: + from requests import HTTPError + resp.raise_for_status.side_effect = HTTPError(response=resp) + else: + resp.raise_for_status.return_value = None + return resp + + +# --------------------------------------------------------------------------- +# Codec / quality helpers +# --------------------------------------------------------------------------- + +class TestCodecHelpers: + def test_codec_key_lowercases(self): + assert _codec_key("FLAC") == "flac" + assert _codec_key("OGG-Vorbis") == "ogg_vorbis" + assert _codec_key("EAC3") == "eac3" + + def test_file_extension_known_codecs(self): + assert _file_extension("FLAC") == "flac" + assert _file_extension("ogg_vorbis") == "ogg" + assert _file_extension("opus") == "opus" + assert _file_extension("eac3") == "eac3" + assert _file_extension("mp4") == "m4a" + assert _file_extension("aac") == "m4a" + assert _file_extension("mp3") == "mp3" + + def test_file_extension_unknown_falls_back(self): + assert _file_extension("wtf_codec") == "bin" + + def test_quality_label_flac_lossless(self): + assert _quality_label("flac", 44100) == "Lossless" + assert _quality_label("FLAC", 48000) == "Lossless" + + def test_quality_label_flac_hires(self): + assert _quality_label("flac", 96000) == "Hi-Res" + assert _quality_label("flac", 192000) == "Hi-Res" + + def test_quality_label_lossy(self): + assert _quality_label("opus") == "Lossy" + assert _quality_label("eac3") == "Lossy" + assert _quality_label("mp3") == "Lossy" + + +# --------------------------------------------------------------------------- +# is_configured / check_connection +# --------------------------------------------------------------------------- + +class TestIsConfigured: + def test_always_true(self, tmp_path): + client = _make_client(tmp_path) + assert client.is_configured() is True + + +class TestCheckConnection: + def test_true_when_client_authenticated(self, tmp_path): + client = _make_client(tmp_path) + client._client = MagicMock() + client._client.is_authenticated.return_value = True + assert run(client.check_connection()) is True + + def test_false_when_client_down(self, tmp_path): + client = _make_client(tmp_path) + client._client = MagicMock() + client._client.is_authenticated.return_value = False + assert run(client.check_connection()) is False + + def test_false_on_exception(self, tmp_path): + client = _make_client(tmp_path) + client._client = MagicMock() + client._client.is_authenticated.side_effect = Exception("timeout") + assert run(client.check_connection()) is False + + +# --------------------------------------------------------------------------- +# search() +# --------------------------------------------------------------------------- + +class TestSearch: + def test_returns_track_results(self, tmp_path): + client = _make_client(tmp_path) + client._client = MagicMock() + client._client.search_raw.return_value = _search_items(n_tracks=2, n_albums=0) + client._client.preferred_codec = "flac" + + tracks, albums = run(client.search("Kendrick Lamar")) + + assert len(tracks) == 2 + assert len(albums) == 0 + assert all(isinstance(t, TrackResult) for t in tracks) + + def test_track_fields(self, tmp_path): + client = _make_client(tmp_path) + client._client = MagicMock() + client._client.search_raw.return_value = _search_items(n_tracks=1, n_albums=0) + client._client.preferred_codec = "flac" + + tracks, _ = run(client.search("Not Like Us")) + t = tracks[0] + + assert t.username == "amazon" + assert "B0TRACK0" in t.filename + assert "||" in t.filename + assert t.artist == "Kendrick Lamar" + assert t.title == "Track 0" + assert t.album == "GNX" + assert t.quality == "Lossless" + assert t.duration == 200_000 + + def test_track_source_metadata(self, tmp_path): + client = _make_client(tmp_path) + client._client = MagicMock() + client._client.search_raw.return_value = _search_items(n_tracks=1, n_albums=0) + client._client.preferred_codec = "flac" + + tracks, _ = run(client.search("test")) + meta = tracks[0]._source_metadata + + assert meta["asin"] == "B0TRACK0" + assert meta["album_asin"] == "B0ALBUM1" + assert "isrc" in meta + + def test_returns_album_results(self, tmp_path): + client = _make_client(tmp_path) + client._client = MagicMock() + client._client.search_raw.return_value = _search_items(n_tracks=0, n_albums=2) + client._client.preferred_codec = "flac" + + _, albums = run(client.search("GNX")) + + assert len(albums) == 2 + assert all(isinstance(a, AlbumResult) for a in albums) + + def test_album_deduplication(self, tmp_path): + from core.amazon_client import T2TunesSearchItem + client = _make_client(tmp_path) + client._client = MagicMock() + client._client.preferred_codec = "flac" + # Two hits with same album_asin + dup = T2TunesSearchItem( + asin="B0ALBUM0", + title="GNX", + artist_name="Kendrick Lamar", + item_type="MusicAlbum", + album_asin="B0ALBUM0", + ) + client._client.search_raw.return_value = [dup, dup] + + _, albums = run(client.search("GNX")) + assert len(albums) == 1 + + def test_returns_empty_on_error(self, tmp_path): + client = _make_client(tmp_path) + client._client = MagicMock() + client._client.search_raw.side_effect = AmazonClientError("fail") + + tracks, albums = run(client.search("anything")) + assert tracks == [] + assert albums == [] + + def test_soulseek_compat_fields(self, tmp_path): + client = _make_client(tmp_path) + client._client = MagicMock() + client._client.search_raw.return_value = _search_items(n_tracks=1, n_albums=0) + client._client.preferred_codec = "flac" + + tracks, _ = run(client.search("test")) + t = tracks[0] + + assert t.free_upload_slots == 999 + assert t.upload_speed == 999_999 + assert t.queue_length == 0 + assert t.size == 0 + + +# --------------------------------------------------------------------------- +# _unique_path +# --------------------------------------------------------------------------- + +class TestUniquePath: + def test_returns_original_when_no_conflict(self, tmp_path): + p = tmp_path / "track.flac" + result = AmazonDownloadClient._unique_path(p) + assert result == p + + def test_appends_counter_on_conflict(self, tmp_path): + p = tmp_path / "track.flac" + p.touch() + result = AmazonDownloadClient._unique_path(p) + assert result != p + assert "(1)" in result.name + + def test_increments_counter(self, tmp_path): + p = tmp_path / "track.flac" + p.touch() + (tmp_path / "track (1).flac").touch() + result = AmazonDownloadClient._unique_path(p) + assert "(2)" in result.name + + +# --------------------------------------------------------------------------- +# _record_to_status +# --------------------------------------------------------------------------- + +class TestRecordToStatus: + def test_fields_mapped(self): + rec = { + "original_filename": "B1||Artist - Title", + "state": "downloading", + "progress": 0.5, + "size": 10_000_000, + "transferred": 5_000_000, + "speed": 1_000_000, + "time_remaining": 5, + "file_path": "/tmp/track.flac", + } + status = AmazonDownloadClient._record_to_status("dl-001", rec) + + assert status.id == "dl-001" + assert status.filename == "B1||Artist - Title" + assert status.username == "amazon" + assert status.state == "downloading" + assert status.progress == 0.5 + assert status.size == 10_000_000 + assert status.transferred == 5_000_000 + assert status.speed == 1_000_000 + assert status.time_remaining == 5 + assert status.file_path == "/tmp/track.flac" + + def test_defaults_for_missing_fields(self): + status = AmazonDownloadClient._record_to_status("dl-002", {}) + assert status.state == "queued" + assert status.progress == 0.0 + assert status.size == 0 + assert status.transferred == 0 + assert status.speed == 0 + assert status.time_remaining is None + assert status.file_path is None + + +# --------------------------------------------------------------------------- +# _stream_to_file +# --------------------------------------------------------------------------- + +class TestStreamToFile: + def test_writes_file_and_returns_size(self, tmp_path): + client = _make_client(tmp_path) + data = b"X" * (MIN_AUDIO_BYTES + 1024) + client.session = MagicMock() + client.session.get.return_value = _fake_chunked_response(data) + + out = tmp_path / "output.flac" + downloaded = client._stream_to_file("https://example.com/t.flac", out, "dl-001") + + assert downloaded == len(data) + assert out.exists() + assert out.read_bytes() == data + + def test_raises_on_http_error(self, tmp_path): + from requests import HTTPError + client = _make_client(tmp_path) + client.session = MagicMock() + client.session.get.return_value = _fake_chunked_response(b"", status_code=403) + + out = tmp_path / "output.flac" + with pytest.raises(HTTPError): + client._stream_to_file("https://example.com/t.flac", out, "dl-001") + + def test_respects_shutdown_check(self, tmp_path): + client = _make_client(tmp_path) + data = b"X" * (MIN_AUDIO_BYTES + 1024) + client.session = MagicMock() + client.session.get.return_value = _fake_chunked_response(data) + client.shutdown_check = lambda: True # trigger immediately + + out = tmp_path / "output.flac" + with pytest.raises(RuntimeError, match="Shutdown"): + client._stream_to_file("https://example.com/t.flac", out, "dl-001") + assert not out.exists() + + def test_updates_engine_progress(self, tmp_path): + import itertools + client = _make_client(tmp_path) + data = b"X" * (MIN_AUDIO_BYTES + 1024) + client.session = MagicMock() + client.session.get.return_value = _fake_chunked_response(data) + engine = MagicMock() + client._engine = engine + + out = tmp_path / "output.flac" + counter = itertools.count(0.0, 1.0) + with patch("core.amazon_download_client.time") as mock_time: + mock_time.monotonic.side_effect = lambda: next(counter) + client._stream_to_file("https://example.com/t.flac", out, "dl-001") + + assert engine.update_record.called + + +# --------------------------------------------------------------------------- +# _decrypt_with_ffmpeg +# --------------------------------------------------------------------------- + +class TestDecryptWithFfmpeg: + def test_calls_ffmpeg_with_key(self, tmp_path): + client = _make_client(tmp_path) + enc = tmp_path / "track.enc.flac" + enc.write_bytes(b"encrypted") + out = tmp_path / "track.flac" + + with patch("shutil.which", return_value="/usr/bin/ffmpeg"): + with patch("subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stderr=b"") + client._decrypt_with_ffmpeg(enc, out, "deadbeef1234") + + cmd = mock_run.call_args[0][0] + assert "ffmpeg" in cmd[0] + assert "-decryption_key" in cmd + assert "deadbeef1234" in cmd + assert str(enc) in cmd + assert str(out) in cmd + + def test_raises_on_ffmpeg_failure(self, tmp_path): + client = _make_client(tmp_path) + enc = tmp_path / "track.enc.flac" + enc.write_bytes(b"bad") + out = tmp_path / "track.flac" + + with patch("shutil.which", return_value="/usr/bin/ffmpeg"): + with patch("subprocess.run") as mock_run: + mock_run.return_value = MagicMock( + returncode=1, stderr=b"Invalid data found" + ) + with pytest.raises(RuntimeError, match="FFmpeg decryption failed"): + client._decrypt_with_ffmpeg(enc, out, "deadbeef1234") + + def test_raises_when_ffmpeg_missing(self, tmp_path): + client = _make_client(tmp_path) + enc = tmp_path / "track.enc.flac" + out = tmp_path / "track.flac" + + with patch("shutil.which", return_value=None): + # Ensure tools/ffmpeg.exe also absent + with pytest.raises(RuntimeError, match="ffmpeg is required"): + client._decrypt_with_ffmpeg(enc, out, "deadbeef1234") + + def test_uses_tools_ffmpeg_when_not_on_path(self, tmp_path): + client = _make_client(tmp_path) + enc = tmp_path / "track.enc.flac" + enc.write_bytes(b"enc") + out = tmp_path / "track.flac" + + fake_ffmpeg = tmp_path / "ffmpeg.exe" + fake_ffmpeg.touch() + + with patch("shutil.which", return_value=None): + with patch( + "core.amazon_download_client.Path.__file__", + create=True, + ): + import os as _os + is_nt = _os.name == "nt" + ffmpeg_name = "ffmpeg.exe" if is_nt else "ffmpeg" + tools_dir = ROOT / "tools" + tools_ffmpeg = tools_dir / ffmpeg_name + if tools_ffmpeg.exists(): + with patch("subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stderr=b"") + client._decrypt_with_ffmpeg(enc, out, "aabbcc") + assert str(tools_ffmpeg) in mock_run.call_args[0][0][0] + + +# --------------------------------------------------------------------------- +# _download_sync — integration of stream + decrypt +# --------------------------------------------------------------------------- + +class TestDownloadSync: + def _setup(self, tmp_path: Path, decryption_key: Optional[str] = "deadbeef"): + client = _make_client(tmp_path) + stream = _stream_info(decryption_key=decryption_key) + client._client = MagicMock() + client._client.media_from_asin.return_value = [stream] + client._client.preferred_codec = "flac" + + audio_data = b"A" * (MIN_AUDIO_BYTES + 1024) + client.session = MagicMock() + client.session.get.return_value = _fake_chunked_response(audio_data) + return client, audio_data + + def test_returns_output_path_on_success(self, tmp_path): + client, audio_data = self._setup(tmp_path) + + with patch("shutil.which", return_value="/usr/bin/ffmpeg"): + with patch("subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stderr=b"") + # simulate ffmpeg writing output file + def _ffmpeg_side_effect(cmd, capture_output=False): + # Write dummy decrypted data + out_path = Path(cmd[-1]) + out_path.write_bytes(audio_data) + return MagicMock(returncode=0, stderr=b"") + + mock_run.side_effect = _ffmpeg_side_effect + result = client._download_sync("dl-001", "B09XYZ1234", "Kendrick Lamar - Not Like Us") + + assert result is not None + assert Path(result).exists() + assert Path(result).suffix == ".flac" + + def test_tries_next_codec_when_stream_unavailable(self, tmp_path): + client = _make_client(tmp_path) + # FLAC: no streamable results; Opus: success + flac_stream = _stream_info(streamable=False, codec="FLAC") + opus_stream = _stream_info(streamable=True, codec="OPUS", + stream_url="https://cdn.example.com/t.opus", + decryption_key=None) + client._client = MagicMock() + client._client.preferred_codec = "flac" + + def _media(asin, codec): + if codec == "flac": + return [flac_stream] + if codec == "opus": + return [opus_stream] + return [] + + client._client.media_from_asin.side_effect = _media + + audio_data = b"B" * (MIN_AUDIO_BYTES + 1024) + client.session = MagicMock() + client.session.get.return_value = _fake_chunked_response(audio_data) + + result = client._download_sync("dl-001", "B09XYZ1234", "Kendrick - Track") + + assert result is not None + assert ".opus" in result + + def test_returns_none_when_file_too_small(self, tmp_path): + client = _make_client(tmp_path) + client._client = MagicMock() + client._client.preferred_codec = "flac" + client._client.media_from_asin.return_value = [ + _stream_info(decryption_key=None) + ] + tiny_data = b"X" * 100 # way below MIN_AUDIO_BYTES + client.session = MagicMock() + client.session.get.return_value = _fake_chunked_response(tiny_data) + + result = client._download_sync("dl-001", "B09XYZ1234", "Artist - Title") + assert result is None + + def test_tries_next_codec_when_media_fails(self, tmp_path): + client = _make_client(tmp_path) + client._client = MagicMock() + client._client.preferred_codec = "flac" + + call_count = {"n": 0} + + def _media(asin, codec): + call_count["n"] += 1 + if call_count["n"] == 1: + raise AmazonClientError("quota exceeded") + stream = _stream_info(codec="OPUS", decryption_key=None) + return [stream] + + client._client.media_from_asin.side_effect = _media + + audio_data = b"C" * (MIN_AUDIO_BYTES + 1024) + client.session = MagicMock() + client.session.get.return_value = _fake_chunked_response(audio_data) + + result = client._download_sync("dl-001", "B09XYZ1234", "Artist - Track") + assert result is not None + + def test_decryption_failure_tries_next_codec(self, tmp_path): + client, audio_data = self._setup(tmp_path, decryption_key="badkey") + + with patch("shutil.which", return_value="/usr/bin/ffmpeg"): + with patch("subprocess.run") as mock_run: + # First codec (flac) decryption fails; no more codecs succeed + mock_run.return_value = MagicMock(returncode=1, stderr=b"decrypt error") + result = client._download_sync("dl-001", "B09XYZ1234", "Artist - Track") + + # All codecs should fail since we return the same bad result for all + assert result is None + + def test_updates_engine_state(self, tmp_path): + client, audio_data = self._setup(tmp_path, decryption_key=None) + engine = MagicMock() + client._engine = engine + + result = client._download_sync("dl-001", "B09XYZ1234", "Artist - Track") + + update_calls = engine.update_record.call_args_list + states = [c[0][2].get("state") for c in update_calls if "state" in c[0][2]] + assert "downloading" in states + + def test_clear_stream_skips_ffmpeg(self, tmp_path): + client, audio_data = self._setup(tmp_path, decryption_key=None) + + with patch("subprocess.run") as mock_run: + result = client._download_sync("dl-001", "B09XYZ1234", "Artist - Track") + + mock_run.assert_not_called() + assert result is not None + + def test_safe_filename_sanitisation(self, tmp_path): + client, audio_data = self._setup(tmp_path, decryption_key=None) + result = client._download_sync( + "dl-001", "B09XYZ1234", "Björk / Sigur Rós: Hvarf<>Heim" + ) + assert result is not None + # Path must not contain illegal filesystem chars + path = Path(result) + assert "/" not in path.name + assert "<" not in path.name + assert ">" not in path.name + + +# --------------------------------------------------------------------------- +# download() — async dispatch +# --------------------------------------------------------------------------- + +class TestDownloadDispatch: + def test_raises_without_engine(self, tmp_path): + client = _make_client(tmp_path) + with pytest.raises(RuntimeError, match="_engine"): + run(client.download("amazon", "B09XYZ1234||Artist - Title")) + + def test_returns_none_on_bad_filename(self, tmp_path): + client = _make_client(tmp_path) + client._engine = MagicMock() + result = run(client.download("amazon", "no-pipe-delimiter-here")) + assert result is None + + def test_dispatches_to_engine_worker(self, tmp_path): + client = _make_client(tmp_path) + engine = MagicMock() + engine.worker.dispatch.return_value = "dl-abc123" + client._engine = engine + + result = run(client.download("amazon", "B09XYZ1234||Kendrick Lamar - Not Like Us")) + + assert result == "dl-abc123" + dispatch_call = engine.worker.dispatch.call_args + assert dispatch_call[1]["source_name"] == "amazon" + assert dispatch_call[1]["target_id"] == "B09XYZ1234" + assert dispatch_call[1]["display_name"] == "Kendrick Lamar - Not Like Us" + assert dispatch_call[1]["impl_callable"] == client._download_sync + + def test_strips_whitespace_from_asin_and_name(self, tmp_path): + client = _make_client(tmp_path) + engine = MagicMock() + engine.worker.dispatch.return_value = "dl-xyz" + client._engine = engine + + run(client.download("amazon", " B09XYZ1234 || Artist - Title ")) + + call = engine.worker.dispatch.call_args + assert call[1]["target_id"] == "B09XYZ1234" + assert call[1]["display_name"] == "Artist - Title" + + +# --------------------------------------------------------------------------- +# Status interface +# --------------------------------------------------------------------------- + +class TestStatusInterface: + def test_get_all_downloads_empty_without_engine(self, tmp_path): + client = _make_client(tmp_path) + result = run(client.get_all_downloads()) + assert result == [] + + def test_get_all_downloads_converts_records(self, tmp_path): + client = _make_client(tmp_path) + engine = MagicMock() + engine.get_all_records.return_value = { + "dl-001": { + "original_filename": "B1||A - T", + "state": "complete", + "progress": 1.0, + "size": 5_000_000, + "transferred": 5_000_000, + "speed": 0, + } + } + client._engine = engine + + statuses = run(client.get_all_downloads()) + assert len(statuses) == 1 + assert statuses[0].id == "dl-001" + assert statuses[0].state == "complete" + + def test_get_download_status_returns_none_without_engine(self, tmp_path): + client = _make_client(tmp_path) + assert run(client.get_download_status("dl-001")) is None + + def test_get_download_status_hit(self, tmp_path): + client = _make_client(tmp_path) + engine = MagicMock() + engine.get_record.return_value = { + "original_filename": "B1||A - T", + "state": "downloading", + "progress": 0.7, + "size": 10_000_000, + "transferred": 7_000_000, + "speed": 500_000, + } + client._engine = engine + + status = run(client.get_download_status("dl-001")) + assert status is not None + assert status.state == "downloading" + assert status.progress == 0.7 + engine.get_record.assert_called_once_with("amazon", "dl-001") + + def test_get_download_status_miss(self, tmp_path): + client = _make_client(tmp_path) + engine = MagicMock() + engine.get_record.return_value = None + client._engine = engine + + assert run(client.get_download_status("nonexistent")) is None + + def test_cancel_returns_false_without_engine(self, tmp_path): + client = _make_client(tmp_path) + assert run(client.cancel_download("dl-001")) is False + + def test_cancel_delegates_to_engine(self, tmp_path): + client = _make_client(tmp_path) + engine = MagicMock() + engine.cancel_record.return_value = True + client._engine = engine + + result = run(client.cancel_download("dl-001", remove=True)) + assert result is True + engine.cancel_record.assert_called_once_with("amazon", "dl-001", remove=True) + + def test_clear_completed_returns_false_without_engine(self, tmp_path): + client = _make_client(tmp_path) + assert run(client.clear_all_completed_downloads()) is False + + def test_clear_completed_delegates_to_engine(self, tmp_path): + client = _make_client(tmp_path) + engine = MagicMock() + client._engine = engine + + result = run(client.clear_all_completed_downloads()) + assert result is True + engine.clear_completed.assert_called_once_with("amazon") + + def test_status_methods_return_gracefully_on_engine_error(self, tmp_path): + client = _make_client(tmp_path) + engine = MagicMock() + engine.get_all_records.side_effect = Exception("engine crash") + engine.get_record.side_effect = Exception("engine crash") + engine.cancel_record.side_effect = Exception("engine crash") + engine.clear_completed.side_effect = Exception("engine crash") + client._engine = engine + + assert run(client.get_all_downloads()) == [] + assert run(client.get_download_status("dl-001")) is None + assert run(client.cancel_download("dl-001")) is False + assert run(client.clear_all_completed_downloads()) is False diff --git a/tools/t2tunes_media_plan.py b/tools/t2tunes_media_plan.py new file mode 100644 index 00000000..531e2f89 --- /dev/null +++ b/tools/t2tunes_media_plan.py @@ -0,0 +1,199 @@ +"""Build a T2Tunes media plan for a track without downloading media. + +This runner calls ``tools.t2tunes_probe.T2TunesClient`` and validates the +parts of a future download source that are safe to exercise live: + +- search resolution +- album metadata lookup +- cover URL discovery + HEAD probe +- per-codec media lookup +- stream URL HEAD/range probe + +It writes a JSON plan that a developer can inspect while building an +integration. It intentionally does not download or decrypt protected media. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional + +TOOL_DIR = Path(__file__).resolve().parent +if str(TOOL_DIR) not in sys.path: + sys.path.insert(0, str(TOOL_DIR)) + +from t2tunes_probe import T2TunesClient, T2TunesSearchItem # noqa: E402 + + +def _slug(value: str) -> str: + cleaned = "".join(ch.lower() if ch.isalnum() else "-" for ch in value) + parts = [part for part in cleaned.split("-") if part] + return "-".join(parts[:12]) or "t2tunes" + + +def _item_score(item: T2TunesSearchItem, query: str, explicit: Optional[bool]) -> int: + text = f"{item.artist_name} {item.title} {item.album_name}".lower() + score = 0 + for token in query.lower().split(): + if token in text: + score += 10 + if item.is_track: + score += 20 + if explicit is True and "explicit" in text: + score += 50 + if explicit is False and "clean" in text: + score += 50 + if explicit is True and "clean" in text: + score -= 40 + if explicit is False and "explicit" in text: + score -= 40 + return score + + +def _choose_result( + items: Iterable[T2TunesSearchItem], + *, + query: str, + asin: Optional[str], + explicit: Optional[bool], +) -> Optional[T2TunesSearchItem]: + items = list(items) + if asin: + return next((item for item in items if item.asin == asin or item.album_asin == asin), None) + if not items: + return None + return max(items, key=lambda item: _item_score(item, query, explicit)) + + +def _first_album(metadata: Dict[str, Any]) -> Dict[str, Any]: + albums = metadata.get("albumList") + if isinstance(albums, list) and albums: + album = albums[0] + return album if isinstance(album, dict) else {} + return {} + + +def _cover_url_from_metadata(metadata: Dict[str, Any]) -> str: + album = _first_album(metadata) + image = album.get("image") + return image if isinstance(image, str) else "" + + +def _head_url(client: T2TunesClient, url: str) -> Dict[str, Any]: + if not url: + return {"ok": False, "reason": "missing url"} + return client.probe_stream(url) + + +def build_plan( + *, + query: str, + base_url: str, + country: str, + codecs: List[str], + asin: Optional[str], + explicit: Optional[bool], + timeout: int, + probe_urls: bool, +) -> Dict[str, Any]: + client = T2TunesClient(base_url, country=country, timeout=timeout) + status = client.status() + search_items = client.search(query) + selected = _choose_result(search_items, query=query, asin=asin, explicit=explicit) + + plan: Dict[str, Any] = { + "base_url": base_url, + "country": country, + "query": query, + "status": status, + "result_count": len(search_items), + "selected": selected.__dict__ if selected else None, + "album_metadata": None, + "cover": None, + "formats": [], + "notes": [ + "This plan validates API behavior only.", + "It intentionally does not download or decrypt protected media.", + ], + } + + if not selected: + return plan + + album_asin = selected.album_asin or selected.asin + metadata = client.album_metadata(album_asin) + cover_url = _cover_url_from_metadata(metadata) + plan["album_metadata"] = { + "asin": album_asin, + "title": _first_album(metadata).get("title"), + "track_count": _first_album(metadata).get("trackCount"), + "image": cover_url, + "label": _first_album(metadata).get("label"), + } + plan["cover"] = { + "url": cover_url, + "probe": _head_url(client, cover_url) if probe_urls else None, + } + + for codec in codecs: + format_client = T2TunesClient(base_url, country=country, codec=codec, timeout=timeout) + streams = format_client.media_from_asin(selected.asin) + entries = [] + for stream in streams: + entry = stream.__dict__.copy() + entry["stream_probe"] = _head_url(format_client, stream.stream_url) if probe_urls else None + entries.append(entry) + plan["formats"].append({ + "requested_codec": codec, + "stream_count": len(entries), + "streams": entries, + }) + + return plan + + +def main() -> int: + parser = argparse.ArgumentParser(description="Create a safe T2Tunes media plan for a track.") + parser.add_argument("query", help="Search query, e.g. 'Kendrick Lamar Not Like Us'") + parser.add_argument("--base-url", default="https://t2tunes.site") + parser.add_argument("--country", default="US") + parser.add_argument("--codec", action="append", dest="codecs", choices=("flac", "opus", "eac3")) + parser.add_argument("--asin", help="Prefer a specific track or album ASIN from search results") + group = parser.add_mutually_exclusive_group() + group.add_argument("--explicit", action="store_true", help="Prefer explicit search results") + group.add_argument("--clean", action="store_true", help="Prefer clean search results") + parser.add_argument("--timeout", type=int, default=30) + parser.add_argument("--no-probe", action="store_true", help="Do not HEAD-probe cover/stream URLs") + parser.add_argument( + "--output", + help="Optional JSON output path. Defaults to ./testTEST/-plan.json", + ) + args = parser.parse_args() + + explicit = True if args.explicit else False if args.clean else None + plan = build_plan( + query=args.query, + base_url=args.base_url, + country=args.country, + codecs=args.codecs or ["flac", "opus", "eac3"], + asin=args.asin, + explicit=explicit, + timeout=args.timeout, + probe_urls=not args.no_probe, + ) + + text = json.dumps(plan, indent=2, sort_keys=True) + output_path = Path(args.output) if args.output else Path.cwd() / "testTEST" / f"{_slug(args.query)}-plan.json" + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(text + "\n", encoding="utf-8") + plan["_written_to"] = str(output_path) + text = json.dumps(plan, indent=2, sort_keys=True) + print(text) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/t2tunes_probe.py b/tools/t2tunes_probe.py new file mode 100644 index 00000000..95e18c09 --- /dev/null +++ b/tools/t2tunes_probe.py @@ -0,0 +1,369 @@ +"""Standalone T2Tunes/TripleTriple API probe. + +This is intentionally not wired into SoulSync's download source registry. +It validates the API surface Tubifarry targets: + + /api/status + /api/amazon-music/search + /api/amazon-music/metadata + /api/amazon-music/media-from-asin + +The probe can inspect returned stream metadata and optionally issue a HEAD +request against the stream URL. It does not download or decrypt audio. +""" + +from __future__ import annotations + +import argparse +import json +from dataclasses import dataclass +from typing import Any, Dict, Iterable, List, Optional +from urllib.error import HTTPError, URLError +from urllib.parse import urlencode, urljoin +from urllib.request import Request, urlopen + + +DEFAULT_BASE_URL = "https://t2tunes.site" +DEFAULT_TIMEOUT_SECONDS = 30 + + +class T2TunesError(RuntimeError): + """Raised when the T2Tunes API returns an invalid or failed response.""" + + +class _HttpResponse: + def __init__(self, *, body: bytes, status_code: int, headers: Dict[str, str], url: str) -> None: + self.body = body + self.status_code = status_code + self.headers = headers + self.url = url + self.ok = 200 <= status_code < 400 + self.text = body.decode("utf-8", errors="replace") + + def json(self) -> Any: + return json.loads(self.text) + + def raise_for_status(self) -> None: + if not self.ok: + raise T2TunesError(f"HTTP {self.status_code} for {self.url}") + + +class _UrllibSession: + def get( + self, + url: str, + *, + params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, + timeout: int = DEFAULT_TIMEOUT_SECONDS, + allow_redirects: bool = True, # kept for test/session compatibility + stream: bool = False, # kept for test/session compatibility + ) -> _HttpResponse: + del allow_redirects, stream + if params: + separator = "&" if "?" in url else "?" + url = f"{url}{separator}{urlencode(params)}" + return self._request("GET", url, headers=headers, timeout=timeout) + + def head( + self, + url: str, + *, + headers: Optional[Dict[str, str]] = None, + timeout: int = DEFAULT_TIMEOUT_SECONDS, + allow_redirects: bool = True, # kept for test/session compatibility + ) -> _HttpResponse: + del allow_redirects + return self._request("HEAD", url, headers=headers, timeout=timeout) + + def _request( + self, + method: str, + url: str, + *, + headers: Optional[Dict[str, str]] = None, + timeout: int = DEFAULT_TIMEOUT_SECONDS, + ) -> _HttpResponse: + request = Request(url, headers=headers or {}, method=method) + try: + with urlopen(request, timeout=timeout) as response: + return _HttpResponse( + body=response.read(), + status_code=response.status, + headers={k.lower(): v for k, v in response.headers.items()}, + url=response.url, + ) + except HTTPError as exc: + body = exc.read() if hasattr(exc, "read") else b"" + return _HttpResponse( + body=body, + status_code=exc.code, + headers={k.lower(): v for k, v in exc.headers.items()} if exc.headers else {}, + url=exc.url, + ) + except URLError as exc: + raise T2TunesError(f"Network error for {url}: {exc}") from exc + + +@dataclass(frozen=True) +class T2TunesSearchItem: + asin: str + title: str + artist_name: str + item_type: str + album_name: str = "" + album_asin: str = "" + duration_seconds: int = 0 + isrc: str = "" + + @property + def is_album(self) -> bool: + return "album" in self.item_type.lower() + + @property + def is_track(self) -> bool: + return "track" in self.item_type.lower() + + +@dataclass(frozen=True) +class T2TunesStreamInfo: + asin: str + streamable: bool + codec: str + format: str + sample_rate: Optional[int] + stream_url: str + has_decryption_key: bool + title: str = "" + artist: str = "" + album: str = "" + isrc: str = "" + + +class T2TunesClient: + def __init__( + self, + base_url: str = DEFAULT_BASE_URL, + *, + country: str = "US", + codec: str = "flac", + timeout: int = DEFAULT_TIMEOUT_SECONDS, + session: Optional[Any] = None, + ) -> None: + self.base_url = base_url.rstrip("/") + self.country = country.upper() + self.codec = codec.lower() + self.timeout = timeout + self.session = session or _UrllibSession() + + def status(self) -> Dict[str, Any]: + return self._get_json("/api/status") + + def amazon_music_is_up(self) -> bool: + status = self.status() + return str(status.get("amazonMusic", "")).lower() == "up" + + def search(self, query: str, *, types: str = "track,album") -> List[T2TunesSearchItem]: + data = self._get_json( + "/api/amazon-music/search", + params={ + "query": query, + "types": types, + "country": self.country, + }, + ) + return list(_iter_search_items(data)) + + def album_metadata(self, asin: str) -> Dict[str, Any]: + return self._get_json( + "/api/amazon-music/metadata", + params={ + "asin": asin, + "country": self.country, + }, + ) + + def media_from_asin(self, asin: str) -> List[T2TunesStreamInfo]: + data = self._get_json( + "/api/amazon-music/media-from-asin", + params={ + "asin": asin, + "country": self.country, + "codec": self.codec, + }, + ) + if isinstance(data, list): + return [_media_response_to_stream_info(item) for item in data if isinstance(item, dict)] + if isinstance(data, dict): + return [_media_response_to_stream_info(data)] + raise T2TunesError(f"Unexpected media response type: {type(data).__name__}") + + def probe_stream(self, stream_url: str) -> Dict[str, Any]: + """Probe a stream URL without downloading audio bytes.""" + try: + response = self.session.head(stream_url, timeout=self.timeout, allow_redirects=True) + method = "HEAD" + if response.status_code in (405, 403): + response = self.session.get( + stream_url, + timeout=self.timeout, + allow_redirects=True, + stream=True, + headers={"Range": "bytes=0-0"}, + ) + method = "GET range" + return { + "ok": response.ok, + "method": method, + "status_code": response.status_code, + "content_type": response.headers.get("content-type", ""), + "content_length": response.headers.get("content-length", ""), + "accept_ranges": response.headers.get("accept-ranges", ""), + "final_url": response.url, + } + except Exception as exc: + return { + "ok": False, + "error": str(exc), + } + + def _get_json(self, path: str, params: Optional[Dict[str, Any]] = None) -> Any: + url = urljoin(f"{self.base_url}/", path.lstrip("/")) + headers = { + "Accept": "application/json", + "Referer": self.base_url, + "User-Agent": "SoulSync-T2Tunes-Probe/0.1", + } + try: + response = self.session.get(url, params=params, headers=headers, timeout=self.timeout) + response.raise_for_status() + except Exception as exc: + raise T2TunesError(f"Request failed for {url}: {exc}") from exc + + try: + return response.json() + except ValueError as exc: + preview = response.text[:200].replace("\n", " ") + raise T2TunesError(f"Response was not JSON for {url}: {preview!r}") from exc + + +def _iter_search_items(response: Any) -> Iterable[T2TunesSearchItem]: + if not isinstance(response, dict): + raise T2TunesError(f"Unexpected search response type: {type(response).__name__}") + for result in response.get("results") or []: + if not isinstance(result, dict): + continue + for hit in result.get("hits") or []: + if not isinstance(hit, dict): + continue + document = hit.get("document") + if not isinstance(document, dict): + continue + asin = str(document.get("asin") or "") + if not asin: + continue + yield T2TunesSearchItem( + asin=asin, + title=str(document.get("title") or ""), + artist_name=str(document.get("artistName") or ""), + item_type=str(document.get("__type") or ""), + album_name=str(document.get("albumName") or ""), + album_asin=str(document.get("albumAsin") or ""), + duration_seconds=int(document.get("duration") or 0), + isrc=str(document.get("isrc") or ""), + ) + + +def _media_response_to_stream_info(item: Dict[str, Any]) -> T2TunesStreamInfo: + stream_info = item.get("streamInfo") if isinstance(item.get("streamInfo"), dict) else {} + tags = item.get("tags") if isinstance(item.get("tags"), dict) else {} + return T2TunesStreamInfo( + asin=str(item.get("asin") or ""), + streamable=bool(item.get("streamable") if item.get("streamable") is not None else item.get("stremeable")), + codec=str(stream_info.get("codec") or ""), + format=str(stream_info.get("format") or ""), + sample_rate=stream_info.get("sampleRate") if isinstance(stream_info.get("sampleRate"), int) else None, + stream_url=str(stream_info.get("streamUrl") or ""), + has_decryption_key=bool(item.get("decryptionKey")), + title=str(tags.get("title") or ""), + artist=str(tags.get("artist") or ""), + album=str(tags.get("album") or ""), + isrc=str(tags.get("isrc") or ""), + ) + + +def _print_json(data: Any) -> None: + print(json.dumps(data, indent=2, sort_keys=True, default=lambda o: getattr(o, "__dict__", str(o)))) + + +def main() -> int: + parser = argparse.ArgumentParser(description="Probe a T2Tunes/TripleTriple API instance.") + parser.add_argument("--base-url", default=DEFAULT_BASE_URL) + parser.add_argument("--country", default="US") + parser.add_argument("--codec", default="flac", choices=("flac", "opus", "eac3")) + parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT_SECONDS) + sub = parser.add_subparsers(dest="command", required=True) + + sub.add_parser("status") + + search_parser = sub.add_parser("search") + search_parser.add_argument("query") + search_parser.add_argument("--types", default="track,album") + + metadata_parser = sub.add_parser("metadata") + metadata_parser.add_argument("asin") + + media_parser = sub.add_parser("media") + media_parser.add_argument("asin") + media_parser.add_argument("--probe-stream", action="store_true") + + smoke_parser = sub.add_parser("smoke") + smoke_parser.add_argument("query") + smoke_parser.add_argument("--probe-stream", action="store_true") + + args = parser.parse_args() + client = T2TunesClient(args.base_url, country=args.country, codec=args.codec, timeout=args.timeout) + + if args.command == "status": + _print_json(client.status()) + return 0 + + if args.command == "search": + _print_json([item.__dict__ for item in client.search(args.query, types=args.types)]) + return 0 + + if args.command == "metadata": + _print_json(client.album_metadata(args.asin)) + return 0 + + if args.command == "media": + streams = client.media_from_asin(args.asin) + payload = [stream.__dict__ for stream in streams] + if args.probe_stream: + for index, stream in enumerate(streams): + payload[index]["stream_probe"] = client.probe_stream(stream.stream_url) if stream.stream_url else {"ok": False} + _print_json(payload) + return 0 + + if args.command == "smoke": + status = client.status() + search_items = client.search(args.query) + first = next((item for item in search_items if item.is_track or item.is_album), None) + media = client.media_from_asin(first.asin) if first else [] + payload = { + "status": status, + "amazon_music_up": str(status.get("amazonMusic", "")).lower() == "up", + "result_count": len(search_items), + "first_result": first.__dict__ if first else None, + "media": [stream.__dict__ for stream in media[:3]], + } + if args.probe_stream and media and media[0].stream_url: + payload["first_stream_probe"] = client.probe_stream(media[0].stream_url) + _print_json(payload) + return 0 + + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/webui/static/helper.js b/webui/static/helper.js index fb2c9c64..bb609b5e 100644 --- a/webui/static/helper.js +++ b/webui/static/helper.js @@ -3413,6 +3413,10 @@ function closeHelperSearch() { // projects that span multiple commits before shipping. Strip the flag at // release time and add a real `date:` line at the top of the version block. const WHATS_NEW = { + '2.5.3': [ + { unreleased: true }, + { title: 'Amazon Music Download Source', desc: 'new download source backed by T2Tunes proxy. searches the Amazon Music catalog, downloads 24-bit/48kHz FLAC (or Opus 320kbps / Dolby Atmos EAC3 fallback). codec waterfall mirrors the same pattern as Tidal/Qobuz — best quality first, auto-fallback. not yet wired into the main source selector — coming in a follow-up PR.', page: 'settings' }, + ], '2.5.2': [ // --- May 13, 2026 — 2.5.2 release --- { date: 'May 13, 2026 — 2.5.2 release' },