diff --git a/requirements_dev.txt b/requirements_dev.txt index 9789196..d99062b 100644 --- a/requirements_dev.txt +++ b/requirements_dev.txt @@ -1,7 +1,6 @@ mutagen -bs4 aiohttp pytest pytest-asyncio -pylint \ No newline at end of file +pylint diff --git a/sclib/asyncio.py b/sclib/asyncio.py index 1bd4f96..8ae4471 100644 --- a/sclib/asyncio.py +++ b/sclib/asyncio.py @@ -1,45 +1,44 @@ """ Asyncio """ -import sys -import random -import json -import itertools import asyncio +import itertools +import json +import sys + import aiohttp import mutagen from . import sync, util + async def get_resource(url) -> bytes: - """ Get a resource based on url """ + """Get a resource based on url""" async with aiohttp.ClientSession() as session: async with session as conn: - async with conn.request('GET', url) as request: + async with conn.request("GET", url) as request: return await request.content.read() - async def fetch_soundcloud_client_id(): - """ Get soundlcoud client id """ - url = random.choice(util.SCRAPE_URLS) - page_text = await get_resource(url) - script_urls = util.find_script_urls(page_text.decode()) - results = await asyncio.gather(*[get_resource(u) for u in script_urls]) - script_text = "".join([r.decode() for r in results]) - return util.find_client_id(script_text) - -__all__ = [ - "Track", - "Playlist", - "SoundcloudAPI" -] + """Get soundlcoud client id""" + data = await get_resource(util.SCRAPE_URL) + page_text = data.decode() + js_link = util.find_script_urls(page_text) + data = await get_resource(js_link) + js_text = data.decode() + return util.find_client_id(js_text) + + +__all__ = ["Track", "Playlist", "SoundcloudAPI"] + def eprint(*values, **kwargs): - """ Stderr print """ + """Stderr print""" print(*values, file=sys.stderr, **kwargs) + async def get_obj_from(url): - """ Get a json object from a url """ + """Get a json object from a url""" try: return json.loads(await get_resource(url)) except Exception as exc: # pylint: disable=broad-except @@ -47,53 +46,50 @@ async def get_obj_from(url): return False - - -async def embed_artwork(audio:mutagen.File, artwork_url): - """ Embed an artwork image into a mp3 """ +async def embed_artwork(audio: mutagen.File, artwork_url): + """Embed an artwork image into a mp3""" if artwork_url: audio.tags.add( mutagen.id3.APIC( encoding=3, - mime='image/jpeg', + mime="image/jpeg", type=3, - desc='Cover', - data=await get_resource(artwork_url) + desc="Cover", + data=await get_resource(artwork_url), ) ) return audio - class SoundcloudAPI(sync.SoundcloudAPI): - """ Asynchronous Soundcloud API Client """ + """Asynchronous Soundcloud API Client""" async def get_credentials(self): # pylint: disable=invalid-overridden-method) - """ Find api credentials """ + """Find api credentials""" self.client_id = await fetch_soundcloud_client_id() if self.client_id is None: raise RuntimeError( - 'ScLib could not automatically find a public client id. ' - 'This means Soundcloud has changed where the public client id is located. ' - 'Please report this to the package author.' + "ScLib could not automatically find a public client id. " + "This means Soundcloud has changed where the public client id is located. " + "Please report this to the package author." ) async def resolve(self, url): # pylint: disable=invalid-overridden-method - """ Resolve an api url to a soundcloud object """ + """Resolve an api url to a soundcloud object""" if not self.client_id: await self.get_credentials() - full_url = f"https://api-v2.soundcloud.com/resolve?url={url}&client_id={self.client_id}&app_version=1499347238" + full_url = util.RESOLVE_URL.format(url=url, client_id=self.client_id) obj = await get_obj_from(full_url) - if obj['kind'] == 'track': + if obj["kind"] == "track": return Track(obj=obj, client=self) - if obj['kind'] in ('playlist', 'system-playlist'): + if obj["kind"] in ("playlist", "system-playlist"): playlist = Playlist(obj=obj, client=self) await playlist.clean_attributes() return playlist async def get_tracks(self, *track_ids): # pylint: disable=invalid-overridden-method - """ Get a list of tracks from a list of ids """ + """Get a list of tracks from a list of ids""" if not self.client_id: await self.get_credentials() @@ -105,15 +101,15 @@ async def get_tracks(self, *track_ids): # pylint: disable=invalid-overridden-me response = await asyncio.gather(*tasks) tracks = list(itertools.chain.from_iterable(response)) - tracks = sorted(tracks, key=lambda x: track_ids.index(x['id'])) + tracks = sorted(tracks, key=lambda x: track_ids.index(x["id"])) return tracks class Track(sync.Track): - """ Asynchronous track object """ + """Asynchronous track object""" async def write_mp3_to(self, file): # pylint: disable=invalid-overridden-method) - """ Write the mp3 representation of this track to a file object """ + """Write the mp3 representation of this track to a file object""" try: file.seek(0) stream_url = await self.get_stream_url() @@ -124,30 +120,30 @@ async def write_mp3_to(self, file): # pylint: disable=invalid-overridden-method album_artwork = None if self.artwork_url: album_artwork = await get_resource( - util.get_large_artwork_url( - self.artwork_url - ) + util.get_large_artwork_url(self.artwork_url) ) self.write_track_id3(file, album_artwork) except (TypeError, ValueError) as exc: - util.eprint('File object passed to "write_mp3_to" must be opened in read/write binary ("wb+") mode') + util.eprint( + 'File object passed to "write_mp3_to" must be opened in read/write binary ("wb+") mode' + ) util.eprint(exc) raise exc async def get_stream_url(self): # pylint: disable=invalid-overridden-method - """ get the stream url for this track """ + """get the stream url for this track""" prog_url = self.get_prog_url() stream_response = await get_obj_from(prog_url) try: - return stream_response['url'] + return stream_response["url"] except Exception as exc: # pylint: disable=broad-except) eprint(exc) return None def to_dict(self) -> dict: - """ Conver this track object to a dict """ - ignore_attributes = ['client', 'ready'] + """Conver this track object to a dict""" + ignore_attributes = ["client", "ready"] track_dict = {} for attr in set(self.__slots__): if attr not in ignore_attributes: @@ -156,28 +152,31 @@ def to_dict(self) -> dict: return track_dict - class Playlist(sync.Playlist): - """ Playlist """ + """Playlist""" RESOLVE_THRESHOLD = 100 - async def clean_attributes(self): # pylint: disable=invalid-overridden-method + async def clean_attributes(self): # pylint: disable=invalid-overridden-method if self.ready: return self.ready = True track_objects = [] # type: [Track] # all completed track objects - incomplete_track_ids = [] # tracks that do not have metadata + incomplete_track_ids = [] # tracks that do not have metadata - while self.tracks and 'title' in self.tracks[0]: # remove completed track objects + while ( + self.tracks and "title" in self.tracks[0] + ): # remove completed track objects track_objects.append(Track(obj=self.tracks.pop(0), client=self.client)) - while self.tracks: # while built tracks are less than all tracks - incomplete_track_ids.append(self.tracks.pop(0)['id']) + while self.tracks: # while built tracks are less than all tracks + incomplete_track_ids.append(self.tracks.pop(0)["id"]) if len(incomplete_track_ids) == self.RESOLVE_THRESHOLD or not self.tracks: new_tracks = await self.client.get_tracks(*incomplete_track_ids) - track_objects.extend([Track(obj=t, client=self.client) for t in new_tracks]) + track_objects.extend( + [Track(obj=t, client=self.client) for t in new_tracks] + ) incomplete_track_ids.clear() for track in track_objects: @@ -190,8 +189,8 @@ async def __aiter__(self): yield track def to_dict(self): - """ convert this object to a dict """ - ignore_attributes = ['client', 'ready'] + """convert this object to a dict""" + ignore_attributes = ["client", "ready"] playlist_dict = {} for attr in set(self.__slots__): if attr not in ignore_attributes: diff --git a/sclib/sync.py b/sclib/sync.py index e45e6f3..9b3a093 100644 --- a/sclib/sync.py +++ b/sclib/sync.py @@ -1,33 +1,37 @@ """ Soundcloud api sync objects """ -from urllib.request import urlopen import json -import random -from ssl import SSLContext from concurrent import futures +from ssl import SSLContext +from urllib.request import urlopen + import mutagen + from . import util +SSL_VERIFY = True -SSL_VERIFY=True def get_ssl_setting(): - """ Get ssl context """ + """Get ssl context""" if SSL_VERIFY: return None return SSLContext() + def get_url(url): - """ Get url """ + """Get url""" with urlopen(url, context=get_ssl_setting()) as client: text = client.read() return text + def get_page(url): - """ get text from url """ - return get_url(url).decode('utf-8') + """get text from url""" + return get_url(url).decode("utf-8") + def get_obj_from(url): - """ Get object from url """ + """Get object from url""" try: return json.loads(get_page(url)) except Exception as exc: # pylint: disable=broad-except @@ -36,20 +40,15 @@ def get_obj_from(url): class UnsupportedFormatError(Exception): - """ unsupported format """ - + """unsupported format""" class SoundcloudAPI: - """ Soundcloud api client """ + """Soundcloud api client""" + __slots__ = [ - 'client_id', + "client_id", ] - RESOLVE_URL = "https://api-v2.soundcloud.com/resolve?url={url}&client_id={client_id}" - SEARCH_URL = "https://api-v2.soundcloud.com/search?q={query}&client_id={client_id}&limit={limit}&offset={offset}" - STREAM_URL = "https://api.soundcloud.com/i1/tracks/{track_id}/streams?client_id={client_id}" - TRACKS_URL = "https://api-v2.soundcloud.com/tracks?ids={track_ids}&client_id={client_id}" - PROGRESSIVE_URL = "https://api-v2.soundcloud.com/media/soundcloud:tracks:723290971/53dc4e74-0414-4ab8-8741-a07ac56c787f/stream/progressive?client_id={client_id}" TRACK_API_MAX_REQUEST_SIZE = 50 @@ -59,31 +58,23 @@ def __init__(self, client_id=None): else: self.client_id = None - def get_credentials(self): - """ get creds """ - url = random.choice(util.SCRAPE_URLS) - page_text = get_page(url) - script_urls = util.find_script_urls(page_text) - for script in script_urls: - if not self.client_id: - if type(script) is str and not "": # pylint: disable=simplifiable-condition - js_text = f'{get_page(script)}' - self.client_id = util.find_client_id(js_text) + """get creds""" + page_text = get_page(util.SCRAPE_URL) + js_link = util.find_script_urls(page_text) + js_text = get_page(js_link) + self.client_id = util.find_client_id(js_text) def resolve(self, url): - """ Resolve url """ + """Resolve url""" if not self.client_id: self.get_credentials() - url = SoundcloudAPI.RESOLVE_URL.format( - url=url, - client_id=self.client_id - ) + url = util.RESOLVE_URL.format(url=url, client_id=self.client_id) obj = get_obj_from(url) - if obj['kind'] == 'track': + if obj["kind"] == "track": return Track(obj=obj, client=self) - if obj['kind'] in ('playlist', 'system-playlist'): + if obj["kind"] in ("playlist", "system-playlist"): playlist = Playlist(obj=obj, client=self) playlist.clean_attributes() return playlist @@ -94,15 +85,15 @@ def _format_get_tracks_urls(self, track_ids): for start_offset in range(0, len(track_ids), self.TRACK_API_MAX_REQUEST_SIZE): end_offset = start_offset + self.TRACK_API_MAX_REQUEST_SIZE track_ids_slice = track_ids[start_offset:end_offset] - url = self.TRACKS_URL.format( - track_ids=','.join([str(i) for i in track_ids_slice]), - client_id=self.client_id + url = util.TRACKS_URL.format( + track_ids=",".join([str(i) for i in track_ids_slice]), + client_id=self.client_id, ) urls.append(url) return urls def get_tracks(self, *track_ids): - """ Get a list of track ids """ + """Get a list of track ids""" threads = [] with futures.ThreadPoolExecutor() as executor: for url in self._format_get_tracks_urls(track_ids): @@ -114,12 +105,13 @@ def get_tracks(self, *track_ids): result = thread.result() tracks.extend(result) - tracks = sorted(tracks, key=lambda x: track_ids.index(x['id'])) + tracks = sorted(tracks, key=lambda x: track_ids.index(x["id"])) return tracks class Track: - """ Track object """ + """Track object""" + __slots__ = [ # Track Attributes "artwork_url", @@ -167,21 +159,21 @@ class Track: "monetization_model", "policy", "user", - - #extra attributes + # extra attributes "album", "track_no", - # Internal Attributes "client", - "ready" + "ready", ] - STREAM_URL = "https://api.soundcloud.com/i1/tracks/{track_id}/streams?client_id={client_id}" + def __init__(self, *, obj=None, client=None): if not obj: raise ValueError("[Track]: obj must not be None") if not isinstance(client, SoundcloudAPI): - raise ValueError(f"[Track]: client must be an instance of SoundcloudAPI not {type(client)}") + raise ValueError( + f"[Track]: client must be an instance of SoundcloudAPI not {type(client)}" + ) for key in self.__slots__: self.__setattr__(key, obj[key] if key in obj else None) @@ -191,8 +183,8 @@ def __init__(self, *, obj=None, client=None): self.clean_attributes() def clean_attributes(self): - """ clean attrs """ - username = self.user['username'] + """clean attrs""" + username = self.user["username"] title = self.title if " - " in title: parts = title.split("-") @@ -200,79 +192,88 @@ def clean_attributes(self): self.title = "-".join(parts[1:]).strip() else: self.artist = username -# -# Uses urllib -# + + # + # Uses urllib + # def write_mp3_to(self, file): - """ Write mp3 data to file """ + """Write mp3 data to file""" try: file.seek(0) stream_url = self.get_stream_url() - with urlopen(stream_url,context=get_ssl_setting()) as client: + with urlopen(stream_url, context=get_ssl_setting()) as client: data = client.read() file.write(data) file.seek(0) album_artwork = None if self.artwork_url: - with urlopen(util.get_large_artwork_url(self.artwork_url),context=get_ssl_setting()) as client: + with urlopen( + util.get_large_artwork_url(self.artwork_url), + context=get_ssl_setting(), + ) as client: album_artwork = client.read() self.write_track_id3(file, album_artwork) except (TypeError, ValueError) as exc: - util.eprint('File object passed to "write_mp3_to" must be opened in read/write binary ("wb+") mode') + util.eprint( + 'File object passed to "write_mp3_to" must be opened in read/write binary ("wb+") mode' + ) util.eprint(exc) raise exc def get_prog_url(self): - """ Get url """ - for transcode in self.media['transcodings']: - if transcode['format']['protocol'] == 'progressive': - return transcode['url'] + "?client_id=" + self.client.client_id - raise UnsupportedFormatError("As of soundcloud-lib 0.5.0, tracks that are not marked as 'Downloadable' cannot be downloaded because this library does not yet assemble HLS streams.") -# -# Uses urllib -# + """Get url""" + for transcode in self.media["transcodings"]: + if transcode["format"]["protocol"] == "progressive": + return transcode["url"] + "?client_id=" + self.client.client_id + raise UnsupportedFormatError( + "As of soundcloud-lib 0.5.0, tracks that are not marked as 'Downloadable' cannot be downloaded because this library does not yet assemble HLS streams." + ) + + # + # Uses urllib + # def get_stream_url(self): - """ Get stream url """ + """Get stream url""" prog_url = self.get_prog_url() url_response = get_obj_from(prog_url) - return url_response['url'] + return url_response["url"] - def write_track_id3(self, track_fp, album_artwork:bytes = None): - """ Write track meta """ + def write_track_id3(self, track_fp, album_artwork: bytes = None): + """Write track meta""" try: audio = mutagen.File(track_fp, filename="x.mp3") audio.add_tags() - # SET TITLE + # SET TITLE frame = mutagen.id3.TIT2(encoding=3) frame.append(self.title) audio.tags.add(frame) - # SET ARTIST + # SET ARTIST frame = mutagen.id3.TPE1(encoding=3) frame.append(self.artist) audio.tags.add(frame) - # SET ALBUM + # SET ALBUM if self.album: frame = mutagen.id3.TALB(encoding=3) frame.append(self.album) audio.tags.add(frame) - # SET TRACK NO + # SET TRACK NO if self.track_no: frame = mutagen.id3.TRCK(encoding=3) frame.append(str(self.track_no)) audio.tags.add(frame) - # SET ARTWORK + # SET ARTWORK if album_artwork: audio.tags.add( mutagen.id3.APIC( encoding=3, - mime='image/jpeg', + mime="image/jpeg", type=3, - desc='Cover', - data=album_artwork + desc="Cover", + data=album_artwork, ) ) audio.save(track_fp, v1=2) @@ -280,13 +281,15 @@ def write_track_id3(self, track_fp, album_artwork:bytes = None): track_fp.seek(0) return track_fp except (TypeError, ValueError) as exc: - util.eprint('File object passed to "write_track_metadata" must be opened in read/write binary ("wb+") mode') + util.eprint( + 'File object passed to "write_track_metadata" must be opened in read/write binary ("wb+") mode' + ) raise exc - class Playlist: - """ Playlist """ + """Playlist""" + __slots__ = [ "artwork_url", "created_at", @@ -321,9 +324,8 @@ class Playlist: "user", "tracks", "track_count", - "client", - "ready" + "ready", ] RESOLVE_THRESHOLD = 100 @@ -337,21 +339,25 @@ def __init__(self, *, obj=None, client=None): self.ready = False def clean_attributes(self): - """ Clean attributes """ + """Clean attributes""" if self.ready: return self.ready = True track_objects = [] # type: [Track] # all completed track objects incomplete_track_ids = [] # tracks that do not have metadata - while self.tracks and 'title' in self.tracks[0]: # remove completed track objects + while ( + self.tracks and "title" in self.tracks[0] + ): # remove completed track objects track_objects.append(Track(obj=self.tracks.pop(0), client=self.client)) while self.tracks: # while built tracks are less than all tracks - incomplete_track_ids.append(self.tracks.pop(0)['id']) + incomplete_track_ids.append(self.tracks.pop(0)["id"]) if len(incomplete_track_ids) == self.RESOLVE_THRESHOLD or not self.tracks: new_tracks = self.client.get_tracks(*incomplete_track_ids) - track_objects.extend([Track(obj=t, client=self.client) for t in new_tracks]) + track_objects.extend( + [Track(obj=t, client=self.client) for t in new_tracks] + ) incomplete_track_ids.clear() self.tracks = track_objects diff --git a/sclib/util.py b/sclib/util.py index 2c1de9c..b263ca0 100644 --- a/sclib/util.py +++ b/sclib/util.py @@ -1,38 +1,40 @@ """ Common utils """ -import sys import re -from bs4 import BeautifulSoup +import sys + +SCRAPE_URL = "https://soundcloud.com/discover" +RESOLVE_URL = "https://api-v2.soundcloud.com/resolve?url={url}&client_id={client_id}" +TRACKS_URL = ( + "https://api-v2.soundcloud.com/tracks?ids={track_ids}&client_id={client_id}" +) +# SEARCH_URL = "https://api-v2.soundcloud.com/search?q={query}&client_id={client_id}&limit={limit}&offset={offset}" +# STREAM_URL = "https://api.soundcloud.com/i1/tracks/{track_id}/streams?client_id={client_id}" +# PROGRESSIVE_URL = "https://api-v2.soundcloud.com/media/soundcloud:tracks:{track_id}/53dc4e74-0414-4ab8-8741-a07ac56c787f/stream/progressive?client_id={client_id}" def eprint(*values, **kwargs): - """ Print to stderr """ + """Print to stderr""" print(*values, file=sys.stderr, **kwargs) -SCRAPE_URLS = [ - 'https://soundcloud.com/mt-marcy/cold-nights' -] - def find_script_urls(html_text): - """ Get script url that has client_id in it """ - dom = BeautifulSoup(html_text, 'html.parser') - scripts = dom.findAll('script', attrs={'src': True}) - scripts_list = [] - for script in scripts: - src = script['src'] - if 'cookielaw.org' not in src: # filter out cookielaw.org - scripts_list.append(src) - return scripts_list + """Get script url that has client_id in it""" + return re.findall( + r'', + html_text, + flags=re.IGNORECASE, + )[0] def find_client_id(script_text): - """ Extract client_id from script """ - client_id = re.findall(r'client_id=([a-zA-Z0-9]+)', script_text) - if len(client_id) > 0: - return client_id[0] + """Extract client_id from script""" + key_data = re.findall(r'client_id:"(.*?)"', script_text, flags=re.IGNORECASE) + if key_data is not None: + return key_data[0] return False + def get_large_artwork_url(artwork_url): - """ Get 300x300 arwork url """ - return artwork_url.replace('large', 't300x300') if artwork_url else None + """Get 300x300 arwork url""" + return artwork_url.replace("large", "t300x300") if artwork_url else None diff --git a/setup.py b/setup.py index d624d5c..fc9fbe3 100644 --- a/setup.py +++ b/setup.py @@ -8,7 +8,6 @@ requirements = [ 'mutagen', - 'bs4', 'aiohttp' ] @@ -27,4 +26,4 @@ install_requires=requirements, test_suite='pytest', tests_require=['pytest', 'pytest-asyncio'], -) \ No newline at end of file +)