Merge pull request #615 from Nezreka/feature/amazon-music-download

Feature/amazon music download
This commit is contained in:
BoulderBadgeDad 2026-05-16 12:56:09 -07:00 committed by GitHub
commit e8b9f80597
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
20 changed files with 3645 additions and 26 deletions

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

View file

@ -0,0 +1,452 @@
"""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._quality = config_manager.get("amazon_download.quality", "flac")
self._allow_fallback = config_manager.get("amazon_download.allow_fallback", True)
self._client = AmazonClient(preferred_codec=self._quality)
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 set_engine(self, engine) -> None:
"""Engine callback — wires the central thread worker + state store."""
self._engine = engine
def set_shutdown_check(self, check_callable) -> None:
self.shutdown_check = check_callable
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)
codecs = CODEC_PREFERENCE if self._allow_fallback else [self._quality]
for codec in codecs:
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)
final_size = out_path.stat().st_size
logger.info(
f"Amazon download complete ({codec}): {out_path} "
f"({final_size / (1024 * 1024):.1f} MB)"
)
# Sync size == transferred so the download monitor's bytes-incomplete
# guard doesn't block post-processing. The throttled updates in
# _stream_to_file leave transferred < size after the last 0.5s tick;
# other streaming clients avoid this by not tracking bytes at all
# (size stays 0, the guard is skipped). Writing the final output size
# here restores parity.
if self._engine is not None:
self._engine.update_record(
"amazon", download_id, {'size': final_size, 'transferred': final_size}
)
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 []
return [
self._record_to_status(record)
for record in self._engine.iter_records_for_source('amazon')
]
async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]:
if self._engine is None:
return None
record = self._engine.get_record('amazon', download_id)
return self._record_to_status(record) if record is not None else None
async def cancel_download(
self,
download_id: str,
username: Optional[str] = None,
remove: bool = False,
) -> bool:
if self._engine is None:
return False
if self._engine.get_record('amazon', download_id) is None:
return False
self._engine.update_record('amazon', download_id, {'state': 'Cancelled'})
if remove:
self._engine.remove_record('amazon', download_id)
return True
async def clear_all_completed_downloads(self) -> bool:
if self._engine is None:
return True
terminal = {'Completed, Succeeded', 'Cancelled', 'Errored', 'Aborted'}
for record in list(self._engine.iter_records_for_source('amazon')):
if record.get('state') in terminal:
self._engine.remove_record('amazon', record['id'])
return True
# ------------------------------------------------------------------
# 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(rec: Dict[str, Any]) -> DownloadStatus:
return DownloadStatus(
id=str(rec.get('id', '')),
filename=str(rec.get('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'),
)

View file

@ -30,6 +30,7 @@ RATE_LIMITS = {
'tidal': 120, # MIN_API_INTERVAL=0.5s → ~120/min
'qobuz': 60, # Variable throttle, ~60/min estimate
'discogs': 60, # MIN_API_INTERVAL=1.0s with auth → ~60/min
'amazon': 120, # MIN_API_INTERVAL=0.5s → ~120/min (T2Tunes proxy)
}
# Display names for UI
@ -44,12 +45,13 @@ SERVICE_LABELS = {
'tidal': 'Tidal',
'qobuz': 'Qobuz',
'discogs': 'Discogs',
'amazon': 'Amazon Music',
}
# Display order
SERVICE_ORDER = [
'spotify', 'itunes', 'deezer', 'lastfm', 'genius',
'musicbrainz', 'audiodb', 'tidal', 'qobuz', 'discogs',
'musicbrainz', 'audiodb', 'tidal', 'qobuz', 'discogs', 'amazon',
]

View file

@ -104,6 +104,15 @@ class DownloadOrchestrator:
deezer_dl.reconnect(deezer_arl)
deezer_dl._quality = config_manager.get('deezer_download.quality', 'flac')
# Reload Amazon quality preference (T2Tunes needs no reconnect — public proxy)
amazon = self.client('amazon')
if amazon:
quality = config_manager.get('amazon_download.quality', 'flac')
amazon._quality = quality
amazon._allow_fallback = config_manager.get('amazon_download.allow_fallback', True)
if hasattr(amazon, '_client') and amazon._client:
amazon._client.preferred_codec = quality
# Reload download path for all clients that cache it.
# Soulseek owns the path config and is reloaded above; every
# other source mirrors that path so files all land in one
@ -319,7 +328,7 @@ class DownloadOrchestrator:
return None
# 2. Filter and validate results
_streaming_sources = ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud')
_streaming_sources = ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud', 'amazon')
is_streaming = tracks[0].username in _streaming_sources if tracks else False
if is_streaming and expected_track:

View file

@ -38,6 +38,7 @@ from core.download_plugins.base import DownloadSourcePlugin
# than the legacy module-top imports here. Importing everything at
# registry-load time pins the bindings the same way the legacy
# orchestrator did.
from core.amazon_download_client import AmazonDownloadClient
from core.deezer_download_client import DeezerDownloadClient
from core.hifi_client import HiFiClient
from core.lidarr_download_client import LidarrDownloadClient
@ -176,6 +177,7 @@ def build_default_registry() -> DownloadPluginRegistry:
"""
registry = DownloadPluginRegistry()
registry.register(PluginSpec(name='amazon', factory=AmazonDownloadClient, display_name='Amazon Music'))
registry.register(PluginSpec(name='soulseek', factory=SoulseekClient, display_name='Soulseek'))
registry.register(PluginSpec(name='youtube', factory=YouTubeClient, display_name='YouTube'))
registry.register(PluginSpec(name='tidal', factory=TidalDownloadClient, display_name='Tidal'))

