"""SoundCloud API client. Thin async wrapper over the SoundCloud public REST API. Mirrors Sony.Filtr.SoundCloud.SoundCloudApi from the .NET codebase. Authentication: per-service-account OAuth tokens stored in tblServiceAccount. """ import logging from typing import Any, Optional import httpx logger = logging.getLogger("playlist_sync.soundcloud_client") _SC_BASE = "https://api.soundcloud.com" async def _get(path: str, access_token: str, params: Optional[dict] = None) -> Any: headers = {"Authorization": f"OAuth {access_token}"} params = params or {} async with httpx.AsyncClient() as client: resp = await client.get( f"{_SC_BASE}{path}", headers=headers, params=params, timeout=30 ) resp.raise_for_status() return resp.json() async def _put(path: str, access_token: str, json: dict) -> Any: headers = { "Authorization": f"OAuth {access_token}", "Content-Type": "application/json", } async with httpx.AsyncClient() as client: resp = await client.put( f"{_SC_BASE}{path}", headers=headers, json=json, timeout=30 ) resp.raise_for_status() return resp.json() async def get_playlist_by_id(playlist_id: int, access_token: str) -> Optional[dict]: """Fetch a SoundCloud playlist by numeric ID. Mirrors: SoundCloudApi.GetPlaylistAsync(int) """ try: return await _get(f"/playlists/{playlist_id}", access_token) # type: ignore[no-any-return] except httpx.HTTPStatusError as exc: logger.warning("Failed to fetch SoundCloud playlist %s: %s", playlist_id, exc) return None async def get_playlist_by_url(url: str, access_token: str) -> Optional[dict]: """Resolve a SoundCloud playlist URL. Mirrors: SoundCloudApi.GetPlaylistAsync(string url) """ try: return await _get("/resolve", access_token, {"url": url}) # type: ignore[no-any-return] except httpx.HTTPStatusError as exc: logger.warning("Failed to resolve SoundCloud URL %s: %s", url, exc) return None async def search_tracks( name: str, isrc: Optional[str], access_token: str ) -> list[dict]: """Search SoundCloud for tracks matching a name/ISRC query. Mirrors: SoundCloudApi.GetTrackAsync(name, isrc) """ query = isrc if isrc else name try: result = await _get("/tracks", access_token, {"q": query, "limit": 20}) # Result may be a list or a dict with 'collection' if isinstance(result, list): return result # type: ignore[no-any-return] return result.get("collection", []) # type: ignore[no-any-return] except httpx.HTTPStatusError as exc: logger.warning("SoundCloud track search failed for '%s': %s", query, exc) return [] async def add_tracks_to_playlist( playlist_id: int, tracks: list[dict], access_token: str, ) -> None: """Replace the track list on a SoundCloud playlist. The .NET SoundCloudApi.AddTracksToPlayListAsync replaces the full track list. Mirrors: SoundCloudApi.AddTracksToPlayListAsync() """ try: await _put( f"/playlists/{playlist_id}", access_token, {"playlist": {"tracks": tracks}}, ) except httpx.HTTPStatusError as exc: logger.error( "Failed to add tracks to SoundCloud playlist %s: %s", playlist_id, exc ) raise async def update_playlist_info( playlist_id: str, title: Optional[str], description: Optional[str], access_token: str, ) -> None: """Update SoundCloud playlist title and/or description. Mirrors: SoundCloudApi.UpdateTitleAndDescription() """ payload: dict = {"playlist": {}} if title is not None: payload["playlist"]["title"] = title if description is not None: payload["playlist"]["description"] = description if not payload["playlist"]: return try: await _put(f"/playlists/{playlist_id}", access_token, payload) except httpx.HTTPStatusError as exc: logger.error( "Failed to update SoundCloud playlist info %s: %s", playlist_id, exc ) raise