Formatting

Signed-off-by: nathom <nathanthomas707@gmail.com>

streamrip/cli.py

@@ -7,7 +7,7 @@ from getpass import getpass
 from hashlib import md5
 
 import click
-import requests  # type: ignore
+import requests
 
 from . import __version__
 from .clients import TidalClient
@@ -15,7 +15,6 @@ from .config import Config
 from .constants import CACHE_DIR, CONFIG_DIR, CONFIG_PATH, QOBUZ_FEATURED_KEYS
 from .core import MusicDL
 
-
 logging.basicConfig(level="WARNING")
 logger = logging.getLogger("streamrip")
 
@@ -26,9 +25,7 @@ if not os.path.isdir(CACHE_DIR):
 
 
 @click.group(invoke_without_command=True)
-@click.option(
-    "-c", "--convert", metavar="CODEC", help="alac, mp3, flac, or ogg"
-)
+@click.option("-c", "--convert", metavar="CODEC", help="alac, mp3, flac, or ogg")
 @click.option(
     "-u",
     "--urls",
@@ -141,9 +138,7 @@ def filter_discography(ctx, **kwargs):
 
 
 @cli.command()
-@click.option(
-    "-t", "--type", default="album", help="album, playlist, track, or artist"
-)
+@click.option("-t", "--type", default="album", help="album, playlist, track, or artist")
 @click.option(
     "-s",
     "--source",
@@ -264,9 +259,7 @@ def lastfm(ctx, source, url):
 
 @cli.command()
 @click.option("-o", "--open", is_flag=True, help="Open the config file")
-@click.option(
-    "-d", "--directory", is_flag=True, help="Open the config directory"
-)
+@click.option("-d", "--directory", is_flag=True, help="Open the config directory")
 @click.option("-q", "--qobuz", is_flag=True, help="Set Qobuz credentials")
 @click.option("-t", "--tidal", is_flag=True, help="Re-login into Tidal")
 @click.option("--reset", is_flag=True, help="RESET the config file")
@@ -310,9 +303,7 @@ def config(ctx, **kwargs):
         click.launch(config_dir)
 
     if kwargs["qobuz"]:
-        config.file["qobuz"]["email"] = input(
-            click.style("Qobuz email: ", fg="blue")
-        )
+        config.file["qobuz"]["email"] = input(click.style("Qobuz email: ", fg="blue"))
 
         click.secho("Qobuz password (will not show on screen):", fg="blue")
         config.file["qobuz"]["password"] = md5(
@@ -320,9 +311,7 @@ def config(ctx, **kwargs):
         ).hexdigest()
 
         config.save()
-        click.secho(
-            "Qobuz credentials hashed and saved to config.", fg="green"
-        )
+        click.secho("Qobuz credentials hashed and saved to config.", fg="green")
 
     if kwargs["tidal"]:
         client = TidalClient()

streamrip/clients.py

@@ -216,9 +216,7 @@ class QobuzClient(Client):
         :rtype: dict
         """
         page, status_code = self._api_request(epoint, params)
-        logger.debug(
-            "Keys returned from _gen_pages: %s", ", ".join(page.keys())
-        )
+        logger.debug("Keys returned from _gen_pages: %s", ", ".join(page.keys()))
         key = epoint.split("/")[0] + "s"
         total = page.get(key, {})
         total = total.get("total") or total.get("items")
@@ -242,9 +240,7 @@ class QobuzClient(Client):
         for secret in self.secrets:
             if self._test_secret(secret):
                 self.sec = secret
-                logger.debug(
-                    "Working secret and app_id: %s - %s", secret, self.app_id
-                )
+                logger.debug("Working secret and app_id: %s - %s", secret, self.app_id)
                 break
         if not hasattr(self, "sec"):
             raise InvalidAppSecretError(f"Invalid secrets: {self.secrets}")
@@ -278,15 +274,11 @@ class QobuzClient(Client):
 
         response, status_code = self._api_request(epoint, params)
         if status_code != 200:
-            raise Exception(
-                f'Error fetching metadata. "{response["message"]}"'
-            )
+            raise Exception(f'Error fetching metadata. "{response["message"]}"')
 
         return response
 
-    def _api_search(
-        self, query: str, media_type: str, limit: int = 500
-    ) -> Generator:
+    def _api_search(self, query: str, media_type: str, limit: int = 500) -> Generator:
         """Send a search request to the API.
 
         :param query:
@@ -338,18 +330,14 @@ class QobuzClient(Client):
         resp, status_code = self._api_request(epoint, params)
 
         if status_code == 401:
-            raise AuthenticationError(
-                f"Invalid credentials from params {params}"
-            )
+            raise AuthenticationError(f"Invalid credentials from params {params}")
         elif status_code == 400:
             raise InvalidAppIdError(f"Invalid app id from params {params}")
         else:
             logger.info("Logged in to Qobuz")
 
         if not resp["user"]["credential"]["parameters"]:
-            raise IneligibleError(
-                "Free accounts are not eligible to download tracks."
-            )
+            raise IneligibleError("Free accounts are not eligible to download tracks.")
 
         self.uat = resp["user_auth_token"]
         self.session.headers.update({"X-User-Auth-Token": self.uat})
@@ -398,9 +386,7 @@ class QobuzClient(Client):
         }
         response, status_code = self._api_request("track/getFileUrl", params)
         if status_code == 400:
-            raise InvalidAppSecretError(
-                "Invalid app secret from params %s" % params
-            )
+            raise InvalidAppSecretError("Invalid app secret from params %s" % params)
 
         return response
 
@@ -418,9 +404,7 @@ class QobuzClient(Client):
         try:
             return r.json(), r.status_code
         except Exception:
-            logger.error(
-                "Problem getting JSON. Status code: %s", r.status_code
-            )
+            logger.error("Problem getting JSON. Status code: %s", r.status_code)
             raise
 
     def _test_secret(self, secret: str) -> bool:
@@ -451,9 +435,7 @@ class DeezerClient(Client):
         # no login required
         self.logged_in = True
 
-    def search(
-        self, query: str, media_type: str = "album", limit: int = 200
-    ) -> dict:
+    def search(self, query: str, media_type: str = "album", limit: int = 200) -> dict:
         """Search API for query.
 
         :param query:
@@ -490,9 +472,7 @@ class DeezerClient(Client):
         url = f"{DEEZER_BASE}/{media_type}/{meta_id}"
         item = self.session.get(url).json()
         if media_type in ("album", "playlist"):
-            tracks = self.session.get(
-                f"{url}/tracks", params={"limit": 1000}
-            ).json()
+            tracks = self.session.get(f"{url}/tracks", params={"limit": 1000}).json()
             item["tracks"] = tracks["data"]
             item["track_total"] = len(tracks["data"])
         elif media_type == "artist":
@@ -588,9 +568,7 @@ class TidalClient(Client):
         logger.debug(resp)
         return resp
 
-    def search(
-        self, query: str, media_type: str = "album", limit: int = 100
-    ) -> dict:
+    def search(self, query: str, media_type: str = "album", limit: int = 100) -> dict:
         """Search for a query.
 
         :param query:
@@ -619,19 +597,13 @@ class TidalClient(Client):
             return self._get_video_stream_url(track_id)
 
         params = {
-            "audioquality": get_quality(
-                min(quality, TIDAL_MAX_Q), self.source
-            ),
+            "audioquality": get_quality(min(quality, TIDAL_MAX_Q), self.source),
             "playbackmode": "STREAM",
             "assetpresentation": "FULL",
         }
-        resp = self._api_request(
-            f"tracks/{track_id}/playbackinfopostpaywall", params
-        )
+        resp = self._api_request(f"tracks/{track_id}/playbackinfopostpaywall", params)
         try:
-            manifest = json.loads(
-                base64.b64decode(resp["manifest"]).decode("utf-8")
-            )
+            manifest = json.loads(base64.b64decode(resp["manifest"]).decode("utf-8"))
         except KeyError:
             raise Exception(resp["userMessage"])
 
@@ -837,9 +809,7 @@ class TidalClient(Client):
                     offset += 100
                     tracks_left -= 100
                     resp["items"].extend(
-                        self._api_request(f"{url}/items", {"offset": offset})[
-                            "items"
-                        ]
+                        self._api_request(f"{url}/items", {"offset": offset})["items"]
                     )
 
             item["tracks"] = [item["item"] for item in resp["items"]]
@@ -884,9 +854,7 @@ class TidalClient(Client):
             r'#EXT-X-STREAM-INF:BANDWIDTH=\d+,AVERAGE-BANDWIDTH=\d+,CODECS="[^"]+"'
             r",RESOLUTION=\d+x\d+\n(.+)"
         )
