Merge pull request #278 from tamland/feature/v0.7.7

Feature/v0.7.7-prep

HISTORY.rst

@@ -4,6 +4,10 @@ History
 =======
 v0.7.7
 ------
+* Feat.: Provide "Share Link", "Listen link" as an attribute to album/artist/media. Add relevant tests (Fixes #266) - tehkillerbee_
+* Allow switching authentication method oauth/pkce for tests. Default: oauth - tehkillerbee_
+* Tests: Added track stream tests (BTS, MPD) - tehkillerbee_
+* Bugfix: Always use last element in segment timeline. (Fixes #273) - tehkillerbee_
 * Add method to get detailed request error response if an error occurred during request. - tehkillerbee_
 * Tests: Add tests tests for ISRC, barcode methods and cleanup exception handling. - tehkillerbee_
 * Feat.: Add support to get tracks by ISRC. - tehkillerbee_, M4TH1EU_

tests/conftest.py

@@ -101,38 +101,51 @@ def save(self, key: str, val: dict) -> None:
 KEY = "python-tidal"
 
 
-def get_credential_store() -> tuple[List[Credentials], Optional[dict]]:
+def get_credential_store(
+    datastore_key: str = KEY,
+) -> tuple[List[Credentials], Optional[dict]]:
     stores = []
     for store in (EnvCredentials, CachedCredentials, KeyringCredentials):
         with suppress(Exception):
             stores.append(store())
     for store in stores:
-        data = store.load(KEY)
+        data = store.load(datastore_key)
         if data:
             return [store], data
     stores = [s for s in stores if not isinstance(s, EnvCredentials)]
     return stores, None
 
 
-def login(request):
-    stores, credentials = get_credential_store()
+def login(request, use_pkce_auth: bool = False):
+    # Select datastore depending on the required authentication method
+    if use_pkce_auth:
+        datastore_key = f"{KEY}-pkce"
+    else:
+        datastore_key = KEY
+    stores, credentials = get_credential_store(datastore_key)
     config = tidalapi.Config()
 
     tidal_session = tidalapi.Session(config)
     if credentials and tidal_session.load_oauth_session(**credentials):
+        # override pkce state to allow returning non DASH streams
         return tidal_session
     else:
-        credentials = _oauth_login(request, tidal_session)
-
+        # Generate a new login using the required authentication method
+        if use_pkce_auth:
+            _pkce_login(request, tidal_session)
+        else:
+            _oauth_login(request, tidal_session)
     if stores:
+        # Update credentials datastore
         for store in stores:
             with suppress(Exception):
                 store.save(
-                    KEY,
+                    datastore_key,
                     {
                         "token_type": tidal_session.token_type,
                         "access_token": tidal_session.access_token,
                         "refresh_token": tidal_session.refresh_token,
+                        "is_pkce": tidal_session.is_pkce,
                     },
                 )
                 break
@@ -142,6 +155,7 @@ def login(request):
 def _oauth_login(request, tidal_session):
     login, future = tidal_session.login_oauth()
     # https://github.com/pytest-dev/pytest/issues/2704
+    # To be able to print to terminal, global capture must be disabled temporarily:
     capmanager = request.config.pluginmanager.getplugin("capturemanager")
     with capmanager.global_and_fixture_disabled():
         print(
@@ -154,6 +168,14 @@ def _oauth_login(request, tidal_session):
     future.result()
 
 
+def _pkce_login(request, tidal_session):
+    # To be able to print to terminal; read from stdin, global capture must be disabled temporarily:
+    capmanager = request.config.pluginmanager.getplugin("capturemanager")
+    capmanager.suspend_global_capture(in_=True)
+    tidal_session.login_pkce()
+    capmanager.resume_global_capture()
+
+
 def pytest_collection_modifyitems(config, items):
     if config.getoption("--interactive"):
         return

tests/test_album.py

@@ -49,6 +49,8 @@ def test_album(session):
     assert 0 < album.popularity < 100
     assert album.artist.name == "Lasgo"
     assert album.artists[0].name == "Lasgo"
+    assert album.listen_url == "https://listen.tidal.com/album/17927863"
+    assert album.share_url == "https://tidal.com/browse/album/17927863"
 
     with pytest.raises(AttributeError):
         session.album(17927863).video(1280)

tests/test_artist.py

@@ -30,6 +30,8 @@ def test_artist(session):
     artist = session.artist(16147)
     assert artist.id == 16147
     assert artist.name == "Lasgo"
+    assert artist.listen_url == "https://listen.tidal.com/artist/16147"
+    assert artist.share_url == "https://tidal.com/browse/artist/16147"
     assert all(
         role in artist.roles
         for role in [tidalapi.Role.artist, tidalapi.Role.contributor]

tests/test_media.py

@@ -20,11 +20,11 @@
 from datetime import datetime
 
 import pytest
-import requests
 from dateutil import tz
 
 import tidalapi
 from tidalapi.exceptions import MetadataNotAvailable
+from tidalapi.media import AudioExtensions, AudioMode, ManifestMimeType, MimeType
 
 from .cover import verify_image_resolution, verify_video_resolution
 
@@ -52,8 +52,13 @@ def test_track(session):
     )
     assert track.isrc == "NOG841907010"
     assert track.explicit is False
-    assert track.audio_quality == tidalapi.Quality.hi_res
+    assert track.audio_quality == tidalapi.Quality.high_lossless
     assert track.album.name == "Alone, Pt. II"
+    assert track.album.id == 125169472
+    assert (
+        track.listen_url == "https://listen.tidal.com/album/125169472/track/125169484"
+    )
+    assert track.share_url == "https://tidal.com/browse/track/125169484"
 
     assert track.artist.name == "Alan Walker"
     artist_names = [artist.name for artist in track.artists]
@@ -99,7 +104,7 @@ def test_track_with_album(session):
 def test_track_streaming(session):
     track = session.track(62392768)
     stream = track.get_stream()
-    assert stream.audio_mode == tidalapi.media.AudioMode.stereo
+    assert stream.audio_mode == AudioMode.stereo
     assert (
         stream.audio_quality == tidalapi.Quality.low_320k
     )  # i.e. the default quality for the current session
@@ -122,9 +127,13 @@ def test_video(session):
     assert video.album is None
 
     assert video.artist.name == "Alan Walker"
+    assert video.artist.id == 6159368
     artist_names = [artist.name for artist in video.artists]
     assert [artist in artist_names for artist in ["Alan Walker", "Ava Max"]]
 
+    assert video.listen_url == "https://listen.tidal.com/artist/6159368/video/125506698"
+    assert video.share_url == "https://tidal.com/browse/video/125506698"
+
 
 def test_video_no_release_date(session):
     video = session.video(151050672)
@@ -233,3 +242,100 @@ def test_get_track_radio_limit_100(session):
     track = session.track(182912246)
     similar_tracks = track.get_track_radio(limit=100)
     assert len(similar_tracks) == 100
+
+
+def test_get_stream_bts(session):
+    track = session.track(77646170)  # Beck: Sea Change, Track: The Golden Age
+    # Set session as BTS type (i.e. HIGH Quality)
+    session.audio_quality = "HIGH"
+    # Attempt to get stream and validate
+    stream = track.get_stream()
+    validate_stream(stream, False)
+    # Get parsed stream manifest, audio resolutions
+    manifest = stream.get_stream_manifest()
+    validate_stream_manifest(manifest, False)
+    audio_resolution = stream.get_audio_resolution()
+    assert audio_resolution[0] == 16
+    assert audio_resolution[1] == 44100
+
+
+def test_get_stream_mpd(session):
+    track = session.track(77646170)
+    # Set session as MPD/DASH type (i.e. HI_RES_LOSSLESS Quality).
+    session.audio_quality = "HI_RES_LOSSLESS"
+    # Attempt to get stream and validate
+    stream = track.get_stream()
+    validate_stream(stream, True)
+    # Get parsed stream manifest
+    manifest = stream.get_stream_manifest()
+    validate_stream_manifest(manifest, True)
+
+
+def test_manifest_element_count(session):
+    # Certain tracks has only one element in their SegmentTimeline
+    #   and must be handled slightly differently when parsing the stream manifest DashInfo
+    track = session.track(281047832)
+    # Set session as MPD/DASH type (i.e. HI_RES_LOSSLESS Quality).
+    session.audio_quality = "HI_RES_LOSSLESS"
+    # Attempt to get stream
+    stream = track.get_stream()
+    # Get parsed stream manifest
+    stream.get_stream_manifest()
+
+
+def validate_stream(stream, is_hi_res_lossless: bool = False):
+    assert stream.album_peak_amplitude == 1.0
+    assert stream.album_replay_gain == -11.8
+    assert stream.asset_presentation == "FULL"
+    assert stream.audio_mode == "STEREO"
+    if not is_hi_res_lossless:
+        assert stream.audio_quality == "HIGH"
+        assert stream.is_BTS == True
+        assert stream.is_MPD == False
+        assert stream.bit_depth == 16
+        assert stream.sample_rate == 44100
+        assert stream.manifest_mime_type == ManifestMimeType.BTS
+        audio_resolution = stream.get_audio_resolution()
+        assert audio_resolution[0] == 16
+        assert audio_resolution[1] == 44100
+    else:
+        assert stream.audio_quality == "HI_RES_LOSSLESS"
+        assert stream.is_BTS == False
+        assert stream.is_MPD == True
+        assert stream.bit_depth == 24
+        assert stream.sample_rate == 192000  # HI_RES_LOSSLESS: 24bit/192kHz
+        assert stream.manifest_mime_type == ManifestMimeType.MPD
+        audio_resolution = stream.get_audio_resolution()
+        assert audio_resolution[0] == 24
+        assert audio_resolution[1] == 192000
+    assert stream.track_id == 77646170
+    assert stream.track_peak_amplitude == 1.0
+    assert stream.track_replay_gain == -9.62
+
+
+def validate_stream_manifest(manifest, is_hi_res_lossless: bool = False):
+    if not is_hi_res_lossless:
+        assert manifest.is_BTS == True
+        assert manifest.is_MPD == False
+        assert manifest.codecs == "MP4A"
+        assert manifest.dash_info is None
+        assert manifest.encryption_key is None
+        assert manifest.encryption_type == "NONE"
+        assert manifest.file_extension == AudioExtensions.MP4
+        assert manifest.is_encrypted == False
+        assert manifest.manifest_mime_type == ManifestMimeType.BTS
+        assert manifest.mime_type == MimeType.audio_mp4
+        assert manifest.sample_rate == 44100
+    else:
+        assert manifest.is_BTS == False
+        assert manifest.is_MPD == True
+        assert manifest.codecs == "flac"
+        assert manifest.dash_info is not None
+        assert manifest.encryption_key is None
+        assert manifest.encryption_type == "NONE"
+        assert manifest.file_extension == AudioExtensions.MP4
+        assert manifest.is_encrypted == False
+        assert manifest.manifest_mime_type == ManifestMimeType.MPD
+        assert manifest.mime_type == MimeType.audio_mp4
+        assert manifest.sample_rate == 192000
+    # TODO Validate stream URL contents

tidalapi/album.py

@@ -73,6 +73,11 @@ class Album:
     artist: Optional["Artist"] = None
     artists: Optional[List["Artist"]] = None
 
+    # Direct URL to https://listen.tidal.com/album/<album_id>
+    listen_url: str = ""
+    # Direct URL to https://tidal.com/browse/album/<album_id>
+    share_url: str = ""
+
     def __init__(self, session: "Session", album_id: Optional[str]):
         self.session = session
         self.request = session.request
@@ -148,6 +153,8 @@ def parse(
         self.user_date_added = (
             dateutil.parser.isoparse(user_date_added) if user_date_added else None
         )
+        self.listen_url = f"{self.session.config.listen_base_url}/album/{self.id}"
+        self.share_url = f"{self.session.config.share_base_url}/album/{self.id}"
 
         return copy.copy(self)
 

tidalapi/artist.py

@@ -48,6 +48,11 @@ class Artist:
     user_date_added: Optional[datetime] = None
     bio: Optional[str] = None
 
+    # Direct URL to https://listen.tidal.com/artist/<artist_id>
+    listen_url: str = ""
+    # Direct URL to https://tidal.com/browse/artist/<artist_id>
+    share_url: str = ""
+
     def __init__(self, session: "Session", artist_id: Optional[str]):
         """Initialize the :class:`Artist` object, given a TIDAL artist ID :param
         session: The current TIDAL :class:`Session` :param str artist_id: TIDAL artist
@@ -97,6 +102,9 @@ def parse_artist(self, json_obj: JsonObj) -> "Artist":
             dateutil.parser.isoparse(user_date_added) if user_date_added else None
         )
 
+        self.listen_url = f"{self.session.config.listen_base_url}/artist/{self.id}"
+        self.share_url = f"{self.session.config.share_base_url}/artist/{self.id}"
+
         return copy.copy(self)
 
     def parse_artists(self, json_obj: List[JsonObj]) -> List["Artist"]:

tidalapi/media.py

@@ -141,6 +141,7 @@ def __str__(self) -> str:
 class MimeType(str, Enum):
     audio_mpeg = "audio/mpeg"
     audio_mp3 = "audio/mp3"
+    audio_mp4 = "audio/mp4"
     audio_m4a = "audio/m4a"
     audio_flac = "audio/flac"
     audio_xflac = "audio/x-flac"
@@ -199,6 +200,10 @@ class Media:
     artists: Optional[List["tidalapi.artist.Artist"]] = None
     album: Optional["tidalapi.album.Album"] = None
     type: Optional[str] = None
+    # Direct URL to media https://listen.tidal.com/track/<id> or https://listen.tidal.com/browse/album/<album_id>/track/<track_id>
+    listen_url: str = ""
+    # Direct URL to media https://tidal.com/browse/track/<id>
+    share_url: str = ""
 
     def __init__(
         self, session: "tidalapi.session.Session", media_id: Optional[str] = None
@@ -305,6 +310,12 @@ def parse_track(self, json_obj: JsonObj) -> Track:
             self.full_name = f"{json_obj['title']} ({json_obj['version']})"
         else:
             self.full_name = json_obj["title"]
+        # Generate share URLs from track ID and album (if it exists)
+        if self.album:
+            self.listen_url = f"{self.session.config.listen_base_url}/album/{self.album.id}/track/{self.id}"
+        else:
+            self.listen_url = f"{self.session.config.listen_base_url}/track/{self.id}"
+        self.share_url = f"{self.session.config.share_base_url}/track/{self.id}"
 
         return copy.copy(self)
 
@@ -721,7 +732,7 @@ def __init__(self, mpd_xml):
             .representations[0]
             .segment_templates[0]
             .segment_timelines[0]
-            .Ss[1]
+            .Ss[-1]  # Always use last element in segment timeline.
             .d
         )
 
@@ -816,6 +827,13 @@ def parse_video(self, json_obj: JsonObj) -> Video:
         # Videos found in the /pages endpoints don't have quality
         self.video_quality = json_obj.get("quality")
 
+        # Generate share URLs from track ID and artist (if it exists)
+        if self.artist:
+            self.listen_url = f"{self.session.config.listen_base_url}/artist/{self.artist.id}/video/{self.id}"
+        else:
+            self.listen_url = f"{self.session.config.listen_base_url}/video/{self.id}"
+        self.share_url = f"{self.session.config.share_base_url}/video/{self.id}"
+
         return copy.copy(self)
 
     def _get(self, media_id: str) -> Video:

tidalapi/session.py

@@ -119,6 +119,9 @@ class Config:
     code_challenge: str
     pkce_uri_redirect: str = "https://tidal.com/android/login/auth"
     client_id_pkce: str
+    # Base URLs for sharing, listen URLs
+    listen_base_url: str = "https://listen.tidal.com"
+    share_base_url: str = "https://tidal.com/browse"
 
     @no_type_check
     def __init__(