from __future__ import annotations import re from http.cookiejar import CookieJar from typing import Any, Optional from urllib.parse import urlparse import click from unshackle.core.config import config from unshackle.core.credential import Credential from unshackle.core.music import MusicTrackOption from unshackle.core.service import Service from unshackle.core.titles import Music, Song, Titles_T from unshackle.core.tracks import Audio, Chapters, Tracks from unshackle.core.tracks.track import Track _INVISIBLE = re.compile(r"[​-‏‪-‮⁠]") class SNDC(Service): """ Service code for SoundCloud (https://soundcloud.com). www.nostalgic.cc Authorization: Cookies, Credentials Security: None """ ALIASES = ("SNDC", "soundcloud", "sc") GROUP_AUDIO_DOWNLOADS = True TITLE_RE = r"^(?:https?://)?(?:www\.|m\.)?soundcloud\.com/[^\s?#]+" _PRESET_BITRATE = { "aac_hq": 256000, "aac_256k": 256000, "aac_160k": 160000, "aac_96k": 96000, "aac_1_0": 128000, "mp3_1_0": 128000, "mp3_0_0": 128000, "mp3_0_1": 64000, "opus_0_0": 72000, } @staticmethod @click.command(name="SNDC", short_help="https://soundcloud.com", help=__doc__) @click.argument("title", type=str) @click.option("-q", "--quality", "quality", type=click.Choice(["original", "aac", "mp3"], case_sensitive=False), default=None, help="original = Uploader's file if downloadable else AAC; " "aac = Best AAC stream (256k on Go+) or mp3 = 128k MP3. Default: original.") @click.pass_context def cli(ctx, **kwargs): return SNDC(ctx, **kwargs) def __init__(self, ctx, title: str, quality: Optional[str]): super().__init__(ctx) self.title = title self.quality_pref = (quality or self.config.get("default_quality", "original")).lower() self.client_id: Optional[str] = None self.oauth_token: Optional[str] = None def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None: super().authenticate(cookies, credential) self.session.headers.update({ "User-Agent": self.config["user_agent"], "Origin": self.config["web_url"], "Referer": self.config["web_url"] + "/", }) self.client_id = self._extract_client_id() or (str(self.config.get("client_id") or "").strip() or None) if not self.client_id: self.log.error(" - Could not get a SoundCloud client_id."); raise SystemExit(1) token = None if credential: user = (credential.username or "").strip() pw = (credential.password or "").strip() if user.lower() in ("token", "oauth", "oauth_token") and pw: token = pw elif pw and not user: token = pw if not token and cookies: for cookie in cookies: if cookie.name == "oauth_token": token = cookie.value break if token: self.oauth_token = token if token.lower().startswith("oauth ") else token self.session.headers["Authorization"] = f"OAuth {token}" self.log.info(" + Authenticated with SoundCloud (OAuth)") else: self.log.warning(" - No oauth_token found; Fallback to mp3 128kb/s.") def _extract_client_id(self) -> Optional[str]: try: page = self.session.get(self.config["web_url"] + "/").text scripts = re.findall(r']+src="(https://[^"]+\.js)"', page) for src in reversed(scripts): try: js = self.session.get(src).text except Exception: continue m = re.search(r'client_id\s*[:=]\s*"([a-zA-Z0-9]{20,})"', js) if m: self.log.debug(f" + Extracted client_id from {src}") return m.group(1) except Exception as e: self.log.debug(f"client_id extraction failed: {e}") return None def _api(self, endpoint: str, params: Optional[dict] = None, allow_error: bool = False) -> Optional[Any]: url = endpoint if endpoint.startswith("http") else f"{self.config['base_url']}/{endpoint.lstrip('/')}" p = {"client_id": self.client_id} if params: p.update(params) resp = self.session.get(url, params=p) if resp.status_code != 200: if allow_error: return None self.log.error(f" - SoundCloud API error on {endpoint}: {resp.status_code} {resp.text[:200]}") raise SystemExit(1) return resp.json() def get_titles(self) -> Titles_T: obj = self._api("resolve", params={"url": self.title}) kind = obj.get("kind") if kind == "track": song = self._build_song(obj, total_tracks=1) return Music([song], kind="single", title=song.album, artist=song.artist, year=song.year, total_tracks=1, artwork_url=song.artwork_url) if kind == "playlist": return self._build_playlist(obj) self.log.error(f" - Unsupported SoundCloud URL: {kind!r} (Need a track or playlist/album URL).") raise SystemExit(1) def _build_playlist(self, playlist: dict) -> Music: items = playlist.get("tracks", []) full: dict[int, dict] = {} need_ids = [t["id"] for t in items if isinstance(t, dict) and "media" not in t and t.get("id")] for t in items: if isinstance(t, dict) and t.get("media"): full[t["id"]] = t for i in range(0, len(need_ids), 50): batch = need_ids[i:i + 50] fetched = self._api("tracks", params={"ids": ",".join(map(str, batch))}, allow_error=True) or [] for ft in fetched: if ft.get("id"): full[ft["id"]] = ft album_title = self._clean(playlist.get("title")) or "Unknown Album" is_album = bool(playlist.get("is_album")) total = playlist.get("track_count") or len([t for t in items if full.get(t.get("id") if isinstance(t, dict) else t)]) songs = [] position = 0 for entry in items: tid = entry.get("id") if isinstance(entry, dict) else entry track = full.get(tid) if not track: continue position += 1 songs.append(self._build_song(track, album_ctx=album_title, position=position, total_tracks=total)) if not songs: self.log.error(" - No playable tracks found in playlist."); raise SystemExit(1) return Music( songs, kind="album" if is_album else "playlist", title=album_title, artist=self._clean((playlist.get("user") or {}).get("username")), year=self._year(playlist), total_tracks=len(songs), artwork_url=self._cover(playlist.get("artwork_url") or songs[0].artwork_url), ) def _build_song(self, track: dict, album_ctx: Optional[str] = None, position: Optional[int] = None, total_tracks: Optional[int] = None) -> Song: pm = track.get("publisher_metadata") or {} user = track.get("user") or {} title = self._clean(track.get("title")) or "Unknown" artist = self._clean(pm.get("artist")) or self._clean(user.get("username")) or "Unknown Artist" album = self._clean(album_ctx) or self._clean(pm.get("album_title")) or title year = self._year(track) or 1 artwork = self._cover(track.get("artwork_url") or user.get("avatar_url")) genre = self._clean(track.get("genre")) or None isrc = pm.get("isrc") or None label = self._clean(track.get("label_name")) or None release_date = self._release_date(track) data = { "service": self.ALIASES[0], "source": self.ALIASES[0], "track_id": str(track.get("id")), "track_url": track.get("permalink_url"), "title": title, "artist": artist, "performer": artist, "album": album, "album_artist": artist, "track_number": position or 1, "total_tracks": total_tracks, "disc_number": 1, "genre": genre, "isrc": isrc, "label": label, "release_date": release_date, "year": year, "copyright": self._clean(pm.get("c_line")) or None, "artwork_url": artwork, "duration": round((track.get("duration") or 0) / 1000) or None, "channels": 2, "downloadable": bool(track.get("downloadable")), "has_downloads_left": bool(track.get("has_downloads_left")), "original_format": track.get("original_format"), "track_authorization": track.get("track_authorization"), "transcodings": (track.get("media") or {}).get("transcodings", []), } if config.tag: data["comment"] = config.tag data["quality"] = self._quality_label(data) return Song( id_=str(track.get("id")), service=self.__class__, name=title, artist=artist, album=album, track=int(position or 1), disc=1, year=int(year), album_artist=artist, release_type="single" if not album_ctx else "album", total_tracks=total_tracks if total_tracks else None, genre=genre, isrc=isrc if isinstance(isrc, str) else None, label=label, artwork_url=artwork, data=data, ) def get_music_track_options(self, song: Song) -> list[MusicTrackOption]: data = song.data if isinstance(song.data, dict) else {} use_original = self._will_use_original(data) if use_original: fmt = (data.get("original_format") or "flac").lower() codec = {"flac": "FLAC", "wav": "WAV", "aiff": "AIFF", "aif": "AIFF", "mp3": "MP3", "m4a": "AAC"}.get(fmt, fmt.upper()) return [MusicTrackOption(codec=codec, channels=2.0, lossless=codec in ("FLAC", "WAV", "AIFF"), duration=data.get("duration"), quality_label=self._quality_label(data))] transcoding = self._best_stream_transcoding(data) codec = "MP3" if transcoding and self._tc_codec(transcoding) == "mp3" else "AAC" bitrate = self._tc_bitrate(transcoding) if transcoding else None return [MusicTrackOption(codec=codec, bitrate=bitrate, channels=2.0, lossless=False, duration=data.get("duration"), quality_label=self._quality_label(data))] def get_tracks(self, title: Song) -> Tracks: data = title.data if isinstance(title.data, dict) else {} track_id = str(title.id) if self._will_use_original(data): url, ext = self._original_download(track_id, data) if url: codec = Audio.Codec.FLAC if ext == "flac" else None audio = Audio( url, language=title.language or "en", codec=codec, channels=2, descriptor=Track.Descriptor.URL, id_=track_id, data={"ext": ext}, ) return Tracks([audio]) self.log.warning(" - Original download unavailable; falling back to stream.") transcoding = self._best_stream_transcoding(data) if not transcoding: self.log.error(f" - No usable stream for track {track_id}."); raise SystemExit(1) stream_url = self._resolve_stream(transcoding, data.get("track_authorization")) is_mp3 = self._tc_codec(transcoding) == "mp3" is_progressive = self._tc_progressive(transcoding) self.log.debug( f" + Stream: preset={transcoding.get('preset')} quality={transcoding.get('quality')} " f"protocol={'progressive' if is_progressive else 'hls'}" ) audio = Audio( stream_url, language=title.language or "en", codec=None if is_mp3 else Audio.Codec.AAC, bitrate=self._tc_bitrate(transcoding), channels=2, descriptor=Track.Descriptor.URL if is_progressive else Track.Descriptor.HLS, id_=track_id, data={"ext": "mp3" if is_mp3 else "m4a"}, ) return Tracks([audio]) def get_chapters(self, title: Song) -> Chapters: return Chapters() def _will_use_original(self, data: dict) -> bool: return (self.quality_pref == "original" and bool(data.get("downloadable")) and bool(data.get("has_downloads_left"))) def _stream_transcodings(self, data: dict) -> list[dict]: out = [] for t in (data.get("transcodings") or []): if not isinstance(t, dict) or not t.get("url") or not t.get("preset"): continue if str(t.get("preset")).lower().startswith("abr_"): continue out.append(t) return out @staticmethod def _tc_codec(t: dict) -> str: preset = str(t.get("preset") or "").lower() mime = str((t.get("format") or {}).get("mime_type") or "").lower() if "aac" in preset or "mp4a" in mime: return "aac" if "opus" in preset or "opus" in mime: return "opus" if "mp3" in preset or "mpeg" in mime: return "mp3" return "other" @staticmethod def _tc_progressive(t: dict) -> bool: return str((t.get("format") or {}).get("protocol") or "").lower() == "progressive" def _tc_bitrate(self, t: dict) -> int: preset = str(t.get("preset") or "").lower() m = re.search(r"(\d+)k", preset) if m: return int(m.group(1)) * 1000 if preset in self._PRESET_BITRATE: return self._PRESET_BITRATE[preset] quality = str(t.get("quality") or "").lower() codec = self._tc_codec(t) if codec == "aac": return 256000 if quality == "hq" else 128000 if codec == "mp3": return 128000 if codec == "opus": return 72000 return 128000 if quality == "hq" else 96000 def _best_stream_transcoding(self, data: dict) -> Optional[dict]: candidates = self._stream_transcodings(data) if not candidates: return None prefer_mp3 = self.quality_pref == "mp3" def codec_pref(t: dict) -> int: codec = self._tc_codec(t) if prefer_mp3: return {"mp3": 3, "aac": 2, "opus": 1}.get(codec, 0) return {"aac": 3, "opus": 2, "mp3": 1}.get(codec, 0) def rank(t: dict) -> tuple: quality = str(t.get("quality") or "").lower() return ( 0 if t.get("snipped") else 1, codec_pref(t), 1 if quality == "hq" else 0, 1 if self._tc_progressive(t) else 0, 0 if t.get("is_legacy_transcoding") else 1, self._tc_bitrate(t), ) return max(candidates, key=rank) def _resolve_stream(self, transcoding: dict, track_authorization: Optional[str]) -> str: params = {} if track_authorization: params["track_authorization"] = track_authorization res = self._api(transcoding["url"], params=params) url = res.get("url") if not url: self.log.error(f" - No stream URL returned: {res}"); raise SystemExit(1) return url def _original_download(self, track_id: str, data: dict) -> tuple[Optional[str], str]: res = self._api(f"tracks/{track_id}/download", params={"app_version": self.config.get("app_version"), "app_locale": "en"}, allow_error=True) redirect = (res or {}).get("redirectUri") if not redirect: return None, "flac" ext = (data.get("original_format") or "").lower().replace("aif", "aiff") if not ext: try: head = self.session.head(redirect, allow_redirects=True, timeout=15) cd = head.headers.get("Content-Disposition", "") m = re.search(r'filename="?[^"]+\.([A-Za-z0-9]+)"?', cd) if m: ext = m.group(1).lower() except Exception: pass return redirect, (ext or "flac") def on_track_downloaded(self, track: Any) -> None: try: path = getattr(track, "path", None) tdata = getattr(track, "data", None) if not path or not path.exists() or not isinstance(tdata, dict): return ext = tdata.get("ext") if not ext or path.suffix.lower() == f".{ext}": return new_path = path.with_suffix(f".{ext}") if new_path.exists(): new_path.unlink() path.rename(new_path) track.path = new_path except Exception as e: self.log.debug(f"Extension rename skipped: {e}") @staticmethod def _clean(value: Any) -> str: if not value: return "" return _INVISIBLE.sub("", str(value)).strip() @staticmethod def _year(obj: dict) -> int: for key in ("release_date", "display_date", "created_at"): value = obj.get(key) if value: m = re.match(r"(\d{4})", str(value)) if m and int(m.group(1)) > 0: return int(m.group(1)) return 0 @staticmethod def _release_date(obj: dict) -> Optional[str]: for key in ("release_date", "display_date", "created_at"): value = obj.get(key) if value and re.match(r"\d{4}-\d{2}-\d{2}", str(value)): return str(value)[:10] return None @staticmethod def _cover(url: Optional[str]) -> Optional[str]: if not url: return None return re.sub(r"-(large|t\d+x\d+|small|badge|tiny|mini|original)\.(jpg|png)", r"-original.\2", url) def _quality_label(self, data: dict) -> str: if self._will_use_original(data): return f"Original {(data.get('original_format') or 'FLAC').upper()}" transcoding = self._best_stream_transcoding(data) if not transcoding: return "" codec = self._tc_codec(transcoding) kbps = self._tc_bitrate(transcoding) // 1000 if codec == "mp3": return f"MP3 {kbps} kb/s" if codec == "opus": return f"Opus {kbps} kb/s" return f"AAC {kbps} kb/s"