View file

@ -87,7 +87,7 @@ class StatusDeps:
# Streaming sources the engine fallback applies to. Soulseek goes through
# slskd's live_transfers path and must NOT hit the engine fallback.
_STREAMING_SOURCE_NAMES = frozenset((
'youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud',
'youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud', 'amazon',
))
# Keep these in sync with the engine plugins' state strings.

View file

@ -294,7 +294,7 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
source_clients = {
name: orch.client(name)
for name in ('soulseek', 'youtube', 'tidal', 'qobuz',
'hifi', 'deezer_dl', 'lidarr', 'soundcloud')
'hifi', 'deezer_dl', 'lidarr', 'soundcloud', 'amazon')
}
# The orchestrator tried sources in order but stopped at the first with results.

View file

@ -79,7 +79,7 @@ def get_valid_candidates(results, spotify_track, query):
# Streaming sources (YouTube, Tidal, Qobuz, HiFi, Deezer, SoundCloud) return structured API results
# with proper artist/title metadata — score using the same matching engine as Soulseek
_streaming_sources = ("youtube", "tidal", "qobuz", "hifi", "deezer_dl", "soundcloud")
_streaming_sources = ("youtube", "tidal", "qobuz", "hifi", "deezer_dl", "soundcloud", "amazon")
if results[0].username in _streaming_sources:
source_label = results[0].username.replace('_dl', '').title()
expected_artists = spotify_track.artists if spotify_track else []

View file

@ -235,7 +235,7 @@ def record_library_history_download(context: Dict[str, Any]) -> None:
source_track_id = search_result.get("track_id", "") or search_result.get("id", "") or ti.get("id", "")
source_track_title = search_result.get("title", "") or search_result.get("name", "")
source_artist = search_result.get("artist", "")
if source_filename and "||" in source_filename and username in ("tidal", "youtube", "qobuz", "hifi", "deezer_dl", "lidarr", "soundcloud"):
if source_filename and "||" in source_filename and username in ("tidal", "youtube", "qobuz", "hifi", "deezer_dl", "lidarr", "soundcloud", "amazon"):
stream_id = source_filename.split("||")[0]
if stream_id and not source_track_id:
source_track_id = stream_id

View file

