Various improvements

pyproject.toml

@@ -0,0 +1,24 @@
+[build-system]
+requires = ["setuptools >= 61.0", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "spotify_to_tidal"
+version = "0.1.0"
+requires-python = ">= 3.10"
+
+dependencies = [
+  "spotipy==2.21.0",
+  "tidalapi==0.7.3",
+  "pyyaml==6.0",
+  "tqdm==4.64.1",
+]
+
+
+[tools.setuptools.packages."spotify_to_tidal"]
+where = "src"
+include = "spotify_to_tidal*"
+
+[project.scripts]
+spotify_to_tidal = "spotify_to_tidal.__main__:main"
+

readme.md

@@ -5,7 +5,8 @@ Installation
 Clone this git repository and then run:
 
 ```bash
-python3 -m pip install -r requirements.txt
+python3 -m pip install -e .
+spotify_to_tidal -h
 ```
 
 Setup
@@ -21,6 +22,8 @@ Usage
 To synchronize all of your Spotify playlists with your Tidal account run the following
 
 ```bash
+pip install -e .
+spotify_to_tidal
 python3 sync.py
 ```
 
@@ -29,7 +32,7 @@ This will take a long time because the Tidal API is really slow.
 You can also just synchronize a specific playlist by doing the following:
 
 ```bash
-python3 sync.py --uri 1ABCDEqsABCD6EaABCDa0a
+spotify_to_tidal --uri 1ABCDEqsABCD6EaABCDa0a
 ```
 
 See example_config.yml for more configuration options, and `sync.py --help` for more options.

requirements.txt

@@ -3,3 +3,4 @@ requests>=2.28.1 # for tidalapi
 tidalapi==0.7.2
 pyyaml==6.0
 tqdm==4.64.1
+cachetools==5.3.2

src/spotify_to_tidal/__init__.py

@@ -0,0 +1,3 @@
+from .type import __all__ as all_types
+
+__all__ = [].extend(all_types)

src/spotify_to_tidal/__main__.py

@@ -0,0 +1,174 @@
+import logging
+import argparse
+import sys
+import yaml
+from pathlib import Path
+from .filters import Filter429
+from .auth import open_tidal_session, open_spotify_session
+from .parse import (
+    get_tidal_playlists_dict,
+    playlist_id_tuple,
+    create_playlist_id_tuple,
+)
+from .sync import sync_list
+from .type import SyncConfig, SpotifyConfig
+
+from typing import NoReturn
+
+
+def setup_args() -> argparse.ArgumentParser:
+    synopsis = """
+Syncs spotify playlists to Tidal. Can specify a config yaml or specify Spotify Oauth values on the command line.
+
+"""
+    parser = argparse.ArgumentParser(description=synopsis)
+
+    parser.add_argument(
+        "-c",
+        "--config",
+        type=Path,
+        help="Location of the spotify to tidal config file",
+        dest="config",
+    )
+    parser.add_argument(
+        "-i", "--id", type=str, help="Spotify client ID", dest="client_id"
+    )
+    parser.add_argument(
+        "-s", "--secret", type=str, help="Spotify client secret", dest="client_secret"
+    )
+    parser.add_argument(
+        "-U",
+        "--username",
+        type=str,
+        help="Spotify username of the Oauth creds",
+        dest="spotify_uname",
+        default=None,
+    )
+    parser.add_argument(
+        "-u",
+        "--uri",
+        help="Synchronize specific URI(s) instead of the one in the config",
+        nargs='*',
+        dest="uri",
+    )
+    parser.add_argument(
+        "-v",
+        help="verbosity level, increases per v (max of 3) specified",
+        default=0,
+        action="count",
+        dest="verbosity",
+    )
+    parser.add_argument(
+        "-a",
+        "--all",
+        help="Sync all spotify playlists to Tidal. This will not honor the config's exclude list.",
+        action="store_true",
+        default=False,
+        dest="all_playlists",
+    )
+    parser.add_argument(
+        "-e",
+        "--exclude",
+        help="ID of Spotify playlists to exclude",
+        nargs="*",
+        dest="exclude_ids",
+        type=set,
+    )
+
+    return parser
+
+
+logger = None
+
+
+def setup_logging(verbosity: int) -> None:
+    log_map = [logging.WARNING, logging.INFO, logging.DEBUG]
+    strm_hndl = logging.StreamHandler(sys.stdout)
+    logging.root.addFilter(Filter429("tidalapi"))
+    logging.root.addFilter(Filter429("spotipy"))
+    global logger
+    fmt = logging.Formatter(
+        "[%(asctime)s] %(levelname)s %(module)s:%(funcName)s:%(lineno)d - %(message)s"
+    )
+    strm_hndl.setFormatter(fmt)
+    logger = logging.getLogger(__package__)
+    logger.setLevel(log_map[min(verbosity, 2)])
+    logger.addHandler(strm_hndl)
+    logger.debug("Initialized logging.")
+
+
+def parse_args(parser: argparse.ArgumentParser) -> argparse.Namespace | NoReturn:
+    args = parser.parse_args()
+    setup_logging(args.verbosity)
+    if args.config is None:
+        logging.debug("No config specified, checking other args.")
+        if args.client_id is None:
+            raise RuntimeError(
+                "No config specified and Spotify client ID not specified."
+            )
+        if args.client_secret is None:
+            raise RuntimeError(
+                "No config specified and Spotify secret ID not specified."
+            )
+        if args.spotify_uname is None:
+            raise RuntimeError(
+                "No config specified and Spotify username not specified."
+            )
+    if args.config and any(
+        x is not None for x in (args.client_id, args.client_secret, args.spotify_uname)
+    ):
+        raise RuntimeError(
+            "Config specfied with config attributes. Only specify a config or all attributes."
+        )
+    if args.exclude_ids is None:
+        args.exclude_ids = set()
+    # TODO: more validation?
+    return args
+
+
+def main():
+    parser = setup_args()
+    args = parse_args(parser)
+    if args.config:
+        with open(args.config, "r") as f:
+            config: SyncConfig = yaml.safe_load(f)
+        args.exclude_ids.update(*config.get("excluded_playlists", []))
+        spotify_cfg: SpotifyConfig = config.get("spotify", {})
+        args.spotify_uname = spotify_cfg.get("username")
+        args.client_secret = spotify_cfg.get("client_secret")
+        args.client_id = spotify_cfg.get("client_id")
+        args.username = spotify_cfg.get("username")
+        redirect_uri = spotify_cfg.get("redirect_uri", "http://localhost:8888/callback")
+    else:
+        args.config = {}
+        redirect_uri = "http://localhost:8888/callback"
+    spotify_session = open_spotify_session(
+        username=args.username,
+        client_id=args.client_id,
+        client_secret=args.client_secret,
+        redirect_uri=redirect_uri,
+    )
+    tidal_session = open_tidal_session()
+    tidal_playlists = get_tidal_playlists_dict(tidal_session)
+    id_tuples = []
+    if args.uri:
+        # if a playlist ID is explicitly provided as a command line argument then use that
+        for uri in args.uri:
+            spotify_playlist = spotify_session.playlist(uri)
+            id_tuples.append(playlist_id_tuple(spotify_playlist, tidal_playlists))
+    elif args.config and (x := config.get("sync_playlists", [])):
+        for ids in x:
+            id_tuples.append((ids["spotify_id"], ids.get("tidal_id")))
+    elif args.all_playlists or config.get("sync_playlists", None):
+        id_tuples = create_playlist_id_tuple(
+            spotify_session, tidal_session, args.exclude_ids
+        )
+        # sync_list(spotify_session, tidal_session, id_map, config)
+    logger.info("Syncing %d playlists", len(id_tuples))
+
+    sync_list(
+        spotify_session,
+        tidal_session,
+        id_tuples,
+        config,
+    )

src/spotify_to_tidal/auth.py

@@ -0,0 +1,85 @@
+#!/usr/bin/env python3
+import logging
+import sys
+
+from .type import *
+from typing import Union, NoReturn, Optional
+import spotipy
+import tidalapi
+import webbrowser
+import yaml
+
+logger = logging.getLogger(__name__)
+
+
+def open_spotify_session(
+    *, username: str, client_id: str, client_secret: str, redirect_uri: str
+) -> Union[spotipy.Spotify, NoReturn]:
+    credentials_manager = spotipy.SpotifyOAuth(
+        username=username,
+        scope="playlist-read-private",
+        client_id=client_id,
+        client_secret=client_secret,
+        redirect_uri=redirect_uri,
+    )
+    try:
+        credentials_manager.get_access_token(as_dict=False)
+    except spotipy.SpotifyOauthError:
+        logger.critical(
+            "Error opening Spotify sesion; could not get token for username: %s",
+            username,
+        )
+        sys.exit(1)
+
+    return spotipy.Spotify(oauth_manager=credentials_manager)
+
+
+def open_tidal_session(config: Optional[tidalapi.Config] = None) -> tidalapi.Session:
+    try:
+        with open(".session.yml", "r") as session_file:
+            previous_session: TidalConfig = yaml.safe_load(session_file)
+    except OSError:
+        previous_session = None
+
+    if config:
+        session = tidalapi.Session(config=config)
+    else:
+        session = tidalapi.Session()
+    if previous_session:
+        try:
+            if session.load_oauth_session(
+                token_type=previous_session["token_type"],
+                access_token=previous_session["access_token"],
+                refresh_token=previous_session["refresh_token"],
+            ):
+                return session
+        except Exception as e:
+            logger.warn("Error loading previous Tidal Session")
+            logger.debug(e)
+
+    if not session.check_login():
+        logging.critical("Could not connect to Tidal")
+        sys.exit(1)
+
+    login, future = session.login_oauth()
+    print("Login with the webbrowser: " + login.verification_uri_complete)
+    url = login.verification_uri_complete
+    if not url.startswith("https://"):
+        url = "https://" + url
+    webbrowser.open(url)
+    future.result()
+
+    if not session.check_login():
+        logging.critical("Could not connect to Tidal")
+        sys.exit(1)
+    with open(".session.yml", "w") as f:
+        yaml.dump(
+            {
+                "session_id": session.session_id,
+                "token_type": session.token_type,
+                "access_token": session.access_token,
+                "refresh_token": session.refresh_token,
+            },
+            f,
+        )
+    return session