Merge pull request #223 from tamland/138-add-oauth-login-from-file-he…

…lper-functions

feature/138 add oauth login from file helper functions

.gitignore

@@ -75,4 +75,10 @@ prof/
 .venv
 
 # MacOS
-.DS_Store
+.DS_Store
+
+# OAuth json session files
+*tidal-oauth*
+
+# Misc. csv. files that might be generated when executing examples
+*.csv

README.rst

@@ -11,10 +11,6 @@ Unofficial Python API for TIDAL music streaming service.
 
 Requires Python 3.9 or higher.
 
-0.7.x Migration guide
----------------------
-The 0.7.x rewrite is now complete, see the `migration guide <https://tidalapi.netlify.app/migration.html#migrating-from-0-6-x-0-7-x>`_ for dealing with it
-
 Installation
 ------------
 
@@ -29,27 +25,7 @@ Install from `PyPI <https://pypi.python.org/pypi/tidalapi/>`_ using ``pip``:
 Example usage
 -------------
 
-.. code-block:: python
-
-    import tidalapi
-
-    session = tidalapi.Session()
-    # Will run until you visit the printed url and link your account
-    session.login_oauth_simple()
-    # Override the required playback quality, if necessary
-    # Note: Set the quality according to your subscription.
-    # Normal: Quality.low_320k
-    # HiFi: Quality.high_lossless
-    # HiFi+ Quality.hi_res_lossless
-    session.audio_quality = Quality.low_320k
-
-    album = session.album(66236918)
-    tracks = album.tracks()
-    for track in tracks:
-        print(track.name)
-        for artist in track.artists:
-            print(' by: ', artist.name)
-
+For examples on how to use the api, see the `examples` directory.
 
 Documentation
 -------------

examples/simple.py

@@ -0,0 +1,43 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2023- The Tidalapi Developers
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+"""simple.py: A simple example script that describes how to get started using tidalapi"""
+
+import tidalapi
+from tidalapi import Quality
+from pathlib import Path
+
+oauth_file1 = Path("tidal-oauth-user.json")
+
+session = tidalapi.Session()
+# Will run until you visit the printed url and link your account
+session.login_oauth_file(oauth_file1)
+# Override the required playback quality, if necessary
+# Note: Set the quality according to your subscription.
+# Normal: Quality.low_320k
+# HiFi: Quality.high_lossless
+# HiFi+ Quality.hi_res_lossless
+session.audio_quality = Quality.low_320k
+
+album = session.album(66236918) # Electric For Life Episode 099
+tracks = album.tracks()
+print(album.name)
+# list album tracks
+for track in tracks:
+    print(track.name)
+    for artist in track.artists:
+        print(' by: ', artist.name)

examples/transfer_favorites.py

@@ -0,0 +1,145 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2023- The Tidalapi Developers
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+"""transfer_favorites.py: Use this script to transfer your Tidal favourites from Tidal user A to Tidal user B"""
+import logging
+from pathlib import Path
+import csv
+import time
+import sys
+
+import tidalapi
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+logger.addHandler(logging.StreamHandler(sys.stdout))
+
+oauth_file1 = Path("tidal-oauth-user.json")
+oauth_file2 = Path("tidal-oauth-userB.json")
+
+
+class TidalSession:
+    def __init__(self):
+        self._active_session = tidalapi.Session()
+
+    def get_uid(self):
+        return self._active_session.user.id
+
+    def get_session(self):
+        return self._active_session
+
+
+class TidalTransfer:
+    def __init__(self):
+        self.session_src = TidalSession()
+        self.session_dst = TidalSession()
+
+    def export_csv(self, my_tracks, my_albums, my_artists, my_playlists):
+        logger.info("Exporting user A favorites to csv...")
+        # save to csv file
+        with open("fav_tracks.csv", "w") as file:
+            wr = csv.writer(file, quoting=csv.QUOTE_ALL)
+            for track in my_tracks:
+                wr.writerow(
+                    [
+                        track.id,
+                        track.user_date_added,
+                        track.artist.name,
+                        track.album.name,
+                    ]
+                )
+        with open("fav_albums.csv", "w") as file:
+            wr = csv.writer(file, quoting=csv.QUOTE_ALL)
+            for album in my_albums:
+                wr.writerow(
+                    [album.id, album.user_date_added, album.artist.name, album.name]
+                )
+        with open("fav_artists.csv", "w") as file:
+            wr = csv.writer(file, quoting=csv.QUOTE_ALL)
+            for artist in my_artists:
+                wr.writerow([artist.id, artist.user_date_added, artist.name])
+        with open("fav_playlists.csv", "w") as file:
+            wr = csv.writer(file, quoting=csv.QUOTE_ALL)
+            for playlist in my_playlists:
+                wr.writerow(
+                    [playlist.id, playlist.created, playlist.type, playlist.name]
+                )
+
+    def do_transfer(self):
+        # do login for src and dst Tidal account
+        session_src = self.session_src.get_session()
+        session_dst = self.session_dst.get_session()
+        logger.info("Login to user A (source)...")
+        if not session_src.login_oauth_file(oauth_file1):
+            logger.error("Login to Tidal user...FAILED!")
+            exit(1)
+        logger.info("Login to user B (destination)...")
+        if not session_dst.login_oauth_file(oauth_file2):
+            logger.error("Login to Tidal user...FAILED!")
+            exit(1)
+
+        # get current user favourites (source)
+        my_tracks = session_src.user.favorites.tracks()
+        my_albums = session_src.user.favorites.albums()
+        my_artists = session_src.user.favorites.artists()
+        my_playlists = session_src.user.playlist_and_favorite_playlists()
+        # my_mixes = self._active_session.user.mixes()
+
+        # export to csv
+        self.export_csv(my_tracks, my_albums, my_artists, my_playlists)
+
+        # add favourites to new user
+        logger.info("Adding favourites to Tidal user B...")
+        for idx, track in enumerate(my_tracks):
+            logger.info("Adding track {}/{}".format(idx, len(my_tracks)))
+            try:
+                session_dst.user.favorites.add_track(track.id)
+                time.sleep(0.1)
+            except:
+                logger.error("error while adding track {} {}".format(track.id, track.name))
+
+        for idx, album in enumerate(my_albums):
+            logger.info("Adding album {}/{}".format(idx, len(my_albums)))
+            try:
+                session_dst.user.favorites.add_album(album.id)
+                time.sleep(0.1)
+            except:
+                logger.error("error while adding album {} {}".format(album.id, album.name))
+
+        for idx, artist in enumerate(my_artists):
+            logger.info("Adding artist {}/{}".format(idx, len(my_artists)))
+            try:
+                session_dst.user.favorites.add_artist(artist.id)
+                time.sleep(0.1)
+            except:
+                logger.error("error while adding artist {} {}".format(artist.id, artist.name))
+
+        for idx, playlist in enumerate(my_playlists):
+            logger.info("Adding playlist {}/{}".format(idx, len(my_playlists)))
+            try:
+                session_dst.user.favorites.add_playlist(playlist.id)
+                time.sleep(0.1)
+            except:
+                logger.error(
+                    "error while adding playlist {} {}".format(
+                        playlist.id, playlist.name
+                    )
+                )
+
+
+if __name__ == "__main__":
+    TidalTransfer().do_transfer()

