mirror of
https://github.com/beetbox/beets.git
synced 2026-08-31 08:10:41 -04:00
Add soundcloud plugin
This commit is contained in:
Šarūnas Nejus
218
beetsplug/soundcloud/__init__.py
Normal file
218
beetsplug/soundcloud/__init__.py
Normal file
@@ -0,0 +1,218 @@
|
||||
"""Add public SoundCloud track and release matches to the autotagger."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from functools import cached_property
|
||||
from typing import TYPE_CHECKING, ClassVar
|
||||
|
||||
import confuse
|
||||
|
||||
from beets import config
|
||||
from beets.autotag import AlbumInfo, TrackInfo
|
||||
from beets.dbcore import types
|
||||
from beets.metadata_plugins import IDResponse, SearchApiMetadataSourcePlugin
|
||||
|
||||
from .api import SoundCloudAPI
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Sequence
|
||||
|
||||
from beets.library import Item
|
||||
from beets.metadata_plugins import QueryType, SearchParams
|
||||
|
||||
from .api_types import (
|
||||
SoundCloudPlaylist,
|
||||
SoundCloudResource,
|
||||
SoundCloudTrack,
|
||||
)
|
||||
|
||||
|
||||
class SoundCloudSearchResult(IDResponse):
|
||||
pass
|
||||
|
||||
|
||||
class SoundCloudPlugin(SearchApiMetadataSourcePlugin[SoundCloudSearchResult]):
|
||||
item_types: ClassVar[dict[str, types.Type]] = {
|
||||
"soundcloud_track_urn": types.STRING,
|
||||
"soundcloud_artist_urn": types.STRING,
|
||||
}
|
||||
album_types: ClassVar[dict[str, types.Type]] = {
|
||||
"soundcloud_playlist_urn": types.STRING,
|
||||
"soundcloud_artist_urn": types.STRING,
|
||||
}
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.config.add(
|
||||
{
|
||||
"client_id": "",
|
||||
"client_secret": "",
|
||||
"tokenfile": "soundcloud_token.json",
|
||||
}
|
||||
)
|
||||
self.config["client_id"].redact = True
|
||||
self.config["client_secret"].redact = True
|
||||
|
||||
def _tokenfile(self) -> str:
|
||||
return self.config["tokenfile"].get(confuse.Filename(in_app_dir=True))
|
||||
|
||||
@cached_property
|
||||
def api(self) -> SoundCloudAPI:
|
||||
return SoundCloudAPI(
|
||||
self.config["client_id"].as_str(),
|
||||
self.config["client_secret"].as_str(),
|
||||
self._tokenfile(),
|
||||
)
|
||||
|
||||
def get_search_query_with_filters(
|
||||
self,
|
||||
query_type: QueryType,
|
||||
_items: Sequence[Item],
|
||||
artist: str,
|
||||
name: str,
|
||||
va_likely: bool,
|
||||
) -> tuple[str, dict[str, str]]:
|
||||
if query_type == "album" and va_likely:
|
||||
return name, {}
|
||||
return " ".join(filter(None, (artist, name))), {}
|
||||
|
||||
def get_search_response(
|
||||
self, params: SearchParams
|
||||
) -> list[SoundCloudSearchResult]:
|
||||
resources: Sequence[SoundCloudResource]
|
||||
if params.query_type == "track":
|
||||
resources = self.api.search_tracks(params.query, params.limit)
|
||||
else:
|
||||
resources = self.api.search_playlists(
|
||||
params.query, params.limit, playlist_type="album"
|
||||
)
|
||||
return [
|
||||
SoundCloudSearchResult(id=resource["urn"])
|
||||
for resource in resources
|
||||
if resource.get("urn")
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _genres(
|
||||
resource: SoundCloudTrack | SoundCloudPlaylist,
|
||||
) -> list[str] | None:
|
||||
genre = resource.get("genre")
|
||||
return [genre] if isinstance(genre, str) and genre else None
|
||||
|
||||
@staticmethod
|
||||
def _artist(
|
||||
track: SoundCloudTrack,
|
||||
) -> tuple[str | None, str | None, str | None]:
|
||||
user = track.get("user") or {}
|
||||
credit = track.get("metadata_artist")
|
||||
uploader_urn = user.get("urn")
|
||||
if credit:
|
||||
return credit, None, uploader_urn
|
||||
return user.get("username"), uploader_urn, uploader_urn
|
||||
|
||||
def _track_info(
|
||||
self,
|
||||
track: SoundCloudTrack,
|
||||
*,
|
||||
index: int | None = None,
|
||||
total: int | None = None,
|
||||
album: str | None = None,
|
||||
) -> TrackInfo:
|
||||
artist, artist_id, artist_urn = self._artist(track)
|
||||
urn = track["urn"]
|
||||
duration = track.get("duration")
|
||||
bpm = track.get("bpm")
|
||||
return TrackInfo(
|
||||
title=track.get("title"),
|
||||
track_id=urn,
|
||||
soundcloud_track_urn=urn,
|
||||
artist=artist,
|
||||
artist_id=artist_id,
|
||||
soundcloud_artist_urn=artist_urn,
|
||||
album=album or track.get("release"),
|
||||
length=duration / 1000 if duration is not None else None,
|
||||
index=index,
|
||||
medium=1 if index is not None else None,
|
||||
medium_index=index,
|
||||
medium_total=total,
|
||||
year=track.get("release_year"),
|
||||
month=track.get("release_month"),
|
||||
day=track.get("release_day"),
|
||||
genres=self._genres(track),
|
||||
label=track.get("label_name"),
|
||||
isrc=track.get("isrc"),
|
||||
bpm=str(bpm) if bpm is not None else None,
|
||||
initial_key=track.get("key_signature"),
|
||||
cover_art_url=track.get("artwork_url"),
|
||||
data_source=self.data_source,
|
||||
data_url=track.get("permalink_url"),
|
||||
media="Digital Media",
|
||||
)
|
||||
|
||||
def track_for_id(self, track_id: str) -> TrackInfo | None:
|
||||
if track := self.api.get_track(track_id):
|
||||
return self._track_info(track)
|
||||
return None
|
||||
|
||||
def _album_info(self, playlist: SoundCloudPlaylist) -> AlbumInfo:
|
||||
track_data = playlist.get("tracks") or []
|
||||
total = len(track_data)
|
||||
tracks: list[TrackInfo] = []
|
||||
artist_credits: dict[str, tuple[str, set[str]]] = {}
|
||||
for index, raw_track in enumerate(track_data, start=1):
|
||||
track = self._track_info(
|
||||
raw_track, index=index, total=total, album=playlist.get("title")
|
||||
)
|
||||
tracks.append(track)
|
||||
if track.artist:
|
||||
_, artist_ids = artist_credits.setdefault(
|
||||
track.artist.casefold(), (track.artist, set())
|
||||
)
|
||||
if track.artist_id:
|
||||
artist_ids.add(track.artist_id)
|
||||
|
||||
user = playlist.get("user") or {}
|
||||
uploader_urn = user.get("urn")
|
||||
va = len(artist_credits) > 1
|
||||
artist: str | None
|
||||
artist_id: str | None
|
||||
if va:
|
||||
artist = config["va_name"].as_str()
|
||||
artist_id = None
|
||||
elif artist_credits:
|
||||
artist, artist_ids = next(iter(artist_credits.values()))
|
||||
artist_id = next(iter(artist_ids)) if len(artist_ids) == 1 else None
|
||||
else:
|
||||
artist = user.get("username")
|
||||
artist_id = uploader_urn
|
||||
|
||||
urn = playlist["urn"]
|
||||
playlist_type = playlist.get("playlist_type") or "playlist"
|
||||
return AlbumInfo(
|
||||
album=playlist.get("title"),
|
||||
album_id=urn,
|
||||
soundcloud_playlist_urn=urn,
|
||||
artist=artist,
|
||||
artist_id=artist_id,
|
||||
soundcloud_artist_urn=uploader_urn,
|
||||
tracks=tracks,
|
||||
va=va,
|
||||
albumtype=playlist_type,
|
||||
albumtypes=[playlist_type],
|
||||
barcode=playlist.get("ean"),
|
||||
year=playlist.get("release_year"),
|
||||
month=playlist.get("release_month"),
|
||||
day=playlist.get("release_day"),
|
||||
genres=self._genres(playlist),
|
||||
label=playlist.get("label_name"),
|
||||
mediums=1,
|
||||
cover_art_url=playlist.get("artwork_url"),
|
||||
data_source=self.data_source,
|
||||
data_url=playlist.get("permalink_url"),
|
||||
media="Digital Media",
|
||||
)
|
||||
|
||||
def album_for_id(self, album_id: str) -> AlbumInfo | None:
|
||||
if playlist := self.api.get_playlist(album_id):
|
||||
return self._album_info(playlist)
|
||||
return None
|
||||
464
beetsplug/soundcloud/api.py
Normal file
464
beetsplug/soundcloud/api.py
Normal file
@@ -0,0 +1,464 @@
|
||||
"""Access public SoundCloud metadata with renewable application credentials."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
from contextlib import contextmanager, suppress
|
||||
from http import HTTPStatus
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any, Literal, TypeGuard, overload
|
||||
|
||||
import requests
|
||||
|
||||
from beets import __version__
|
||||
from beets.exceptions import UserError
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable, Iterator
|
||||
|
||||
from .api_types import (
|
||||
SoundCloudCollection,
|
||||
SoundCloudPlaylist,
|
||||
SoundCloudResource,
|
||||
SoundCloudToken,
|
||||
SoundCloudTrack,
|
||||
)
|
||||
|
||||
API_URL = "https://api.soundcloud.com"
|
||||
AUTH_URL = "https://secure.soundcloud.com/oauth/token"
|
||||
TOKEN_EXPIRY_MARGIN = 30
|
||||
TOKEN_LOCK_POLL_INTERVAL = 0.05
|
||||
TOKEN_LOCK_TIMEOUT = 15
|
||||
TOKEN_LOCK_STALE_AFTER = 30
|
||||
|
||||
|
||||
class SoundCloudAPIError(UserError):
|
||||
"""Report an API failure that a user can act on."""
|
||||
|
||||
|
||||
class SoundCloudNotFoundError(SoundCloudAPIError):
|
||||
"""Indicate that a requested public resource does not exist."""
|
||||
|
||||
|
||||
class SoundCloudAPI:
|
||||
"""Retrieve public SoundCloud resources and keep OAuth tokens current."""
|
||||
|
||||
def __init__(
|
||||
self, client_id: str, client_secret: str, token_path: str | Path
|
||||
) -> None:
|
||||
self.client_id = client_id
|
||||
self.client_secret = client_secret
|
||||
self.token_path = Path(token_path)
|
||||
self.session = requests.Session()
|
||||
self.session.headers["User-Agent"] = (
|
||||
f"beets/{__version__} https://beets.io/"
|
||||
)
|
||||
self._token = self._load_token()
|
||||
self._token_lock = threading.Lock()
|
||||
|
||||
def _load_token(self) -> SoundCloudToken:
|
||||
try:
|
||||
with self.token_path.open() as token_file:
|
||||
token = json.load(token_file)
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return {}
|
||||
if not isinstance(token, dict):
|
||||
return {}
|
||||
access_token = token.get("access_token")
|
||||
refresh_token = token.get("refresh_token")
|
||||
expires_at = token.get("expires_at")
|
||||
if (
|
||||
not isinstance(access_token, str)
|
||||
or not access_token
|
||||
or not isinstance(refresh_token, str)
|
||||
or not refresh_token
|
||||
or not isinstance(expires_at, (int, float))
|
||||
or isinstance(expires_at, bool)
|
||||
):
|
||||
return {}
|
||||
return {
|
||||
"access_token": access_token,
|
||||
"refresh_token": refresh_token,
|
||||
"expires_at": float(expires_at),
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _lock_is_stale(lock_path: Path) -> bool:
|
||||
try:
|
||||
return (
|
||||
time.time() - lock_path.stat().st_mtime > TOKEN_LOCK_STALE_AFTER
|
||||
)
|
||||
except FileNotFoundError:
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _create_token_lock(lock_path: Path) -> int | None:
|
||||
try:
|
||||
return os.open(
|
||||
lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600
|
||||
)
|
||||
except FileExistsError:
|
||||
return None
|
||||
|
||||
def _acquire_token_lock(self, lock_path: Path) -> int:
|
||||
deadline = time.monotonic() + TOKEN_LOCK_TIMEOUT
|
||||
while True:
|
||||
descriptor = self._create_token_lock(lock_path)
|
||||
if descriptor is not None:
|
||||
return descriptor
|
||||
if self._lock_is_stale(lock_path):
|
||||
with suppress(FileNotFoundError):
|
||||
lock_path.unlink()
|
||||
continue
|
||||
if time.monotonic() >= deadline:
|
||||
raise SoundCloudAPIError(
|
||||
"Timed out waiting for the SoundCloud token cache lock"
|
||||
)
|
||||
time.sleep(TOKEN_LOCK_POLL_INTERVAL)
|
||||
|
||||
@contextmanager
|
||||
def _token_file_lock(self) -> Iterator[None]:
|
||||
lock_path = self.token_path.with_suffix(
|
||||
f"{self.token_path.suffix}.lock"
|
||||
)
|
||||
self.token_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
descriptor = self._acquire_token_lock(lock_path)
|
||||
|
||||
try:
|
||||
os.write(descriptor, str(os.getpid()).encode())
|
||||
yield
|
||||
finally:
|
||||
os.close(descriptor)
|
||||
with suppress(FileNotFoundError):
|
||||
lock_path.unlink()
|
||||
|
||||
def _save_token(self, token: SoundCloudToken) -> None:
|
||||
self.token_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
descriptor, temporary_name = tempfile.mkstemp(
|
||||
dir=self.token_path.parent, prefix=f".{self.token_path.name}."
|
||||
)
|
||||
temporary_path = Path(temporary_name)
|
||||
try:
|
||||
temporary_path.chmod(0o600)
|
||||
with os.fdopen(descriptor, "w") as token_file:
|
||||
json.dump(token, token_file, indent=2)
|
||||
token_file.flush()
|
||||
os.fsync(token_file.fileno())
|
||||
temporary_path.replace(self.token_path)
|
||||
except BaseException:
|
||||
with suppress(OSError):
|
||||
os.close(descriptor)
|
||||
with suppress(FileNotFoundError):
|
||||
temporary_path.unlink()
|
||||
raise
|
||||
|
||||
def _require_credentials(self) -> None:
|
||||
if not self.client_id or not self.client_secret:
|
||||
raise SoundCloudAPIError(
|
||||
"SoundCloud client_id and client_secret must be configured"
|
||||
)
|
||||
|
||||
def _token_is_current(self) -> bool:
|
||||
return bool(self._token.get("access_token")) and (
|
||||
self._token.get("expires_at", 0) > time.time() + TOKEN_EXPIRY_MARGIN
|
||||
)
|
||||
|
||||
def _request_token(self, *, refresh: bool) -> SoundCloudToken:
|
||||
self._require_credentials()
|
||||
if refresh:
|
||||
data = {
|
||||
"grant_type": "refresh_token",
|
||||
"client_id": self.client_id,
|
||||
"client_secret": self.client_secret,
|
||||
"refresh_token": self._token["refresh_token"],
|
||||
}
|
||||
auth = None
|
||||
else:
|
||||
data = {"grant_type": "client_credentials"}
|
||||
auth = (self.client_id, self.client_secret)
|
||||
|
||||
try:
|
||||
response = self.session.post(
|
||||
AUTH_URL, data=data, auth=auth, timeout=10
|
||||
)
|
||||
response.raise_for_status()
|
||||
response_payload = response.json()
|
||||
except (requests.RequestException, ValueError) as exc:
|
||||
raise SoundCloudAPIError(
|
||||
f"SoundCloud authentication failed: {exc}"
|
||||
) from exc
|
||||
|
||||
if not isinstance(response_payload, dict):
|
||||
raise SoundCloudAPIError(
|
||||
"SoundCloud authentication response has an unexpected shape"
|
||||
)
|
||||
access_token = response_payload.get("access_token")
|
||||
refresh_token = response_payload.get("refresh_token")
|
||||
expires_in = response_payload.get("expires_in")
|
||||
if (
|
||||
not isinstance(access_token, str)
|
||||
or not access_token
|
||||
or not isinstance(refresh_token, str)
|
||||
or not refresh_token
|
||||
or not isinstance(expires_in, int)
|
||||
or isinstance(expires_in, bool)
|
||||
or expires_in <= 0
|
||||
):
|
||||
raise SoundCloudAPIError(
|
||||
"SoundCloud authentication response has invalid token data"
|
||||
)
|
||||
payload: SoundCloudToken = {
|
||||
"access_token": access_token,
|
||||
"refresh_token": refresh_token,
|
||||
"expires_in": expires_in,
|
||||
"expires_at": time.time() + expires_in,
|
||||
}
|
||||
scope = response_payload.get("scope")
|
||||
if isinstance(scope, str):
|
||||
payload["scope"] = scope
|
||||
token_type = response_payload.get("token_type")
|
||||
if isinstance(token_type, str):
|
||||
payload["token_type"] = token_type
|
||||
self._token = payload
|
||||
self._save_token(payload)
|
||||
return payload
|
||||
|
||||
def _access_token(self, *, rejected_token: str | None = None) -> str:
|
||||
with self._token_lock:
|
||||
if rejected_token is None and self._token_is_current():
|
||||
return self._token["access_token"]
|
||||
with self._token_file_lock():
|
||||
cached_token = self._load_token()
|
||||
cached_access_token = cached_token.get("access_token")
|
||||
if (
|
||||
isinstance(cached_access_token, str)
|
||||
and cached_access_token
|
||||
and cached_access_token != rejected_token
|
||||
and cached_token.get("expires_at", 0)
|
||||
> time.time() + TOKEN_EXPIRY_MARGIN
|
||||
):
|
||||
self._token = cached_token
|
||||
return cached_access_token
|
||||
if cached_token:
|
||||
self._token = cached_token
|
||||
refresh = bool(self._token.get("refresh_token"))
|
||||
return self._request_token(refresh=refresh)["access_token"]
|
||||
|
||||
def _request(
|
||||
self,
|
||||
url: str,
|
||||
*,
|
||||
params: dict[str, Any] | None = None,
|
||||
retry_unauthorized: bool = True,
|
||||
) -> requests.Response:
|
||||
access_token = self._access_token()
|
||||
response = self.session.get(
|
||||
url,
|
||||
headers={"Authorization": f"OAuth {access_token}"},
|
||||
params=params,
|
||||
timeout=10,
|
||||
)
|
||||
if (
|
||||
response.status_code == HTTPStatus.UNAUTHORIZED
|
||||
and retry_unauthorized
|
||||
):
|
||||
response = self.session.get(
|
||||
url,
|
||||
headers={
|
||||
"Authorization": (
|
||||
"OAuth "
|
||||
+ self._access_token(rejected_token=access_token)
|
||||
)
|
||||
},
|
||||
params=params,
|
||||
timeout=10,
|
||||
)
|
||||
if response.status_code == HTTPStatus.TOO_MANY_REQUESTS:
|
||||
delay = response.headers.get("Retry-After", "an unknown interval")
|
||||
raise SoundCloudAPIError(
|
||||
f"SoundCloud rate limit reached; retry after {delay} seconds"
|
||||
)
|
||||
if response.status_code == HTTPStatus.NOT_FOUND:
|
||||
raise SoundCloudNotFoundError(
|
||||
f"SoundCloud resource was not found: {url}"
|
||||
)
|
||||
try:
|
||||
response.raise_for_status()
|
||||
except requests.RequestException as exc:
|
||||
raise SoundCloudAPIError(
|
||||
f"SoundCloud API request failed: {exc}"
|
||||
) from exc
|
||||
return response
|
||||
|
||||
def _get_json(
|
||||
self, url: str, *, params: dict[str, Any] | None = None
|
||||
) -> dict[str, Any]:
|
||||
try:
|
||||
payload = self._request(url, params=params).json()
|
||||
except ValueError as exc:
|
||||
raise SoundCloudAPIError(
|
||||
"SoundCloud returned an invalid JSON response"
|
||||
) from exc
|
||||
if not isinstance(payload, dict):
|
||||
raise SoundCloudAPIError(
|
||||
"SoundCloud returned an unexpected response shape"
|
||||
)
|
||||
return payload
|
||||
|
||||
def _get_paginated(
|
||||
self,
|
||||
url: str,
|
||||
*,
|
||||
params: dict[str, Any],
|
||||
limit: int,
|
||||
accept: Callable[[SoundCloudResource], bool],
|
||||
) -> list[SoundCloudResource]:
|
||||
resources: list[SoundCloudResource] = []
|
||||
request_params: dict[str, Any] | None = params
|
||||
while url and len(resources) < limit:
|
||||
page: SoundCloudCollection = self._get_json( # type: ignore[assignment]
|
||||
url, params=request_params
|
||||
)
|
||||
collection = page.get("collection")
|
||||
if not isinstance(collection, list):
|
||||
raise SoundCloudAPIError(
|
||||
"SoundCloud collection response is missing collection"
|
||||
)
|
||||
for resource in collection:
|
||||
if accept(resource):
|
||||
resources.append(resource)
|
||||
if len(resources) == limit:
|
||||
break
|
||||
next_href = page.get("next_href")
|
||||
url = next_href if isinstance(next_href, str) else ""
|
||||
request_params = None
|
||||
return resources
|
||||
|
||||
def _get_collection(
|
||||
self,
|
||||
path: str,
|
||||
query: str,
|
||||
limit: int,
|
||||
accept: Callable[[SoundCloudResource], bool],
|
||||
) -> list[SoundCloudResource]:
|
||||
return self._get_paginated(
|
||||
f"{API_URL}/{path}",
|
||||
params={"q": query, "limit": limit, "linked_partitioning": True},
|
||||
limit=limit,
|
||||
accept=accept,
|
||||
)
|
||||
|
||||
def search_tracks(self, query: str, limit: int) -> list[SoundCloudTrack]:
|
||||
return [
|
||||
resource
|
||||
for resource in self._get_collection(
|
||||
"tracks", query, limit, self._is_track
|
||||
)
|
||||
if self._is_track(resource)
|
||||
]
|
||||
|
||||
def search_playlists(
|
||||
self, query: str, limit: int, *, playlist_type: str | None = None
|
||||
) -> list[SoundCloudPlaylist]:
|
||||
def accept(resource: SoundCloudResource) -> bool:
|
||||
return self._is_playlist(resource) and (
|
||||
playlist_type is None
|
||||
or resource.get("playlist_type") == playlist_type
|
||||
)
|
||||
|
||||
return [
|
||||
resource
|
||||
for resource in self._get_collection(
|
||||
"playlists", query, limit, accept
|
||||
)
|
||||
if self._is_playlist(resource)
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _is_track(resource: SoundCloudResource) -> TypeGuard[SoundCloudTrack]:
|
||||
return resource.get("kind") == "track"
|
||||
|
||||
@staticmethod
|
||||
def _is_playlist(
|
||||
resource: SoundCloudResource,
|
||||
) -> TypeGuard[SoundCloudPlaylist]:
|
||||
return resource.get("kind") == "playlist"
|
||||
|
||||
def resolve(self, url: str) -> SoundCloudResource | None:
|
||||
try:
|
||||
resource: SoundCloudResource = self._get_json( # type: ignore[assignment]
|
||||
f"{API_URL}/resolve", params={"url": url}
|
||||
)
|
||||
except SoundCloudNotFoundError:
|
||||
return None
|
||||
self._validate_resource(resource)
|
||||
return resource
|
||||
|
||||
@staticmethod
|
||||
def _validate_resource(resource: SoundCloudResource) -> None:
|
||||
if not resource.get("urn"):
|
||||
raise SoundCloudAPIError(
|
||||
"SoundCloud resource response is missing its urn"
|
||||
)
|
||||
|
||||
@overload
|
||||
def _get_resource(
|
||||
self, identifier: str, kind: Literal["track"]
|
||||
) -> SoundCloudTrack | None: ...
|
||||
|
||||
@overload
|
||||
def _get_resource(
|
||||
self, identifier: str, kind: Literal["playlist"]
|
||||
) -> SoundCloudPlaylist | None: ...
|
||||
|
||||
def _get_resource(
|
||||
self, identifier: str, kind: Literal["track", "playlist"]
|
||||
) -> SoundCloudResource | None:
|
||||
if identifier.startswith(("http://", "https://")):
|
||||
resolved = self.resolve(identifier)
|
||||
if not resolved or resolved.get("kind") != kind:
|
||||
return None
|
||||
identifier = resolved["urn"]
|
||||
elif not identifier.startswith(f"soundcloud:{kind}s:"):
|
||||
return None
|
||||
|
||||
try:
|
||||
resource: SoundCloudResource = self._get_json( # type: ignore[assignment]
|
||||
f"{API_URL}/{kind}s/{identifier}",
|
||||
params={"show_tracks": True} if kind == "playlist" else None,
|
||||
)
|
||||
except SoundCloudNotFoundError:
|
||||
return None
|
||||
self._validate_resource(resource)
|
||||
if resource.get("kind") != kind:
|
||||
return None
|
||||
if kind == "playlist" and self._is_playlist(resource):
|
||||
self._complete_playlist_tracks(resource, identifier)
|
||||
return resource
|
||||
|
||||
def _complete_playlist_tracks(
|
||||
self, playlist: SoundCloudPlaylist, identifier: str
|
||||
) -> None:
|
||||
tracks = playlist.get("tracks") or []
|
||||
track_count = playlist.get("track_count", len(tracks))
|
||||
if len(tracks) >= track_count:
|
||||
return
|
||||
resources = self._get_paginated(
|
||||
f"{API_URL}/playlists/{identifier}/tracks",
|
||||
params={"linked_partitioning": True},
|
||||
limit=track_count,
|
||||
accept=self._is_track,
|
||||
)
|
||||
playlist["tracks"] = [
|
||||
track for track in resources if self._is_track(track)
|
||||
]
|
||||
|
||||
def get_track(self, identifier: str) -> SoundCloudTrack | None:
|
||||
return self._get_resource(identifier, "track")
|
||||
|
||||
def get_playlist(self, identifier: str) -> SoundCloudPlaylist | None:
|
||||
return self._get_resource(identifier, "playlist")
|
||||
66
beetsplug/soundcloud/api_types.py
Normal file
66
beetsplug/soundcloud/api_types.py
Normal file
@@ -0,0 +1,66 @@
|
||||
"""Describe the SoundCloud response fields consumed by the plugin."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TypedDict
|
||||
|
||||
|
||||
class SoundCloudUser(TypedDict, total=False):
|
||||
urn: str
|
||||
username: str
|
||||
|
||||
|
||||
class SoundCloudTrack(TypedDict, total=False):
|
||||
artwork_url: str | None
|
||||
bpm: float | None
|
||||
duration: int
|
||||
genre: str | None
|
||||
isrc: str | None
|
||||
key_signature: str | None
|
||||
kind: str
|
||||
label_name: str | None
|
||||
metadata_artist: str | None
|
||||
permalink_url: str
|
||||
release: str | None
|
||||
release_day: int | None
|
||||
release_month: int | None
|
||||
release_year: int | None
|
||||
title: str
|
||||
urn: str
|
||||
user: SoundCloudUser
|
||||
|
||||
|
||||
class SoundCloudPlaylist(TypedDict, total=False):
|
||||
artwork_url: str | None
|
||||
ean: str | None
|
||||
genre: str | None
|
||||
kind: str
|
||||
label_name: str | None
|
||||
permalink_url: str
|
||||
playlist_type: str
|
||||
release: str | None
|
||||
release_day: int | None
|
||||
release_month: int | None
|
||||
release_year: int | None
|
||||
title: str
|
||||
track_count: int
|
||||
tracks: list[SoundCloudTrack]
|
||||
urn: str
|
||||
user: SoundCloudUser
|
||||
|
||||
|
||||
SoundCloudResource = SoundCloudTrack | SoundCloudPlaylist
|
||||
|
||||
|
||||
class SoundCloudCollection(TypedDict, total=False):
|
||||
collection: list[SoundCloudResource]
|
||||
next_href: str | None
|
||||
|
||||
|
||||
class SoundCloudToken(TypedDict, total=False):
|
||||
access_token: str
|
||||
expires_at: float
|
||||
expires_in: int
|
||||
refresh_token: str
|
||||
scope: str
|
||||
token_type: str
|
||||
@@ -17,6 +17,9 @@ New features
|
||||
new tracks, and keeps the album together rather than splitting it. The option
|
||||
is available both through configuration and from the interactive duplicate
|
||||
prompt. :bug:`4471`
|
||||
- :doc:`plugins/soundcloud`: Add public SoundCloud track and release matches to
|
||||
the autotagger, including direct URL and URN lookup, renewable application
|
||||
authentication, and stable catalog metadata.
|
||||
|
||||
Bug fixes
|
||||
~~~~~~~~~
|
||||
|
||||
@@ -120,6 +120,7 @@ databases. They share the following configuration options:
|
||||
rewrite
|
||||
scrub
|
||||
smartplaylist
|
||||
soundcloud
|
||||
sonosupdate
|
||||
spotify
|
||||
subsonicplaylist
|
||||
@@ -161,6 +162,9 @@ Autotagger Extensions
|
||||
:doc:`spotify <spotify>`
|
||||
Search for releases in the Spotify_ database.
|
||||
|
||||
:doc:`soundcloud <soundcloud>`
|
||||
Search for public tracks and releases in the SoundCloud_ catalog.
|
||||
|
||||
:doc:`tidal <tidal>`
|
||||
Search for releases in the Tidal_ catalog.
|
||||
|
||||
@@ -172,6 +176,8 @@ Autotagger Extensions
|
||||
|
||||
.. _spotify: https://open.spotify.com/
|
||||
|
||||
.. _soundcloud: https://soundcloud.com/
|
||||
|
||||
.. _tidal: https://tidal.com/
|
||||
|
||||
Metadata
|
||||
|
||||
113
docs/plugins/soundcloud.rst
Normal file
113
docs/plugins/soundcloud.rst
Normal file
@@ -0,0 +1,113 @@
|
||||
SoundCloud Plugin
|
||||
=================
|
||||
|
||||
The ``soundcloud`` plugin provides metadata matches for public SoundCloud_
|
||||
tracks and releases during import. It uses SoundCloud's documented public API;
|
||||
it does not download or stream audio.
|
||||
|
||||
.. _soundcloud: https://soundcloud.com/
|
||||
|
||||
Requirements
|
||||
------------
|
||||
|
||||
SoundCloud requires an Artist Pro subscription to register an API application.
|
||||
Create an application using the `SoundCloud registration instructions`_, then
|
||||
copy its client ID and client secret into your beets configuration.
|
||||
|
||||
.. _soundcloud registration instructions: https://developers.soundcloud.com/docs/api/register-app
|
||||
|
||||
The plugin uses application authentication and can access public resources
|
||||
only. It does not ask you to sign in to a SoundCloud user account and cannot
|
||||
match private tracks or playlists.
|
||||
|
||||
Configuration
|
||||
-------------
|
||||
|
||||
Enable the plugin and configure your application credentials:
|
||||
|
||||
.. code-block:: yaml
|
||||
|
||||
plugins: soundcloud
|
||||
|
||||
soundcloud:
|
||||
client_id: YOUR_CLIENT_ID
|
||||
client_secret: YOUR_CLIENT_SECRET
|
||||
tokenfile: soundcloud_token.json
|
||||
search_limit: 5
|
||||
data_source_mismatch_penalty: 0.5
|
||||
|
||||
The plugin obtains an access token when it first needs the API. It saves the
|
||||
access and refresh tokens in the configured token file and refreshes them
|
||||
automatically. Keep both the client secret and token file private.
|
||||
|
||||
.. conf:: client_id
|
||||
|
||||
Client ID issued for your SoundCloud application.
|
||||
|
||||
.. conf:: client_secret
|
||||
|
||||
Client secret issued for your SoundCloud application.
|
||||
|
||||
.. conf:: tokenfile
|
||||
:default: soundcloud_token.json
|
||||
|
||||
File used to store renewable application tokens. Relative paths are
|
||||
resolved inside the beets configuration directory.
|
||||
|
||||
.. include:: ./shared_metadata_source_config.rst
|
||||
|
||||
Matching Behavior
|
||||
-----------------
|
||||
|
||||
SoundCloud represents releases and user-created sets with the same playlist
|
||||
resource. Automatic album searches include only sets SoundCloud marks as
|
||||
albums, which prevents ordinary playlists from appearing as release matches.
|
||||
|
||||
An explicit URL or URN lookup also accepts an ordinary public playlist. This
|
||||
allows you to intentionally tag a local group of files from a set:
|
||||
|
||||
::
|
||||
|
||||
Enter release ID: https://soundcloud.com/artist/sets/release
|
||||
Enter release ID: soundcloud:playlists:123456
|
||||
|
||||
Track URLs and URNs work for singleton imports:
|
||||
|
||||
::
|
||||
|
||||
Enter release ID: https://soundcloud.com/artist/track
|
||||
Enter release ID: soundcloud:tracks:123456
|
||||
|
||||
Track credits use SoundCloud's metadata artist when present and fall back to the
|
||||
uploader username. A set whose tracks have different credits is tagged as a
|
||||
various-artists release. Track numbering follows the order of the SoundCloud
|
||||
set.
|
||||
|
||||
Metadata
|
||||
--------
|
||||
|
||||
The plugin imports titles, artists, track order, duration, release dates,
|
||||
genres, labels, ISRCs, BPM, musical key, source URLs, and artwork URLs when
|
||||
SoundCloud provides them. The :doc:`fetchart` plugin can use the release artwork
|
||||
URL during import.
|
||||
|
||||
It also stores these flexible attributes:
|
||||
|
||||
.. list-table::
|
||||
:header-rows: 1
|
||||
|
||||
- - Attribute
|
||||
- Stored on
|
||||
- Description
|
||||
- - ``soundcloud_track_urn``
|
||||
- Item
|
||||
- SoundCloud track identifier
|
||||
- - ``soundcloud_playlist_urn``
|
||||
- Album
|
||||
- SoundCloud playlist or set identifier
|
||||
- - ``soundcloud_artist_urn``
|
||||
- Item and album
|
||||
- SoundCloud uploader identifier
|
||||
|
||||
Play, like, repost, comment, and download counts are deliberately omitted
|
||||
because they change independently of an autotagging import.
|
||||
511
test/plugins/test_soundcloud.py
Normal file
511
test/plugins/test_soundcloud.py
Normal file
@@ -0,0 +1,511 @@
|
||||
"""Tests for the SoundCloud metadata source."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from urllib.parse import parse_qs
|
||||
|
||||
import pytest
|
||||
|
||||
from beets import config
|
||||
from beets.library import Item
|
||||
from beetsplug.soundcloud import SoundCloudPlugin
|
||||
from beetsplug.soundcloud.api import SoundCloudAPI, SoundCloudAPIError
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
API_URL = "https://api.soundcloud.com"
|
||||
AUTH_URL = "https://secure.soundcloud.com/oauth/token"
|
||||
DEFAULT_BEARER = "access-token"
|
||||
DEFAULT_RENEWAL = "refresh-token"
|
||||
NEW_BEARER = "new-access"
|
||||
REFRESHED_BEARER = "refreshed-access"
|
||||
REPLACEMENT_RENEWAL = "replacement-refresh"
|
||||
TRACK_DURATION_SECONDS = 123.0
|
||||
TRACK_BPM = 128.0
|
||||
|
||||
|
||||
def _write_token(
|
||||
path: Path,
|
||||
*,
|
||||
access_token: str = DEFAULT_BEARER,
|
||||
refresh_token: str = DEFAULT_RENEWAL,
|
||||
expires_at: float = 4_102_444_800,
|
||||
) -> None:
|
||||
path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"access_token": access_token,
|
||||
"refresh_token": refresh_token,
|
||||
"expires_at": expires_at,
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _user(
|
||||
username: str = "Uploader", urn: str = "soundcloud:users:10"
|
||||
) -> dict[str, Any]:
|
||||
return {"username": username, "urn": urn}
|
||||
|
||||
|
||||
def _track(
|
||||
urn: str = "soundcloud:tracks:1",
|
||||
*,
|
||||
title: str = "Track",
|
||||
artist: str | None = "Artist",
|
||||
user: dict[str, Any] | None = None,
|
||||
**fields: Any,
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"kind": "track",
|
||||
"urn": urn,
|
||||
"title": title,
|
||||
"duration": 123_000,
|
||||
"metadata_artist": artist,
|
||||
"permalink_url": "https://soundcloud.com/u/track",
|
||||
"user": user or _user(),
|
||||
**fields,
|
||||
}
|
||||
|
||||
|
||||
def _playlist(
|
||||
urn: str = "soundcloud:playlists:2",
|
||||
*,
|
||||
playlist_type: str = "album",
|
||||
tracks: list[dict[str, Any]] | None = None,
|
||||
**fields: Any,
|
||||
) -> dict[str, Any]:
|
||||
tracks = tracks if tracks is not None else [_track()]
|
||||
return {
|
||||
"kind": "playlist",
|
||||
"urn": urn,
|
||||
"title": "Release",
|
||||
"playlist_type": playlist_type,
|
||||
"permalink_url": "https://soundcloud.com/u/sets/release",
|
||||
"track_count": len(tracks),
|
||||
"tracks": tracks,
|
||||
"user": _user(),
|
||||
**fields,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def token_path(tmp_path: Path) -> Path:
|
||||
return tmp_path / "soundcloud_token.json"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def api(token_path: Path) -> SoundCloudAPI:
|
||||
_write_token(token_path)
|
||||
return SoundCloudAPI("client-id", "client-secret", token_path)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def plugin(token_path: Path) -> SoundCloudPlugin:
|
||||
_write_token(token_path)
|
||||
plugin = SoundCloudPlugin()
|
||||
plugin.config["client_id"].set("client-id")
|
||||
plugin.config["client_secret"].set("client-secret")
|
||||
plugin.config["tokenfile"].set(str(token_path))
|
||||
return plugin
|
||||
|
||||
|
||||
class TestAuthentication:
|
||||
def test_client_credentials_are_cached(
|
||||
self, token_path, requests_mock
|
||||
) -> None:
|
||||
requests_mock.post(
|
||||
AUTH_URL,
|
||||
json={
|
||||
"access_token": NEW_BEARER,
|
||||
"refresh_token": "new-refresh",
|
||||
"expires_in": 3600,
|
||||
},
|
||||
)
|
||||
requests_mock.get(f"{API_URL}/tracks", json={"collection": []})
|
||||
api = SoundCloudAPI("client-id", "client-secret", token_path)
|
||||
|
||||
assert api.search_tracks("Title", 5) == []
|
||||
|
||||
token_request = requests_mock.request_history[0]
|
||||
assert token_request.headers["Authorization"].startswith("Basic ")
|
||||
assert parse_qs(token_request.text) == {
|
||||
"grant_type": ["client_credentials"]
|
||||
}
|
||||
assert json.loads(token_path.read_text())["access_token"] == NEW_BEARER
|
||||
|
||||
def test_expired_token_is_refreshed(
|
||||
self, token_path, requests_mock
|
||||
) -> None:
|
||||
_write_token(token_path, expires_at=0)
|
||||
requests_mock.post(
|
||||
AUTH_URL,
|
||||
json={
|
||||
"access_token": REFRESHED_BEARER,
|
||||
"refresh_token": REPLACEMENT_RENEWAL,
|
||||
"expires_in": 3600,
|
||||
},
|
||||
)
|
||||
requests_mock.get(f"{API_URL}/tracks", json={"collection": []})
|
||||
api = SoundCloudAPI("client-id", "client-secret", token_path)
|
||||
|
||||
api.search_tracks("Title", 5)
|
||||
|
||||
token_request = requests_mock.request_history[0]
|
||||
assert parse_qs(token_request.text) == {
|
||||
"client_id": ["client-id"],
|
||||
"client_secret": ["client-secret"],
|
||||
"grant_type": ["refresh_token"],
|
||||
"refresh_token": ["refresh-token"],
|
||||
}
|
||||
saved = json.loads(token_path.read_text())
|
||||
assert saved["access_token"] == REFRESHED_BEARER
|
||||
assert saved["refresh_token"] == REPLACEMENT_RENEWAL
|
||||
|
||||
def test_unauthorized_request_refreshes_once(
|
||||
self, api, requests_mock
|
||||
) -> None:
|
||||
requests_mock.get(
|
||||
f"{API_URL}/tracks",
|
||||
[
|
||||
{"status_code": 401},
|
||||
{"json": {"collection": [_track()]}, "status_code": 200},
|
||||
],
|
||||
)
|
||||
requests_mock.post(
|
||||
AUTH_URL,
|
||||
json={
|
||||
"access_token": REFRESHED_BEARER,
|
||||
"refresh_token": REPLACEMENT_RENEWAL,
|
||||
"expires_in": 3600,
|
||||
},
|
||||
)
|
||||
|
||||
assert len(api.search_tracks("Title", 5)) == 1
|
||||
|
||||
track_requests = [
|
||||
request
|
||||
for request in requests_mock.request_history
|
||||
if request.path == "/tracks"
|
||||
]
|
||||
assert track_requests[0].headers["Authorization"] == (
|
||||
f"OAuth {DEFAULT_BEARER}"
|
||||
)
|
||||
assert track_requests[1].headers["Authorization"] == (
|
||||
f"OAuth {REFRESHED_BEARER}"
|
||||
)
|
||||
|
||||
def test_second_client_reuses_token_refreshed_by_first(
|
||||
self, token_path, requests_mock
|
||||
) -> None:
|
||||
_write_token(token_path)
|
||||
first_api = SoundCloudAPI("client-id", "client-secret", token_path)
|
||||
second_api = SoundCloudAPI("client-id", "client-secret", token_path)
|
||||
requests_mock.get(
|
||||
f"{API_URL}/tracks",
|
||||
[
|
||||
{"status_code": 401},
|
||||
{"json": {"collection": []}, "status_code": 200},
|
||||
{"status_code": 401},
|
||||
{"json": {"collection": []}, "status_code": 200},
|
||||
],
|
||||
)
|
||||
requests_mock.post(
|
||||
AUTH_URL,
|
||||
json={
|
||||
"access_token": REFRESHED_BEARER,
|
||||
"refresh_token": REPLACEMENT_RENEWAL,
|
||||
"expires_in": 3600,
|
||||
},
|
||||
)
|
||||
|
||||
first_api.search_tracks("Title", 5)
|
||||
second_api.search_tracks("Title", 5)
|
||||
|
||||
token_requests = [
|
||||
request
|
||||
for request in requests_mock.request_history
|
||||
if request.method == "POST"
|
||||
]
|
||||
assert len(token_requests) == 1
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"payload",
|
||||
[
|
||||
[],
|
||||
{"access_token": "access", "expires_in": 3600},
|
||||
{
|
||||
"access_token": "access",
|
||||
"refresh_token": "refresh",
|
||||
"expires_in": "one hour",
|
||||
},
|
||||
],
|
||||
)
|
||||
def test_malformed_token_response_fails_clearly(
|
||||
self, token_path, requests_mock, payload
|
||||
) -> None:
|
||||
requests_mock.post(AUTH_URL, json=payload)
|
||||
api = SoundCloudAPI("client-id", "client-secret", token_path)
|
||||
|
||||
with pytest.raises(SoundCloudAPIError, match="authentication response"):
|
||||
api.search_tracks("Title", 5)
|
||||
|
||||
def test_missing_credentials_fail_before_request(self, token_path) -> None:
|
||||
api = SoundCloudAPI("", "", token_path)
|
||||
|
||||
with pytest.raises(
|
||||
SoundCloudAPIError, match="client_id and client_secret"
|
||||
):
|
||||
api.search_tracks("Title", 5)
|
||||
|
||||
|
||||
class TestAPIRequests:
|
||||
def test_search_follows_linked_pagination(self, api, requests_mock) -> None:
|
||||
next_url = f"{API_URL}/tracks?cursor=next"
|
||||
requests_mock.get(
|
||||
f"{API_URL}/tracks",
|
||||
json={
|
||||
"collection": [_track("soundcloud:tracks:1")],
|
||||
"next_href": next_url,
|
||||
},
|
||||
)
|
||||
requests_mock.get(
|
||||
next_url, json={"collection": [_track("soundcloud:tracks:2")]}
|
||||
)
|
||||
|
||||
tracks = api.search_tracks("Artist Title", 2)
|
||||
|
||||
assert [track["urn"] for track in tracks] == [
|
||||
"soundcloud:tracks:1",
|
||||
"soundcloud:tracks:2",
|
||||
]
|
||||
first_request = requests_mock.request_history[0]
|
||||
assert first_request.qs == {
|
||||
"linked_partitioning": ["true"],
|
||||
"limit": ["2"],
|
||||
"q": ["artist title"],
|
||||
}
|
||||
|
||||
def test_rate_limit_error_includes_retry_delay(
|
||||
self, api, requests_mock
|
||||
) -> None:
|
||||
requests_mock.get(
|
||||
f"{API_URL}/tracks", status_code=429, headers={"Retry-After": "12"}
|
||||
)
|
||||
|
||||
with pytest.raises(SoundCloudAPIError, match=r"rate limit.*12"):
|
||||
api.search_tracks("Title", 5)
|
||||
|
||||
def test_playlist_fetches_all_track_pages(self, api, requests_mock) -> None:
|
||||
urn = "soundcloud:playlists:2"
|
||||
tracks_url = f"{API_URL}/playlists/{urn}/tracks"
|
||||
next_url = f"{tracks_url}?cursor=next"
|
||||
requests_mock.get(
|
||||
f"{API_URL}/playlists/{urn}",
|
||||
json=_playlist(tracks=[], track_count=2),
|
||||
)
|
||||
requests_mock.get(
|
||||
tracks_url,
|
||||
json={
|
||||
"collection": [_track("soundcloud:tracks:1")],
|
||||
"next_href": next_url,
|
||||
},
|
||||
)
|
||||
requests_mock.get(
|
||||
next_url, json={"collection": [_track("soundcloud:tracks:2")]}
|
||||
)
|
||||
|
||||
playlist = api.get_playlist(urn)
|
||||
|
||||
assert playlist is not None
|
||||
assert [track["urn"] for track in playlist["tracks"]] == [
|
||||
"soundcloud:tracks:1",
|
||||
"soundcloud:tracks:2",
|
||||
]
|
||||
tracks_request = next(
|
||||
request
|
||||
for request in requests_mock.request_history
|
||||
if request.path.endswith("/tracks")
|
||||
)
|
||||
assert tracks_request.qs == {"linked_partitioning": ["true"]}
|
||||
|
||||
def test_not_found_returns_none(self, api, requests_mock) -> None:
|
||||
requests_mock.get(
|
||||
f"{API_URL}/tracks/soundcloud:tracks:404", status_code=404
|
||||
)
|
||||
|
||||
assert api.get_track("soundcloud:tracks:404") is None
|
||||
|
||||
def test_malformed_resource_fails_clearly(self, api, requests_mock) -> None:
|
||||
requests_mock.get(
|
||||
f"{API_URL}/tracks/soundcloud:tracks:1",
|
||||
json={"kind": "track", "title": "Missing URN"},
|
||||
)
|
||||
|
||||
with pytest.raises(SoundCloudAPIError, match=r"missing.*urn"):
|
||||
api.get_track("soundcloud:tracks:1")
|
||||
|
||||
|
||||
class TestTrackLookup:
|
||||
def test_maps_stable_track_metadata(self, plugin, requests_mock) -> None:
|
||||
requests_mock.get(
|
||||
f"{API_URL}/tracks/soundcloud:tracks:1",
|
||||
json=_track(
|
||||
artist="Credited Artist",
|
||||
bpm=TRACK_BPM,
|
||||
genre="House",
|
||||
isrc="GB-SC0-24-00001",
|
||||
key_signature="C#m",
|
||||
label_name="Label",
|
||||
release_day=3,
|
||||
release_month=2,
|
||||
release_year=2024,
|
||||
),
|
||||
)
|
||||
|
||||
info = plugin.track_for_id("soundcloud:tracks:1")
|
||||
|
||||
assert info is not None
|
||||
assert info.title == "Track"
|
||||
assert info.artist == "Credited Artist"
|
||||
assert info.artist_id is None
|
||||
assert info.length == TRACK_DURATION_SECONDS
|
||||
assert info.bpm == str(TRACK_BPM)
|
||||
assert info.initial_key == "C#m"
|
||||
assert info.genres == ["House"]
|
||||
assert info.isrc == "GB-SC0-24-00001"
|
||||
assert info.label == "Label"
|
||||
assert (info.year, info.month, info.day) == (2024, 2, 3)
|
||||
assert info.soundcloud_track_urn == "soundcloud:tracks:1"
|
||||
assert info.soundcloud_artist_urn == "soundcloud:users:10"
|
||||
|
||||
def test_falls_back_to_uploader_for_missing_credit(
|
||||
self, plugin, requests_mock
|
||||
) -> None:
|
||||
requests_mock.get(
|
||||
f"{API_URL}/tracks/soundcloud:tracks:1",
|
||||
json=_track(artist=None, user=_user("Fallback Artist")),
|
||||
)
|
||||
|
||||
info = plugin.track_for_id("soundcloud:tracks:1")
|
||||
|
||||
assert info is not None
|
||||
assert info.artist == "Fallback Artist"
|
||||
assert info.artist_id == "soundcloud:users:10"
|
||||
|
||||
def test_url_resolves_only_track_resources(
|
||||
self, plugin, requests_mock
|
||||
) -> None:
|
||||
url = "https://soundcloud.com/u/track"
|
||||
requests_mock.get(
|
||||
f"{API_URL}/resolve", json=_playlist(playlist_type="playlist")
|
||||
)
|
||||
|
||||
assert plugin.track_for_id(url) is None
|
||||
|
||||
|
||||
class TestAlbumLookup:
|
||||
def test_search_excludes_ordinary_playlists(
|
||||
self, plugin, requests_mock
|
||||
) -> None:
|
||||
album = _playlist("soundcloud:playlists:2")
|
||||
ordinary = _playlist("soundcloud:playlists:3", playlist_type="playlist")
|
||||
next_url = f"{API_URL}/playlists?cursor=next"
|
||||
plugin.config["search_limit"].set(1)
|
||||
requests_mock.get(
|
||||
f"{API_URL}/playlists",
|
||||
json={"collection": [ordinary], "next_href": next_url},
|
||||
)
|
||||
requests_mock.get(next_url, json={"collection": [album]})
|
||||
requests_mock.get(
|
||||
f"{API_URL}/playlists/soundcloud:playlists:2", json=album
|
||||
)
|
||||
|
||||
candidates = list(
|
||||
plugin.candidates([Item()], "Artist", "Release", False)
|
||||
)
|
||||
|
||||
assert [candidate.album_id for candidate in candidates] == [
|
||||
"soundcloud:playlists:2"
|
||||
]
|
||||
|
||||
def test_direct_url_accepts_ordinary_playlist(
|
||||
self, plugin, requests_mock
|
||||
) -> None:
|
||||
url = "https://soundcloud.com/u/sets/mix"
|
||||
playlist = _playlist(playlist_type="playlist")
|
||||
requests_mock.get(
|
||||
f"{API_URL}/resolve",
|
||||
json={"kind": "playlist", "urn": playlist["urn"]},
|
||||
)
|
||||
requests_mock.get(
|
||||
f"{API_URL}/playlists/{playlist['urn']}", json=playlist
|
||||
)
|
||||
|
||||
info = plugin.album_for_id(url)
|
||||
|
||||
assert info is not None
|
||||
assert info.album == "Release"
|
||||
assert info.albumtype == "playlist"
|
||||
|
||||
def test_mixed_track_credits_create_compilation(
|
||||
self, plugin, requests_mock
|
||||
) -> None:
|
||||
playlist = _playlist(
|
||||
tracks=[
|
||||
_track(
|
||||
"soundcloud:tracks:1", title="First", artist="Artist One"
|
||||
),
|
||||
_track(
|
||||
"soundcloud:tracks:2", title="Second", artist="Artist Two"
|
||||
),
|
||||
],
|
||||
artwork_url="https://i1.sndcdn.com/artwork.jpg",
|
||||
ean="1234567890123",
|
||||
genre="Electronic",
|
||||
label_name="Label",
|
||||
release_day=6,
|
||||
release_month=5,
|
||||
release_year=2024,
|
||||
)
|
||||
requests_mock.get(
|
||||
f"{API_URL}/playlists/{playlist['urn']}", json=playlist
|
||||
)
|
||||
|
||||
info = plugin.album_for_id(str(playlist["urn"]))
|
||||
|
||||
assert info is not None
|
||||
assert info.artist == config["va_name"].as_str()
|
||||
assert info.va is True
|
||||
assert info.barcode == "1234567890123"
|
||||
assert info.cover_art_url == "https://i1.sndcdn.com/artwork.jpg"
|
||||
assert info.genres == ["Electronic"]
|
||||
assert info.label == "Label"
|
||||
assert (info.year, info.month, info.day) == (2024, 5, 6)
|
||||
assert info.soundcloud_playlist_urn == "soundcloud:playlists:2"
|
||||
assert info.soundcloud_artist_urn == "soundcloud:users:10"
|
||||
assert info.albumstatus is None
|
||||
assert [track.index for track in info.tracks] == [1, 2]
|
||||
assert [track.medium_total for track in info.tracks] == [2, 2]
|
||||
|
||||
def test_shared_track_credit_becomes_album_artist(
|
||||
self, plugin, requests_mock
|
||||
) -> None:
|
||||
playlist = _playlist(
|
||||
tracks=[
|
||||
_track("soundcloud:tracks:1", artist="Shared Artist"),
|
||||
_track("soundcloud:tracks:2", artist="Shared Artist"),
|
||||
]
|
||||
)
|
||||
requests_mock.get(
|
||||
f"{API_URL}/playlists/{playlist['urn']}", json=playlist
|
||||
)
|
||||
|
||||
info = plugin.album_for_id(str(playlist["urn"]))
|
||||
|
||||
assert info is not None
|
||||
assert info.artist == "Shared Artist"
|
||||
assert info.va is False
|
||||
Reference in New Issue
Block a user