danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Amazon Music provider: metadata client + download source (T2Tunes)

Browse source 
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.
This commit is contained in:
Broque Thomas 2026-05-16 07:46:41 -07:00
parent d94a6e4f7a
commit 85984d4174
8 changed files with 3306 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

604
core/amazon_client.py Normal file
View file

@ -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 ""),
)

432
core/amazon_download_client.py Normal file
View file

@ -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 <hex> 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"),
)

0
tests/tools/__init__.py Normal file
View file

915
tests/tools/test_amazon_client.py Normal file
View file

@ -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 = "<html>error</html>"
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

783
tests/tools/test_amazon_download_client.py Normal file
View file

@ -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

199
tools/t2tunes_media_plan.py Normal file
View file

@ -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/<query>-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())

369
tools/t2tunes_probe.py Normal file
View file

@ -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())

4
webui/static/helper.js
View file

@ -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' },