@ -59,6 +59,7 @@ def _import_plugin_classes():
from core.deezer_download_client import DeezerDownloadClient
from core.lidarr_download_client import LidarrDownloadClient
from core.soundcloud_client import SoundcloudClient
from core.amazon_download_client import AmazonDownloadClient
return {
'soulseek': SoulseekClient,
@ -69,10 +70,11 @@ def _import_plugin_classes():
'deezer': DeezerDownloadClient,
'lidarr': LidarrDownloadClient,
'soundcloud': SoundcloudClient,
'amazon': AmazonDownloadClient,
}
def test_default_registry_registers_all_eight_sources():
def test_default_registry_registers_all_sources():
"""Smoke check that the foundation registry knows about every
source the orchestrator historically dispatched to. If someone
drops a registration here, every other test in this module would
@ -82,7 +84,7 @@ def test_default_registry_registers_all_eight_sources():
registry = build_default_registry()
expected = {
'soulseek', 'youtube', 'tidal', 'qobuz',
'hifi', 'deezer', 'lidarr', 'soundcloud',
'hifi', 'deezer', 'lidarr', 'soundcloud', 'amazon',
}
assert set(registry.names()) == expected
@ -102,7 +104,7 @@ def test_deezer_dl_alias_is_registered_against_deezer_spec():
@pytest.mark.parametrize('plugin_name', [
'soulseek', 'youtube', 'tidal', 'qobuz',
'hifi', 'deezer', 'lidarr', 'soundcloud',
'hifi', 'deezer', 'lidarr', 'soundcloud', 'amazon',
])
def test_plugin_class_has_all_required_methods(plugin_name):
"""Every registered plugin class exposes every protocol method
@ -122,7 +124,7 @@ def test_plugin_class_has_all_required_methods(plugin_name):
@pytest.mark.parametrize('plugin_name', [
'soulseek', 'youtube', 'tidal', 'qobuz',
'hifi', 'deezer', 'lidarr', 'soundcloud',
'hifi', 'deezer', 'lidarr', 'soundcloud', 'amazon',
])
def test_plugin_class_async_methods_are_coroutines(plugin_name):
"""Methods declared async in the protocol must be async on every

0
tests/tools/__init__.py Normal file
View 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

View file

@ -0,0 +1,847 @@
"""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.run(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 = {
"id": "dl-001",
"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(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({})
assert status.id == ""
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_final_size_update_syncs_transferred_to_size(self, tmp_path):
"""size == transferred after success so monitor bytes-incomplete guard doesn't block."""
client, audio_data = self._setup(tmp_path, decryption_key=None)
engine = MagicMock()
client._engine = engine
result = client._download_sync("dl-001", "B09XYZ1234", "Artist - Track")
assert result is not None
# Last update must set size == transferred (both equal the output file size)
final_calls = [
c[0][2] for c in engine.update_record.call_args_list
if 'size' in c[0][2] and 'transferred' in c[0][2]
and c[0][2]['size'] == c[0][2]['transferred']
and c[0][2]['size'] > 0
]
assert final_calls, "Expected a final engine update with size == transferred > 0"
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"
def test_set_engine_wires_engine(self, tmp_path):
client = _make_client(tmp_path)
assert client._engine is None
engine = MagicMock()
client.set_engine(engine)
assert client._engine is engine
def test_set_shutdown_check_wires_callback(self, tmp_path):
client = _make_client(tmp_path)
assert client.shutdown_check is None
check = lambda: False
client.set_shutdown_check(check)
assert client.shutdown_check is check
def test_set_engine_allows_download_dispatch(self, tmp_path):
"""set_engine() must unblock download() — the live failure mode."""
client = _make_client(tmp_path)
engine = MagicMock()
engine.worker.dispatch.return_value = "dl-wired"
client.set_engine(engine)
result = run(client.download("amazon", "B09XYZ1234||Artist - Title"))
assert result == "dl-wired"
# ---------------------------------------------------------------------------
# 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.iter_records_for_source.return_value = iter([{
"id": "dl-001",
"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"
engine.iter_records_for_source.assert_called_once_with('amazon')
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 = {
"id": "dl-001",
"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.id == "dl-001"
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.get_record.return_value = {'id': 'dl-001', 'state': 'downloading'}
client._engine = engine
result = run(client.cancel_download("dl-001", remove=True))
assert result is True
engine.update_record.assert_called_once_with('amazon', 'dl-001', {'state': 'Cancelled'})
engine.remove_record.assert_called_once_with('amazon', 'dl-001')
def test_cancel_returns_false_when_record_not_found(self, tmp_path):
client = _make_client(tmp_path)
engine = MagicMock()
engine.get_record.return_value = None
client._engine = engine
result = run(client.cancel_download("nonexistent"))
assert result is False
engine.update_record.assert_not_called()
def test_clear_completed_returns_true_without_engine(self, tmp_path):
client = _make_client(tmp_path)
assert run(client.clear_all_completed_downloads()) is True
def test_clear_completed_removes_terminal_records(self, tmp_path):
client = _make_client(tmp_path)
engine = MagicMock()
engine.iter_records_for_source.return_value = iter([
{'id': 'dl-001', 'state': 'Completed, Succeeded'},
{'id': 'dl-002', 'state': 'InProgress, Downloading'},
{'id': 'dl-003', 'state': 'Errored'},
])
client._engine = engine
result = run(client.clear_all_completed_downloads())
assert result is True
# Only terminal records removed
removed_ids = [c[0][1] for c in engine.remove_record.call_args_list]
assert 'dl-001' in removed_ids
assert 'dl-003' in removed_ids
assert 'dl-002' not in removed_ids
def test_status_methods_no_records(self, tmp_path):
"""Engine with no Amazon records returns empty/None gracefully."""
client = _make_client(tmp_path)
engine = MagicMock()
engine.iter_records_for_source.return_value = iter([])
engine.get_record.return_value = None
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 True

View file

@ -0,0 +1,125 @@
from __future__ import annotations
import unittest
from pathlib import Path
from unittest.mock import Mock
import sys
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "tools"))
from t2tunes_probe import T2TunesClient, T2TunesError # noqa: E402
class _Response:
def __init__(self, payload=None, *, status_code=200, text="", headers=None, url="https://example.test/x"):
self._payload = payload
self.status_code = status_code
self.text = text
self.headers = headers or {}
self.url = url
self.ok = 200 <= status_code < 400
def json(self):
if isinstance(self._payload, Exception):
raise self._payload
return self._payload
def raise_for_status(self):
if not self.ok:
raise RuntimeError(f"{self.status_code} error")
class T2TunesProbeTests(unittest.TestCase):
def test_status_uses_api_status_endpoint(self):
session = Mock()
session.get.return_value = _Response({"amazonMusic": "up"})
client = T2TunesClient("https://t2.example", session=session)
self.assertTrue(client.amazon_music_is_up())
session.get.assert_called_once()
self.assertEqual(session.get.call_args.args[0], "https://t2.example/api/status")
def test_search_flattens_nested_hits(self):
session = Mock()
session.get.return_value = _Response({
"results": [{
"hits": [{
"document": {
"__type": "Track",
"asin": "B001",
"title": "Song",
"artistName": "Artist",
"albumName": "Album",
"duration": 123,
"isrc": "USABC123",
}
}]
}]
})
client = T2TunesClient("https://t2.example", session=session)
items = client.search("artist song")
self.assertEqual(len(items), 1)
self.assertEqual(items[0].asin, "B001")
self.assertTrue(items[0].is_track)
self.assertEqual(items[0].duration_seconds, 123)
self.assertEqual(session.get.call_args.kwargs["params"]["types"], "track,album")
def test_media_handles_streamable_typo_and_decryption_flag(self):
session = Mock()
session.get.return_value = _Response([{
"asin": "B001",
"stremeable": True,
"decryptionKey": "abc123",
"tags": {
"title": "Song",
"artist": "Artist",
"album": "Album",
"isrc": "USABC123",
},
"streamInfo": {
"format": "flac",
"codec": "flac",
"sampleRate": 48000,
"streamUrl": "https://cdn.example/song.flac",
},
}])
client = T2TunesClient("https://t2.example", session=session)
streams = client.media_from_asin("B001")
self.assertEqual(len(streams), 1)
self.assertTrue(streams[0].streamable)
self.assertTrue(streams[0].has_decryption_key)
self.assertEqual(streams[0].stream_url, "https://cdn.example/song.flac")
def test_non_json_response_raises_probe_error(self):
session = Mock()
session.get.return_value = _Response(ValueError("not json"), text="Service Unavailable")
client = T2TunesClient("https://t2.example", session=session)
with self.assertRaises(T2TunesError):
client.status()
def test_probe_stream_falls_back_to_range_get_when_head_is_blocked(self):
session = Mock()
session.head.return_value = _Response(status_code=405)
session.get.return_value = _Response(
status_code=206,
headers={"content-type": "audio/flac", "content-length": "1"},
url="https://cdn.example/song.flac",
)
client = T2TunesClient("https://t2.example", session=session)
result = client.probe_stream("https://cdn.example/song.flac")
self.assertTrue(result["ok"])
self.assertEqual(result["method"], "GET range")
self.assertEqual(result["content_type"], "audio/flac")
self.assertEqual(session.get.call_args.kwargs["headers"], {"Range": "bytes=0-0"})
if __name__ == "__main__":
unittest.main()

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

View file

@ -1750,21 +1750,22 @@ def _find_downloaded_file(download_path, track_data):
audio_extensions = {'.mp3', '.flac', '.ogg', '.aac', '.wma', '.wav', '.m4a'}
target_filename = extract_filename(track_data.get('filename', ''))
# YOUTUBE/TIDAL/QOBUZ/HIFI SUPPORT: Handle encoded filename format "id||title"
# YOUTUBE/TIDAL/QOBUZ/HIFI/AMAZON SUPPORT: Handle encoded filename format "id||title"
# The file on disk will be "title.ext", not "id||title"
is_youtube = track_data.get('username') == 'youtube'
is_tidal = track_data.get('username') == 'tidal'
is_qobuz = track_data.get('username') == 'qobuz'
is_hifi = track_data.get('username') == 'hifi'
is_streaming_source = is_youtube or is_tidal or is_qobuz or is_hifi
is_amazon = track_data.get('username') == 'amazon'
is_streaming_source = is_youtube or is_tidal or is_qobuz or is_hifi or is_amazon
target_filename_youtube = None
if is_streaming_source and '||' in target_filename:
_, title = target_filename.split('||', 1)
if is_tidal or is_qobuz or is_hifi:
# Tidal/Qobuz/HiFi files can be flac or m4a — match any audio extension
if is_tidal or is_qobuz or is_hifi or is_amazon:
# Tidal/Qobuz/HiFi/Amazon files can be flac, opus, eac3, or m4a — match any audio extension
safe_title = re.sub(r'[<>:"/\\|?*]', '_', title)
target_filename_youtube = safe_title # Extension-less for flexible matching
source_name = 'HiFi' if is_hifi else ('Qobuz' if is_qobuz else 'Tidal')
source_name = 'HiFi' if is_hifi else ('Qobuz' if is_qobuz else ('Amazon' if is_amazon else 'Tidal'))
logger.debug(f"[{source_name} Stream] Looking for file starting with: {target_filename_youtube}")
else:
# yt-dlp will create "Title.mp3" from "Title"
@ -1802,11 +1803,11 @@ def _find_downloaded_file(download_path, track_data):
# For Tidal, compare without extension (file could be .flac or .m4a)
compare_target = target_filename_youtube.lower()
compare_file = file.lower()
if is_tidal or is_qobuz or is_hifi:
if is_tidal or is_qobuz or is_hifi or is_amazon:
compare_file = os.path.splitext(compare_file)[0]
similarity = SequenceMatcher(None, compare_file, compare_target).ratio()
source_label = 'HiFi' if is_hifi else ('Qobuz' if is_qobuz else ('Tidal' if is_tidal else 'YouTube'))
source_label = 'HiFi' if is_hifi else ('Qobuz' if is_qobuz else ('Amazon' if is_amazon else ('Tidal' if is_tidal else 'YouTube')))
logger.debug(f"[{source_label} Stream] Comparing: '{file}' vs '{target_filename_youtube}' = {similarity:.2f}")
# Keep track of best match
@ -1826,7 +1827,7 @@ def _find_downloaded_file(download_path, track_data):
# For YouTube/Tidal, if we found a good enough match (80%+), use it
if is_streaming_source and best_match and best_similarity >= 0.80:
source_label = 'Qobuz' if is_qobuz else ('Tidal' if is_tidal else 'YouTube')
source_label = 'Qobuz' if is_qobuz else ('Amazon' if is_amazon else ('Tidal' if is_tidal else 'YouTube'))
logger.debug(f"Found good match ({best_similarity:.2f}) for {source_label} streaming file: {best_match}")
return best_match
@ -2121,7 +2122,7 @@ def get_status():
# don't depend on slskd being reachable — when one of these is the
# active source, surface "connected" without probing slskd so the
# dashboard / sidebar indicator stays green.
serverless_sources = ('youtube', 'hifi', 'qobuz', 'tidal', 'deezer_dl', 'lidarr', 'soundcloud')
serverless_sources = ('youtube', 'hifi', 'qobuz', 'tidal', 'deezer_dl', 'lidarr', 'soundcloud', 'amazon')
is_serverless = (download_mode in serverless_sources or
(download_mode == 'hybrid' and
hybrid_order and any(s in serverless_sources for s in hybrid_order)))
@ -2748,7 +2749,7 @@ def handle_settings():
if 'active_media_server' in new_settings:
config_manager.set_active_media_server(new_settings['active_media_server'])
for service in ['spotify', 'plex', 'jellyfin', 'navidrome', 'soulseek', 'download_source', 'settings', 'database', 'metadata_enhancement', 'file_organization', 'playlist_sync', 'tidal', 'tidal_download', 'qobuz', 'hifi_download', 'deezer_download', 'lidarr_download', 'listenbrainz', 'acoustid', 'lastfm', 'genius', 'import', 'lossy_copy', 'listening_stats', 'ui_appearance', 'youtube', 'content_filter', 'itunes', 'm3u_export', 'musicbrainz', 'deezer', 'audiodb', 'metadata', 'hydrabase', 'security', 'discogs', 'library', 'discover', 'wishlist', 'genre_whitelist', 'post_processing']:
for service in ['spotify', 'plex', 'jellyfin', 'navidrome', 'soulseek', 'download_source', 'settings', 'database', 'metadata_enhancement', 'file_organization', 'playlist_sync', 'tidal', 'tidal_download', 'qobuz', 'hifi_download', 'deezer_download', 'amazon_download', 'lidarr_download', 'listenbrainz', 'acoustid', 'lastfm', 'genius', 'import', 'lossy_copy', 'listening_stats', 'ui_appearance', 'youtube', 'content_filter', 'itunes', 'm3u_export', 'musicbrainz', 'deezer', 'audiodb', 'metadata', 'hydrabase', 'security', 'discogs', 'library', 'discover', 'wishlist', 'genre_whitelist', 'post_processing']:
if service in new_settings:
for key, value in new_settings[service].items():
config_manager.set(f'{service}.{key}', value)
@ -5796,7 +5797,7 @@ def start_download():
if download_id:
# Register download for post-processing (simple transfer to /Transfer)
context_key = _make_context_key(username, filename)
is_streaming_source = username in ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud')
is_streaming_source = username in ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud', 'amazon')
with matched_context_lock:
matched_downloads_context[context_key] = {
'search_result': {
@ -6178,7 +6179,7 @@ def get_download_status():
all_streaming_downloads = run_async(download_orchestrator.get_all_downloads())
for download in all_streaming_downloads:
if download.username in ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud'):
if download.username in ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud', 'amazon'):
source_label = download.username.title()
# Convert DownloadStatus to transfer format that frontend expects
streaming_transfer = {
@ -11105,7 +11106,7 @@ def redownload_search_sources(track_id):
quality = ext if ext in ('FLAC', 'MP3', 'OPUS', 'OGG', 'M4A', 'WAV') else candidate.quality or ''
svc = source_name if source_name != 'default' else 'hybrid'
uname = candidate.username
if uname in ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud'):
if uname in ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud', 'amazon'):
svc = uname
source_candidates.append({
'username': uname,
@ -16411,7 +16412,7 @@ def _try_source_reuse(task_id, batch_id, track):
if not source_tracks or not last_source:
_sr.info("Skipped — no source_tracks or no last_source")
return False
if last_source.get('username') in ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud'):
if last_source.get('username') in ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud', 'amazon'):
_sr.info(f"Skipped — {last_source.get('username')} source (no folder-based reuse)")
return False
@ -16513,7 +16514,7 @@ def _store_batch_source(batch_id, username, filename):
"""Browse the successful download's folder and store results on the batch for reuse."""
_sr = source_reuse_logger
_sr.info(f"_store_batch_source called: batch={batch_id}, user={username}, file={filename}")
if not batch_id or username in ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud'):
if not batch_id or username in ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud', 'amazon'):
_sr.info(f"Skipped — no batch_id or streaming source ({username})")
return
@ -19683,6 +19684,26 @@ def deezer_download_test_download():
# ===================================================================
# AMAZON DOWNLOAD ENDPOINTS
# ===================================================================
@app.route('/api/amazon/test-connection', methods=['GET'])
@admin_only
def amazon_test_connection():
"""Check whether the T2Tunes proxy is up and Amazon Music is reachable."""
try:
from core.amazon_client import AmazonClient
c = AmazonClient()
status = c.status()
amazon_up = str(status.get('amazonMusic', '')).lower() == 'up'
return jsonify({
'connected': amazon_up,
'status': status,
})
except Exception as e:
return jsonify({'connected': False, 'error': str(e)}), 200
# TIDAL DOWNLOAD AUTH ENDPOINTS
# ===================================================================

View file

@ -4294,6 +4294,7 @@
<option value="qobuz">Qobuz Only</option>
<option value="hifi">HiFi Only (Free Lossless)</option>
<option value="deezer_dl">Deezer Only</option>
<option value="amazon">Amazon Music Only</option>
<option value="lidarr">Lidarr Only</option>
<option value="soundcloud">SoundCloud Only</option>
<option value="hybrid">Hybrid (Primary + Fallback)</option>
@ -4616,6 +4617,39 @@
</div>
</div>
<!-- Amazon Music Download Settings (shown only when amazon mode is selected) -->
<div id="amazon-download-settings-container" style="display: none;">
<div class="form-group">
<label>Amazon Music Quality:</label>
<select id="amazon-quality" class="form-select">
<option value="flac">FLAC Lossless (24-bit/48kHz Hi-Res)</option>
<option value="opus">Opus (320kbps)</option>
<option value="eac3">EAC3 Dolby Atmos (768kbps 5.1)</option>
</select>
<div class="setting-help-text">
Preferred codec tier. FLAC is 24-bit/48kHz Hi-Res — no subscription required.
Downloads via T2Tunes proxy.
</div>
<label class="checkbox-inline" style="margin-top: 8px;">
<input type="checkbox" id="amazon-allow-fallback" checked>
Allow quality fallback
</label>
<div class="setting-help-text">
Fall back to the next codec tier if the preferred one is unavailable.
</div>
</div>
<div class="form-group">
<label>Connection:</label>
<div class="form-actions" style="margin-top: 4px;">
<button class="test-button" onclick="testAmazonConnection()">Test Connection</button>
<span id="amazon-connection-status" class="setting-help-text" style="margin-left: 8px;"></span>
</div>
<div class="setting-help-text">
No account required — T2Tunes is a public Amazon Music proxy.
</div>
</div>
</div>
<!-- SoundCloud Download Settings -->
<div id="soundcloud-download-settings-container" style="display: none;">
<div class="form-group">

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 Tidal/Qobuz — best quality first, auto-fallback. selectable as a standalone or hybrid source from Settings.', page: 'settings' },
],
'2.5.2': [
// --- May 13, 2026 — 2.5.2 release ---
{ date: 'May 13, 2026 — 2.5.2 release' },

View file

@ -595,12 +595,13 @@ const HYBRID_SOURCES = [
{ id: 'qobuz', name: 'Qobuz', icon: 'https://www.svgrepo.com/show/504778/qobuz.svg', emoji: '🎧' },
{ id: 'hifi', name: 'HiFi', icon: null, emoji: '🎶' },
{ id: 'deezer_dl', name: 'Deezer', icon: 'https://www.svgrepo.com/show/519734/deezer.svg', emoji: '🎧' },
{ id: 'amazon', name: 'Amazon Music', icon: null, emoji: '🛒' },
{ id: 'lidarr', name: 'Lidarr', icon: null, emoji: '📦' },
{ id: 'soundcloud', name: 'SoundCloud', icon: 'https://www.svgrepo.com/show/452219/soundcloud.svg', emoji: '☁️' },
];
let _hybridSourceOrder = ['soulseek', 'youtube'];
let _hybridSourceEnabled = { soulseek: true, youtube: true, tidal: false, qobuz: false, hifi: false, deezer_dl: false, lidarr: false, soundcloud: false };
let _hybridSourceEnabled = { soulseek: true, youtube: true, tidal: false, qobuz: false, hifi: false, deezer_dl: false, amazon: false, lidarr: false, soundcloud: false };
let _hybridVisualOrder = null; // Full visual order including disabled sources
function buildHybridSourceList() {
@ -942,6 +943,8 @@ async function loadSettingsData() {
document.getElementById('deezer-download-quality').value = settings.deezer_download?.quality || 'flac';
document.getElementById('deezer-allow-fallback').checked = settings.deezer_download?.allow_fallback !== false;
document.getElementById('deezer-download-arl').value = settings.deezer_download?.arl || '';
document.getElementById('amazon-quality').value = settings.amazon_download?.quality || 'flac';
document.getElementById('amazon-allow-fallback').checked = settings.amazon_download?.allow_fallback !== false;
document.getElementById('lidarr-url').value = settings.lidarr_download?.url || '';
document.getElementById('lidarr-api-key').value = settings.lidarr_download?.api_key || '';
// Sync ARL to connections tab field + bidirectional listeners
@ -1477,6 +1480,7 @@ function updateDownloadSourceUI() {
const youtubeContainer = document.getElementById('youtube-settings-container');
const hifiContainer = document.getElementById('hifi-download-settings-container');
const deezerDlContainer = document.getElementById('deezer-download-settings-container');
const amazonContainer = document.getElementById('amazon-download-settings-container');
const lidarrContainer = document.getElementById('lidarr-download-settings-container');
const soundcloudContainer = document.getElementById('soundcloud-download-settings-container');
@ -1499,6 +1503,7 @@ function updateDownloadSourceUI() {
youtubeContainer.style.display = activeSources.has('youtube') ? 'block' : 'none';
hifiContainer.style.display = activeSources.has('hifi') ? 'block' : 'none';
if (deezerDlContainer) deezerDlContainer.style.display = activeSources.has('deezer_dl') ? 'block' : 'none';
if (amazonContainer) amazonContainer.style.display = activeSources.has('amazon') ? 'block' : 'none';
if (lidarrContainer) lidarrContainer.style.display = activeSources.has('lidarr') ? 'block' : 'none';
if (soundcloudContainer) soundcloudContainer.style.display = activeSources.has('soundcloud') ? 'block' : 'none';
@ -1519,6 +1524,9 @@ function updateDownloadSourceUI() {
if (activeSources.has('hifi')) {
testHiFiConnection();
}
if (activeSources.has('amazon')) {
testAmazonConnection();
}
if (activeSources.has('soundcloud')) {
testSoundcloudConnection();
}
@ -1535,6 +1543,7 @@ function updateHybridSecondaryOptions() {
{ value: 'qobuz', label: 'Qobuz' },
{ value: 'hifi', label: 'HiFi' },
{ value: 'deezer_dl', label: 'Deezer' },
{ value: 'amazon', label: 'Amazon Music' },
{ value: 'lidarr', label: 'Lidarr' },
{ value: 'soundcloud', label: 'SoundCloud' },
];
@ -2675,6 +2684,10 @@ async function saveSettings(quiet = false) {
arl: document.getElementById('deezer-download-arl').value || '',
allow_fallback: document.getElementById('deezer-allow-fallback').checked,
},
amazon_download: {
quality: document.getElementById('amazon-quality').value || 'flac',
allow_fallback: document.getElementById('amazon-allow-fallback').checked,
},
lidarr_download: {
url: document.getElementById('lidarr-url').value || '',
api_key: document.getElementById('lidarr-api-key').value || '',
@ -3758,6 +3771,27 @@ async function testDeezerDownloadConnection() {
}
}
async function testAmazonConnection() {
const statusEl = document.getElementById('amazon-connection-status');
if (!statusEl) return;
statusEl.textContent = 'Checking...';
statusEl.style.color = '#aaa';
try {
const resp = await fetch('/api/amazon/test-connection');
const data = await resp.json();
if (data.connected) {
statusEl.textContent = '✓ Connected — T2Tunes up';
statusEl.style.color = '#4caf50';
} else {
statusEl.textContent = '✗ ' + (data.error || 'T2Tunes unreachable');
statusEl.style.color = '#f44336';
}
} catch (e) {
statusEl.textContent = '✗ Connection error';
statusEl.style.color = '#f44336';
}
}
async function checkTidalDownloadAuthStatus() {
const statusEl = document.getElementById('tidal-download-auth-status');
const btn = document.getElementById('tidal-download-auth-btn');