tidalapi/session.py

@@ -21,12 +21,14 @@
 import base64
 import concurrent.futures
 import datetime
+import json
 import logging
 import random
 import time
 import uuid
 from dataclasses import dataclass
 from enum import Enum
+from pathlib import Path
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -399,13 +401,42 @@ def login(self, username: str, password: str) -> bool:
         self.user = user.User(self, user_id=body["userId"]).factory()
         return True
 
+    def login_oauth_file(self, oauth_file: Path) -> bool:
+        """Logs in to the TIDAL api using an existing OAuth session file. If no OAuth
+        session json file exists, a new one will be created after successful login.
+
+        :param oauth_file: The OAuth session json file
+        :return: Returns true if we think the login was successful.
+        """
+        try:
+            # attempt to reload existing session from file
+            with open(oauth_file) as f:
+                log.info("Loading OAuth session from %s...", oauth_file)
+                data = json.load(f)
+                self._load_oauth_session_from_file(**data)
+        except Exception as e:
+            log.info("Could not load OAuth session from %s: %s", oauth_file, e)
+
+        if not self.check_login():
+            log.info("Creating new OAuth session...")
+            self.login_oauth_simple()
+
+        if self.check_login():
+            log.info("TIDAL Login OK")
+            self._save_oauth_session_to_file(oauth_file)
+            return True
+        else:
+            log.info("TIDAL Login KO")
+            return False
+
     def login_oauth_simple(self, function: Callable[[str], None] = print) -> None:
         """Login to TIDAL using a remote link. You can select what function you want to
         use to display the link.
 
         :param function: The function you want to display the link with
         :raises: TimeoutError: If the login takes too long
         """
+
         login, future = self.login_oauth()
         text = "Visit https://{0} to log in, the code will expire in {1} seconds"
         function(text.format(login.verification_uri_complete, login.expires_in))
@@ -423,6 +454,30 @@ def login_oauth(self) -> Tuple[LinkLogin, concurrent.futures.Future[Any]]:
         login, future = self._login_with_link()
         return login, future
 
+    def _save_oauth_session_to_file(self, oauth_file: Path):
+        # create a new session
+        if self.check_login():
+            # store current OAuth session
+            data = {
+                "token_type": {"data": self.token_type},
+                "session_id": {"data": self.session_id},
+                "access_token": {"data": self.access_token},
+                "refresh_token": {"data": self.refresh_token},
+            }
+            with oauth_file.open("w") as outfile:
+                json.dump(data, outfile)
+            self._oauth_saved = True
+
+    def _load_oauth_session_from_file(self, **data):
+        assert self, "No session loaded"
+        args = {
+            "token_type": data.get("token_type", {}).get("data"),
+            "access_token": data.get("access_token", {}).get("data"),
+            "refresh_token": data.get("refresh_token", {}).get("data"),
+        }
+
+        self.load_oauth_session(**args)
+
     def _login_with_link(self) -> Tuple[LinkLogin, concurrent.futures.Future[Any]]:
         url = "https://auth.tidal.com/v1/oauth2/device_authorization"
         params = {"client_id": self.config.client_id, "scope": "r_usr w_usr w_sub"}