-        manifest = json.loads(
-            base64.b64decode(resp["manifest"]).decode("utf-8")
-        )
+        manifest = json.loads(base64.b64decode(resp["manifest"]).decode("utf-8"))
         available_urls = self.session.get(manifest["urls"][0])
         url_info = re.findall(stream_url_regex, available_urls.text)
 
@@ -965,10 +933,7 @@ class SoundCloudClient(Client):
             url = None
             for tc in track["media"]["transcodings"]:
                 fmt = tc["format"]
-                if (
-                    fmt["protocol"] == "hls"
-                    and fmt["mime_type"] == "audio/mpeg"
-                ):
+                if fmt["protocol"] == "hls" and fmt["mime_type"] == "audio/mpeg":
                     url = tc["url"]
                     break
 

streamrip/core.py

@@ -335,9 +335,7 @@ class MusicDL(list):
 
         parsed.extend(self.url_parse.findall(url))  # Qobuz, Tidal, Dezer
         soundcloud_urls = self.soundcloud_url_parse.findall(url)
-        soundcloud_items = [
-            self.clients["soundcloud"].get(u) for u in soundcloud_urls
-        ]
+        soundcloud_items = [self.clients["soundcloud"].get(u) for u in soundcloud_urls]
 
         parsed.extend(
             ("soundcloud", item["kind"], url)
@@ -369,15 +367,11 @@ class MusicDL(list):
 
         # For testing:
         # https://www.last.fm/user/nathan3895/playlists/12058911
-        user_regex = re.compile(
-            r"https://www\.last\.fm/user/([^/]+)/playlists/\d+"
-        )
+        user_regex = re.compile(r"https://www\.last\.fm/user/([^/]+)/playlists/\d+")
         lastfm_urls = self.lastfm_url_parse.findall(urls)
         try:
             lastfm_source = self.config.session["lastfm"]["source"]
-            lastfm_fallback_source = self.config.session["lastfm"][
-                "fallback_source"
-            ]
+            lastfm_fallback_source = self.config.session["lastfm"]["fallback_source"]
         except KeyError:
             self._config_updating_message()
             self.config.update()
@@ -407,9 +401,7 @@ class MusicDL(list):
                 except (NoResultsFound, StopIteration):
                     return None
 
-            track = try_search(lastfm_source) or try_search(
-                lastfm_fallback_source
-            )
+            track = try_search(lastfm_source) or try_search(lastfm_fallback_source)
             if track is None:
                 return False
 
@@ -431,9 +423,7 @@ class MusicDL(list):
                 pl.creator = creator_match.group(1)
 
             tracks_not_found = 0
-            with concurrent.futures.ThreadPoolExecutor(
-                max_workers=15
-            ) as executor:
+            with concurrent.futures.ThreadPoolExecutor(max_workers=15) as executor:
                 futures = [
                     executor.submit(search_query, title, artist, pl)
                     for title, artist in queries
@@ -450,9 +440,7 @@ class MusicDL(list):
             pl.loaded = True
 
             if tracks_not_found > 0:
-                click.secho(
-                    f"{tracks_not_found} tracks not found.", fg="yellow"
-                )
+                click.secho(f"{tracks_not_found} tracks not found.", fg="yellow")
             self.append(pl)
 
     def handle_txt(self, filepath: Union[str, os.PathLike]):
@@ -507,9 +495,7 @@ class MusicDL(list):
         else:
             logger.debug("Not generator")
             items = (
-                results.get("data")
-                or results.get("items")
-                or results.get("collection")
+                results.get("data") or results.get("items") or results.get("collection")
             )
             if items is None:
                 raise NoResultsFound(query)
@@ -549,9 +535,7 @@ class MusicDL(list):
             raise NotImplementedError
 
         fields = (fname for _, fname, _, _ in Formatter().parse(fmt) if fname)
-        ret = fmt.format(
-            **{k: media.get(k, default="Unknown") for k in fields}
-        )
+        ret = fmt.format(**{k: media.get(k, default="Unknown") for k in fields})
         return ret
 
     def interactive_search(  # noqa
@@ -682,9 +666,7 @@ class MusicDL(list):
         playlist_title = html.unescape(playlist_title_match.group(1))
 
         if remaining_tracks > 0:
-            with concurrent.futures.ThreadPoolExecutor(
-                max_workers=15
-            ) as executor:
+            with concurrent.futures.ThreadPoolExecutor(max_workers=15) as executor:
                 last_page = int(remaining_tracks // 50) + int(
                     remaining_tracks % 50 != 0
                 )
@@ -726,7 +708,8 @@ class MusicDL(list):
 
             self.config.save()
             click.secho(
-                f'Credentials saved to config file at "{self.config._path}"'
+                f'Credentials saved to config file at "{self.config._path}"',
+                fg="green",
             )
         else:
             raise Exception

streamrip/db.py

@@ -25,9 +25,7 @@ class MusicDB:
         """Create a database at `self.path`."""
         with sqlite3.connect(self.path) as conn:
             try:
-                conn.execute(
-                    "CREATE TABLE downloads (id TEXT UNIQUE NOT NULL);"
-                )
+                conn.execute("CREATE TABLE downloads (id TEXT UNIQUE NOT NULL);")
                 logger.debug("Download-IDs database created: %s", self.path)
             except sqlite3.OperationalError:
                 pass

streamrip/utils.py

@@ -82,9 +82,7 @@ def get_quality(quality_id: int, source: str) -> Union[str, int]:
         raise InvalidSourceError(source)
 
     possible_keys = set(q_map.keys())
-    assert (
-        quality_id in possible_keys
-    ), f"{quality_id} must be in {possible_keys}"
+    assert quality_id in possible_keys, f"{quality_id} must be in {possible_keys}"
     return q_map[quality_id]
 
 
@@ -125,9 +123,7 @@ def get_stats_from_quality(
         raise InvalidQuality(quality_id)
 
 
-def tqdm_download(
-    url: str, filepath: str, params: dict = None, desc: str = None
-):
+def tqdm_download(url: str, filepath: str, params: dict = None, desc: str = None):
     """Download a file with a progress bar.
 
     :param url: url to direct download
@@ -199,9 +195,7 @@ def tidal_cover_url(uuid, size):
     possibles = (80, 160, 320, 640, 1280)
     assert size in possibles, f"size must be in {possibles}"
 
-    return TIDAL_COVER_URL.format(
-        uuid=uuid.replace("-", "/"), height=size, width=size
-    )
+    return TIDAL_COVER_URL.format(uuid=uuid.replace("-", "/"), height=size, width=size)
 
 
 def init_log(path: Optional[str] = None, level: str = "DEBUG"):
@@ -303,9 +297,7 @@ def gen_threadsafe_session(
         headers = {}
 
     session = requests.Session()
-    adapter = requests.adapters.HTTPAdapter(
-        pool_connections=100, pool_maxsize=100
-    )
+    adapter = requests.adapters.HTTPAdapter(pool_connections=100, pool_maxsize=100)
     session.mount("https://", adapter)
     session.headers.update(headers)
     return session