diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index f1692a6..9f235fe 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -26,4 +26,4 @@ jobs: - run: poetry env use python - run: poetry install # Offline tests only, the remaining tests require a TIDAL account - - run: poetry run pytest tests/test_backoff.py tests/test_parsing.py tests/test_token_refresh.py + - run: poetry run pytest tests/test_backoff.py tests/test_parsing.py tests/test_token_refresh.py tests/test_playlog.py diff --git a/HISTORY.rst b/HISTORY.rst index 1b079e2..4d17204 100644 --- a/HISTORY.rst +++ b/HISTORY.rst @@ -3,6 +3,10 @@ History ======= +Unreleased +---------- +* Added `Session.start_playback` and `Session.report_playback` to measure plays and report them to TIDAL, so they show up in Recently played on all devices and count for recommendations (Fixes #401) - Petru101_ + v0.9.0 ------ * (BREAKING!) Drop support for python3.10 and older, python 3.11 or newer is now required - tehkillerbee_ diff --git a/docs/api.rst b/docs/api.rst index 317b439..c9641a0 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -167,3 +167,28 @@ API Reference .. autoclass:: Mix :members: :undoc-members: + +`tidalapi.playlog` +------------------ + +.. automodule:: tidalapi.playlog + +.. autoclass:: PlaybackTracker + :members: + +.. autoclass:: PlaybackSession + :members: + +.. autoclass:: PlaybackReport + :members: + +.. autoclass:: PlaybackAction + :members: + +.. autoclass:: SourceType + :members: + :undoc-members: + +.. autoclass:: ActionType + :members: + :undoc-members: diff --git a/tests/test_playlog.py b/tests/test_playlog.py new file mode 100644 index 0000000..645b7c7 --- /dev/null +++ b/tests/test_playlog.py @@ -0,0 +1,694 @@ +# 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 . +"""Offline tests for reporting plays (no TIDAL account needed, nothing is sent).""" + +import base64 +import dataclasses +import json +import platform +from unittest import mock + +import pytest +import requests + +import tidalapi +from tidalapi.exceptions import TidalBackoffError +from tidalapi.media import Stream +from tidalapi.playlog import ( + MAX_PLAY_AGE_MS, + ActionType, + PlaybackAction, + PlaybackSession, + SourceType, + get_token_claims, +) + +NOW = 1_800_000_000_000 + + +def make_token(claims): + payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode() + return f"header.{payload.rstrip('=')}.signature" + + +def fake_response(status_code=200, text=""): + response = requests.Response() + response.status_code = status_code + response._content = text.encode() + response.request = requests.Request( + "POST", "https://ec.tidal.com/api/event-batch" + ).prepare() + return response + + +def reply_xml(sent=(), failed=(), sender_fault=True): + """A reply as TIDAL sends it, with the namespace.""" + entries = [ + f"{event_id}" + "m" + for event_id in sent + ] + entries += [ + f"{event_id}" + f"{str(sender_fault).lower()}" + "" + for event_id in failed + ] + return ( + '' + f"{''.join(entries)}" + "" + ) + + +def event_ids(form): + return [v for k, v in form.items() if k.endswith(".Id")] + + +def accept_everything(method, url, data, **kwargs): + return fake_response(text=reply_xml(event_ids(data))) + + +def make_session(): + session = tidalapi.Session(tidalapi.Config()) + session.token_type = "Bearer" + session.access_token = make_token({"uid": 123, "cid": 8017, "sid": "s-1"}) + session.refresh_token = "refresh" + session.is_pkce = True + session.request_session = mock.Mock() + session.request_session.request.side_effect = accept_everything + return session + + +def make_track(session, track_id=1, duration=200, fetched=True): + track = tidalapi.Track(session) + track.id = track_id + track.duration = duration + if fetched: + session.playlog.note_stream(track_id, "HI_RES_LOSSLESS", "DOLBY_ATMOS") + return track + + +def make_play(start=NOW - 300_000, seconds=200, **kwargs): + """A plausible play, `seconds` long, ending before NOW.""" + values = { + "track_id": 1, + "track_duration": 200, + "start_timestamp": start, + "end_timestamp": start + seconds * 1000, + "start_position": 0.0, + "end_position": float(seconds), + } + return PlaybackSession(**{**values, **kwargs}) + + +def sent_forms(session): + return [c.kwargs["data"] for c in session.request_session.request.call_args_list] + + +def entries(form): + """The (body, headers, name) of each event in a SendMessageBatch form.""" + events = [] + index = 1 + while f"SendMessageBatchRequestEntry.{index}.Id" in form: + entry = f"SendMessageBatchRequestEntry.{index}." + events.append( + ( + json.loads(form[entry + "MessageBody"]), + json.loads(form[entry + "MessageAttribute.2.Value.StringValue"]), + form[entry + "MessageAttribute.1.Value.StringValue"], + ) + ) + index += 1 + return events + + +class FakeTime: + """The clocks the library measures with: seconds of wall clock and monotonic.""" + + def __init__(self): + self.now = NOW / 1000 + + def time(self): + return self.now + + def monotonic(self): + return self.now + + def advance(self, seconds): + self.now += seconds + + +@pytest.fixture +def clock(): + fake = FakeTime() + with mock.patch("tidalapi.playlog.time", fake): + yield fake + + +@pytest.fixture +def session(clock): + return make_session() + + +# Measuring a play + + +def test_tracker_measures_a_play(session, clock): + track = make_track(session) + tracker = session.start_playback(track) + clock.advance(120) + play = tracker.stop() + + assert play.track_id == 1 + assert play.start_timestamp == NOW + assert play.end_timestamp == NOW + 120_000 + assert play.start_position == 0.0 + assert play.end_position == 120.0 + assert play.actions == [] + assert play.quality == "HI_RES_LOSSLESS" + assert play.audio_mode == "DOLBY_ATMOS" + assert play.source_type == SourceType.item + assert play.source_id == 1 + + +def test_tracker_doesnt_count_pauses(session, clock): + tracker = session.start_playback(make_track(session)) + clock.advance(40) + tracker.pause() + clock.advance(1000) + assert tracker.position == 40.0 + tracker.resume() + clock.advance(10) + play = tracker.stop() + + assert play.end_position == 50.0 + assert tracker.listened_seconds == 50.0 + assert play.actions == [ + PlaybackAction(NOW + 40_000, 40.0, ActionType.playback_stop), + PlaybackAction(NOW + 1_040_000, 40.0, ActionType.playback_start), + ] + + +def test_tracker_records_seeks_as_the_players_do(session, clock): + tracker = session.start_playback(make_track(session)) + clock.advance(40) + tracker.seek(100.0) + clock.advance(10) + play = tracker.stop() + + assert play.end_position == 110.0 + assert play.actions == [ + PlaybackAction(NOW + 40_000, 40.0, ActionType.playback_stop), + PlaybackAction(NOW + 40_000, 100.0, ActionType.playback_start), + ] + assert play.get_problem(int(clock.now * 1000)) is None + + +def test_seeking_back_counts_the_time_listened(session, clock): + tracker = session.start_playback(make_track(session)) + clock.advance(20) + tracker.seek(0.0) + clock.advance(20) + play = tracker.stop() + + assert play is not None + assert play.end_position == 20.0 + assert play.get_problem(int(clock.now * 1000)) is None + + +def test_a_play_started_paused_begins_at_the_first_resume(session, clock): + tracker = session.start_playback(make_track(session), position=30.0, paused=True) + clock.advance(500) + tracker.seek(50.0) + tracker.resume() + clock.advance(40) + play = tracker.stop() + + assert play.start_timestamp == NOW + 500_000 + assert play.start_position == 50.0 + assert play.end_position == 90.0 + assert play.actions == [] + + +def test_stopping_twice_gives_the_same_play(session, clock): + tracker = session.start_playback(make_track(session)) + clock.advance(40) + play = tracker.stop() + clock.advance(40) + assert tracker.stop() is play + assert play.end_position == 40.0 + + +def test_short_plays_dont_count(session, clock): + tracker = session.start_playback(make_track(session)) + clock.advance(29) + assert tracker.stop() is None + + tracker = session.start_playback(make_track(session), paused=True) + clock.advance(100) + assert tracker.stop() is None + + +def test_a_track_shorter_than_30_seconds_counts_when_played_through(session, clock): + tracker = session.start_playback(make_track(session, duration=20)) + clock.advance(19.5) + assert tracker.stop() is not None + + +def test_the_position_stops_at_the_end_of_the_track(session, clock): + tracker = session.start_playback(make_track(session, duration=200)) + clock.advance(203) + assert tracker.stop().end_position == 200.0 + + +def test_only_tracks_with_a_fetched_stream_can_be_tracked(session): + track = make_track(session, fetched=False) + with pytest.raises(ValueError): + session.start_playback(track) + assert session.start_playback(track, local_copy=True) is not None + + +def test_get_stream_makes_a_track_playable(session, clock): + track = make_track(session, fetched=False) + stream = Stream() + stream.audio_quality = "LOSSLESS" + stream.audio_mode = "STEREO" + with ( + mock.patch.object(track.requests, "request"), + mock.patch.object(track.requests, "map_json", return_value=stream), + ): + track.get_stream() + + tracker = session.start_playback(track) + clock.advance(40) + assert tracker.stop().quality == "LOSSLESS" + + +@pytest.mark.parametrize( + "source_class, source_type", + [ + (tidalapi.Album, SourceType.album), + (tidalapi.Playlist, SourceType.playlist), + (tidalapi.Mix, SourceType.mix), + (tidalapi.MixV2, SourceType.mix), + (tidalapi.Artist, SourceType.artist), + ], +) +def test_source(session, clock, source_class, source_type): + source = source_class.__new__(source_class) + source.id = "source-id" + tracker = session.start_playback(make_track(session), source=source) + clock.advance(40) + play = tracker.stop() + assert play.source_type == source_type + assert play.source_id == "source-id" + + +# The events + + +def test_payload(): + play = make_play( + source_type=SourceType.album, + source_id=55, + actions=[ + PlaybackAction(50_000, 49.5, ActionType.playback_stop), + PlaybackAction(60_000, 49.5, ActionType.playback_start), + ], + quality=tidalapi.Quality.hi_res_lossless, + ) + payload = play.to_json() + + assert payload["requestedProductId"] == "1" + assert payload["actualProductId"] == "1" + assert payload["productType"] == "TRACK" + assert payload["sourceType"] == "ALBUM" + assert payload["sourceId"] == "55" + assert payload["actualQuality"] == "HI_RES_LOSSLESS" + assert payload["actualAudioMode"] == "STEREO" + assert payload["startTimestamp"] == play.start_timestamp + assert payload["endTimestamp"] == play.end_timestamp + assert payload["startAssetPosition"] == 0.0 + assert payload["endAssetPosition"] == 200.0 + assert payload["actions"] == [ + {"timestamp": 50_000, "assetPosition": 49.5, "actionType": "PLAYBACK_STOP"}, + {"timestamp": 60_000, "assetPosition": 49.5, "actionType": "PLAYBACK_START"}, + ] + assert payload["playbackSessionId"] + + +def test_track_played_on_its_own_uses_its_own_id_as_source(): + payload = make_play(track_id=7).to_json() + assert payload["sourceType"] == "ITEM" + assert payload["sourceId"] == "7" + + +def test_token_claims(): + assert get_token_claims(make_token({"uid": 1, "cid": 2})) == {"uid": 1, "cid": 2} + assert get_token_claims("not a token") == {} + assert get_token_claims("a.%%%.b") == {} + + +def test_report_sends_one_playback_session_event_per_play(session): + plays = [make_play(track_id=1), make_play(track_id=2, start=NOW - 700_000)] + + report = session.report_playback(plays, app_version="App 1.0") + assert report.accepted == plays + + call = session.request_session.request.call_args + assert call.args == ("POST", "https://ec.tidal.com/api/event-batch") + assert call.kwargs["headers"]["authorization"] == "Bearer " + session.access_token + + events = entries(call.kwargs["data"]) + assert [body["payload"]["requestedProductId"] for body, _, _ in events] == [ + "1", + "2", + ] + body, headers, name = events[0] + assert name == "playback_session" + assert body["group"] == "play_log" + assert body["version"] == 2 + assert body["uuid"] == plays[0].event_id + assert body["user"] == {"id": 123, "clientId": 8017, "sessionId": "s-1"} + assert body["client"] == { + "token": "8017", + "deviceType": "desktop", + "version": "App 1.0", + "platform": platform.system().lower(), + } + assert headers["app-version"] == "App 1.0" + assert headers["os-name"] == platform.system() + assert headers["client-id"] == session.config.client_id_pkce + assert headers["authorization"] == session.access_token + assert headers["consent-category"] == "NECESSARY" + + +def test_events_describe_the_real_client(session): + session.report_playback([make_play()]) + + body, headers, _ = entries(sent_forms(session)[0])[0] + assert "device-model" not in headers + assert "device-vendor" not in headers + assert body["client"]["platform"] != "android" + assert headers["app-version"] == f"tidalapi {tidalapi.__version__}" + + +# TIDAL's replies + + +def spaced_plays(count): + return [ + make_play(track_id=i, start=NOW - 1000 * 300 * (i + 1), seconds=200) + for i in range(count) + ] + + +def test_at_most_ten_events_per_request(session): + report = session.report_playback(spaced_plays(23)) + assert [len(entries(form)) for form in sent_forms(session)] == [10, 10, 3] + assert len(report.accepted) == 23 + + +@pytest.mark.parametrize("status", [400, 403, 404, 408, 429, 500, 503]) +def test_http_errors_keep_the_plays_for_later(session, status): + session.request_session.request.side_effect = [fake_response(status, "no")] + play = make_play() + + report = session.report_playback([play]) + + assert report.unsent == [play] + assert report.accepted == [] + assert report.rejected == [] + assert report.error is not None + + +def test_being_offline_keeps_the_plays_for_later(session): + error = requests.ConnectionError("offline") + session.request_session.request.side_effect = error + play = make_play() + + report = session.report_playback([play]) + + assert report.unsent == [play] + assert report.error is error + + +def test_a_rate_limit_holds_back_the_next_report(session): + session.request_session.request.side_effect = [fake_response(429, "slow down")] + play = make_play() + assert session.report_playback([play]).unsent == [play] + + report = session.report_playback([play]) + + assert report.unsent == [play] + assert isinstance(report.error, TidalBackoffError) + assert session.request_session.request.call_count == 1 + + +@pytest.mark.parametrize( + "reply", + [ + "", + "Log in to the Wi-Fi", + "Log in to the Wi-Fi", + reply_xml(["some-other-event"]), + reply_xml(), + ], +) +def test_a_success_that_doesnt_confirm_the_plays_keeps_them(session, reply): + session.request_session.request.side_effect = [fake_response(text=reply)] + play = make_play() + + report = session.report_playback([play]) + + assert report.unsent == [play] + assert report.accepted == [] + + +def test_each_play_in_a_reply_gets_its_own_result(session): + sent, sender_fault, server_fault = spaced_plays(3) + reply = reply_xml([sent.event_id], [sender_fault.event_id]).replace( + "", + f"{server_fault.event_id}" + "false" + "", + ) + session.request_session.request.side_effect = [fake_response(text=reply)] + + report = session.report_playback([sent, sender_fault, server_fault]) + + assert report.accepted == [sent] + assert report.rejected == [sender_fault] + assert report.unsent == [server_fault] + + +def test_a_failed_batch_doesnt_hide_the_accepted_ones(session): + plays = spaced_plays(25) + first = fake_response(text=reply_xml([play.event_id for play in plays[:10]])) + session.request_session.request.side_effect = [first, fake_response(503, "no")] + + report = session.report_playback(plays) + + assert report.accepted == plays[:10] + assert report.unsent == plays[10:] + assert session.request_session.request.call_count == 2 + + +def test_retrying_sends_the_same_ids(session): + session.request_session.request.side_effect = [fake_response(503, "no")] + play = make_play() + session.report_playback([play]) + session.request_session.request.side_effect = accept_everything + + assert session.report_playback([play]).accepted == [play] + + first, second = sent_forms(session) + assert event_ids(first) == event_ids(second) == [play.event_id] + + +def test_expired_token_is_refreshed_and_the_request_retried_once(session): + old_token = session.access_token + new_token = make_token({"uid": 123, "cid": 8017, "sid": "s-2"}) + play = make_play() + session.request_session.request.side_effect = [ + fake_response(401, '{"status": 401}'), + fake_response(text=reply_xml([play.event_id])), + ] + + def token_refresh(refresh_token): + session.access_token = new_token + return True + + with mock.patch.object(session, "token_refresh", side_effect=token_refresh): + assert session.report_playback([play]).accepted == [play] + + calls = session.request_session.request.call_args_list + assert len(calls) == 2 + assert calls[0].kwargs["headers"]["authorization"] == "Bearer " + old_token + assert calls[1].kwargs["headers"]["authorization"] == "Bearer " + new_token + body, headers, _ = entries(calls[1].kwargs["data"])[0] + assert headers["authorization"] == new_token + assert body["user"]["sessionId"] == "s-2" + + +# Implausible plays + + +@pytest.mark.parametrize( + "changes", + [ + pytest.param( + {"start_timestamp": NOW - 1000, "end_timestamp": NOW - 200_000}, + id="ends before it starts", + ), + pytest.param( + {"start_timestamp": NOW, "end_timestamp": NOW + 200_000}, + id="ends in the future", + ), + pytest.param( + { + "start_timestamp": NOW - MAX_PLAY_AGE_MS - 300_000, + "end_timestamp": NOW - MAX_PLAY_AGE_MS - 100_000, + }, + id="too old", + ), + pytest.param({"end_position": -1.0}, id="negative position"), + pytest.param({"track_duration": 100}, id="position past the end of the track"), + pytest.param( + {"end_timestamp": NOW - 290_000}, id="more listening than time passed" + ), + pytest.param( + { + "actions": [ + PlaybackAction(NOW - 290_000, 150.0, ActionType.playback_stop), + PlaybackAction(NOW - 289_000, 0.0, ActionType.playback_start), + ] + }, + id="listening before a pause faster than time", + ), + pytest.param( + { + "actions": [ + PlaybackAction(NOW - 200_000, 50.0, ActionType.playback_stop), + PlaybackAction(NOW - 250_000, 50.0, ActionType.playback_start), + ] + }, + id="actions out of order", + ), + pytest.param( + {"actions": [PlaybackAction(NOW, 50.0, ActionType.playback_stop)]}, + id="action after the end", + ), + ], +) +def test_implausible_plays_are_refused_without_sending(session, changes): + play = make_play(**changes) + + report = session.report_playback([play]) + + assert report.rejected == [play] + session.request_session.request.assert_not_called() + + +def test_rounded_durations_are_allowed_for(): + assert make_play(track_duration=199).get_problem(NOW) is None + + +def test_a_clock_adjusted_during_a_play_is_allowed_for(): + play = make_play(seconds=199, end_position=200.5) + assert play.get_problem(NOW) is None + + +def test_a_play_cant_be_reported_twice(session): + play = make_play() + assert session.report_playback([play]).accepted == [play] + + assert session.report_playback([play]).rejected == [play] + assert session.request_session.request.call_count == 1 + + +def test_a_play_cant_be_reported_twice_in_one_call(session): + play = make_play() + report = session.report_playback([play, play]) + assert report.accepted == [play] + assert report.rejected == [play] + + +def test_plays_cant_overlap_a_reported_one(session): + assert session.report_playback([make_play()]).accepted + overlapping = make_play(start=NOW - 250_000, end_position=100.0, seconds=100) + + assert session.report_playback([overlapping]).rejected == [overlapping] + + +def test_plays_cant_overlap_in_one_call(session): + first = make_play() + overlapping = make_play(start=NOW - 250_000, end_position=100.0, seconds=100) + + report = session.report_playback([first, overlapping]) + + assert report.accepted == [first] + assert report.rejected == [overlapping] + + +def test_consecutive_plays_dont_overlap(session): + first = make_play(start=NOW - 500_000) + second = make_play(start=first.end_timestamp) + assert session.report_playback([first, second]).accepted == [first, second] + + +def test_an_unsent_play_can_be_sent_again(session): + session.request_session.request.side_effect = [fake_response(503, "no")] + play = make_play() + assert session.report_playback([play]).unsent == [play] + session.request_session.request.side_effect = accept_everything + assert session.report_playback([play]).accepted == [play] + + +# Keeping plays for later + + +def test_a_kept_play_can_be_reported_later(session, clock): + tracker = session.start_playback(make_track(session)) + clock.advance(40) + tracker.seek(100.0) + clock.advance(10) + tracker.pause() + clock.advance(5) + play = tracker.stop() + + kept = json.loads(json.dumps(play.to_dict())) + restored = PlaybackSession.from_dict(kept) + + assert restored == play + assert restored.to_json() == play.to_json() + new_session = make_session() + assert new_session.report_playback([restored]).accepted == [restored] + + +def test_a_kept_play_goes_through_the_same_checks(session, clock): + kept = make_play().to_dict() + kept["end_position"] = 5000.0 + restored = PlaybackSession.from_dict(kept) + + assert session.report_playback([restored]).rejected == [restored] + + +def test_to_dict_covers_every_field(): + play = make_play(actions=[PlaybackAction(1, 2.0, ActionType.playback_stop)]) + names = {f.name for f in dataclasses.fields(PlaybackSession)} + assert set(play.to_dict()) == names diff --git a/tidalapi/__init__.py b/tidalapi/__init__.py index 6603e01..bedee3c 100644 --- a/tidalapi/__init__.py +++ b/tidalapi/__init__.py @@ -7,6 +7,14 @@ from .mix import Mix, MixV2 # noqa: F401 from .page import Page # noqa: F401 from .playlist import Playlist, UserPlaylist # noqa: F401 +from .playlog import ( # noqa: F401 + ActionType, + PlaybackAction, + PlaybackReport, + PlaybackSession, + PlaybackTracker, + SourceType, +) from .request import Requests # noqa: F401 from .session import Config, Session # noqa: F401 from .user import ( # noqa: F401 diff --git a/tidalapi/media.py b/tidalapi/media.py index 2ff35e9..d56538c 100644 --- a/tidalapi/media.py +++ b/tidalapi/media.py @@ -435,6 +435,9 @@ def get_url(self) -> str: raise e else: json_obj = request.json() + self.session.playlog.note_stream( + self.id, self.session.config.quality, AudioMode.stereo + ) return cast(str, json_obj["urls"][0]) def lyrics(self) -> Lyrics: @@ -527,7 +530,11 @@ def get_stream(self) -> Stream: json_obj = request.json() stream = self.requests.map_json(json_obj, parse=Stream().parse) assert not isinstance(stream, list) - return cast("Stream", stream) + stream = cast("Stream", stream) + self.session.playlog.note_stream( + self.id, stream.audio_quality, stream.audio_mode + ) + return stream @property def is_hi_res_lossless(self) -> bool: diff --git a/tidalapi/playlog.py b/tidalapi/playlog.py new file mode 100644 index 0000000..fc0ad67 --- /dev/null +++ b/tidalapi/playlog.py @@ -0,0 +1,657 @@ +# 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 . +"""Reporting plays to TIDAL (the play log). + +Fetching and playing a stream doesn't count as a play: the official TIDAL players report +every play as a ``playback_session`` event to TIDAL's event producer. Reported plays +show up in Recently played on all devices and in the listening history TIDAL uses for +recommendations. + +The library measures each play with a :class:`PlaybackTracker` and refuses implausible +plays, so reports describe real listening. This keeps honest applications from sending +bad data and makes faking plays less convenient, it can't stop someone who changes the +code. + +The event layout and the handling of replies follow TIDAL's open source SDKs on GitHub: +tidal-music/tidal-sdk-android (modules ``eventproducer`` and ``player/events``) and +tidal-music/tidal-sdk-web (module ``event-producer``). The events describe the real +client (the operating system and the application version), they don't pretend to come +from an official TIDAL app. +""" + +from __future__ import annotations + +import base64 +import json +import logging +import platform +import time +import uuid +from collections.abc import Iterable +from dataclasses import asdict, dataclass, field +from enum import StrEnum +from typing import TYPE_CHECKING, Any +from xml.etree import ElementTree + +import requests + +from tidalapi import media +from tidalapi.album import Album +from tidalapi.artist import Artist +from tidalapi.exceptions import ( + AuthenticationError, + TidalAPIError, + http_error_to_tidal_error, +) +from tidalapi.mix import Mix, MixV2 +from tidalapi.playlist import Playlist +from tidalapi.types import JsonObj + +if TYPE_CHECKING: + from tidalapi.session import Session + +log = logging.getLogger(__name__) + +#: The most events TIDAL's event producer accepts in one request +MAX_BATCH_SIZE = 10 +#: TIDAL counts a play after this much listening, shorter plays aren't reported +MIN_LISTENED_SECONDS = 30 +#: Plays older than this aren't reported anymore, in milliseconds +MAX_PLAY_AGE_MS = 14 * 24 * 3600 * 1000 +#: Slack for clocks being adjusted during a play, in milliseconds +CLOCK_TOLERANCE_MS = 2000 +#: Slack for track durations being rounded to whole seconds, and for the position +#: advancing ahead of the clock, in seconds +POSITION_TOLERANCE_SECONDS = 2.0 + + +class SourceType(StrEnum): + """What a track was played from. + + A track played on its own (from search, a list of tracks, the play queue) is an + ``ITEM`` with the track's own id as source id. + """ + + album = "ALBUM" + playlist = "PLAYLIST" + mix = "MIX" + artist = "ARTIST" + item = "ITEM" + + +class ActionType(StrEnum): + """A change during a play: playback stopped (paused) or started (resumed).""" + + playback_start = "PLAYBACK_START" + playback_stop = "PLAYBACK_STOP" + + +@dataclass +class PlaybackAction: + """A pause, resume or seek during a play. + + A pause is a ``PLAYBACK_STOP``, resuming is a ``PLAYBACK_START``. A seek is a + ``PLAYBACK_STOP`` at the old position followed by a ``PLAYBACK_START`` at the new + one. + + :param timestamp: When it happened, in milliseconds since the epoch + :param position: The position in the track, in seconds + :param action_type: :class:`ActionType` + """ + + timestamp: int + position: float + action_type: ActionType + + def to_json(self) -> JsonObj: + return { + "timestamp": self.timestamp, + "assetPosition": self.position, + "actionType": str(self.action_type), + } + + +@dataclass +class PlaybackSession: + """One finished play of a track, as measured by a :class:`PlaybackTracker`. + + Keep it with :meth:`to_dict` and :meth:`from_dict` to report it later, for example + after playing offline. It goes through the same checks when it's reported. + """ + + track_id: int | str + track_duration: int | None + start_timestamp: int + end_timestamp: int + start_position: float + end_position: float + source_type: SourceType = SourceType.item + source_id: int | str | None = None + actions: list[PlaybackAction] = field(default_factory=list) + quality: str = media.Quality.high_lossless + audio_mode: str = media.AudioMode.stereo + playback_session_id: str = field(default_factory=lambda: str(uuid.uuid4())) + event_id: str = field(default_factory=lambda: str(uuid.uuid4())) + + def to_json(self) -> JsonObj: + """The event payload, as TIDAL's players send it.""" + source_id = self.source_id if self.source_id is not None else self.track_id + return { + "playbackSessionId": self.playback_session_id, + "startTimestamp": self.start_timestamp, + "startAssetPosition": self.start_position, + "isPostPaywall": True, + "productType": "TRACK", + "requestedProductId": str(self.track_id), + "actualProductId": str(self.track_id), + "actualAssetPresentation": "FULL", + "actualAudioMode": str(self.audio_mode), + "actualQuality": str(self.quality), + "sourceType": str(self.source_type), + "sourceId": str(source_id), + "actions": [action.to_json() for action in self.actions], + "endTimestamp": self.end_timestamp, + "endAssetPosition": self.end_position, + } + + def to_dict(self) -> dict[str, Any]: + """The play as JSON-compatible data, for :meth:`from_dict`.""" + data = asdict(self) + data["source_type"] = str(self.source_type) + data["quality"] = str(self.quality) + data["audio_mode"] = str(self.audio_mode) + for action in data["actions"]: + action["action_type"] = str(action["action_type"]) + return data + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> PlaybackSession: + """A play kept with :meth:`to_dict`.""" + data = dict(data) + data["source_type"] = SourceType(data["source_type"]) + data["actions"] = [ + PlaybackAction( + action["timestamp"], + action["position"], + ActionType(action["action_type"]), + ) + for action in data["actions"] + ] + return cls(**data) + + def get_problem(self, now: int) -> str | None: + """Why this play can't have happened, if so. + + :param now: The current time, in milliseconds since the epoch + :return: A description of the problem, or None if the play is plausible + """ + if self.end_timestamp < self.start_timestamp: + return "it ends before it starts" + if self.end_timestamp > now + CLOCK_TOLERANCE_MS: + return "it ends in the future" + if now - self.end_timestamp > MAX_PLAY_AGE_MS: + return "it is too old" + + positions = [self.start_position, self.end_position] + positions += [action.position for action in self.actions] + if min(positions) < 0: + return "a position is negative" + if ( + self.track_duration + and max(positions) > self.track_duration + POSITION_TOLERANCE_SECONDS + ): + return "a position is past the end of the track" + + return self._get_timeline_problem() + + def _get_timeline_problem(self) -> str | None: + last_timestamp = self.start_timestamp + playing_from = (self.start_timestamp, self.start_position) + for action in self.actions: + if action.timestamp < last_timestamp - CLOCK_TOLERANCE_MS: + return "its pauses and seeks are out of order" + if action.timestamp > self.end_timestamp + CLOCK_TOLERANCE_MS: + return "a pause or seek comes after its end" + last_timestamp = action.timestamp + + if action.action_type == ActionType.playback_start: + playing_from = (action.timestamp, action.position) + elif playing_from is not None: + if _advances_too_fast(playing_from, action.timestamp, action.position): + return "more was listened to than time passed" + playing_from = None + + if playing_from is not None and _advances_too_fast( + playing_from, self.end_timestamp, self.end_position + ): + return "more was listened to than time passed" + return None + + +def _advances_too_fast( + playing_from: tuple[int, float], timestamp: int, position: float +) -> bool: + start_timestamp, start_position = playing_from + elapsed = (timestamp - start_timestamp) / 1000 + return position - start_position > elapsed + POSITION_TOLERANCE_SECONDS + + +class PlaybackTracker: + """Measures one play of a track, see :meth:`.Session.start_playback`. + + Tell it when the player pauses (also while it's buffering), resumes or seeks, and + stop it when the track stops playing. + """ + + def __init__( + self, + track: media.Track, + source_type: SourceType, + source_id: int | str, + position: float, + paused: bool, + quality: str, + audio_mode: str, + ): + self.track = track + self.source_type = source_type + self.source_id = source_id + self.quality = quality + self.audio_mode = audio_mode + + self.duration = track.duration if track.duration and track.duration > 0 else 0 + self.actions: list[PlaybackAction] = [] + self.start_timestamp: int | None = None + self.start_position = 0.0 + self.listened_seconds = 0.0 + self.play: PlaybackSession | None = None + self.stopped = False + + self._position = self._limit_position(position) + self._resumed_at: float | None = None + if not paused: + self.resume() + + @property + def position(self) -> float: + """The position in the track, in seconds.""" + self._update() + return self._position + + def pause(self) -> None: + """The player paused, or stopped to buffer.""" + if self.stopped or self._resumed_at is None: + return + self._update() + self._resumed_at = None + self._add_action(ActionType.playback_stop, self._position) + + def resume(self) -> None: + """The player started playing. + + A play starts when it first does. + """ + if self.stopped or self._resumed_at is not None: + return + if self.start_timestamp is None: + self.start_timestamp = _now() + self.start_position = self._position + else: + self._add_action(ActionType.playback_start, self._position) + self._resumed_at = time.monotonic() + + def seek(self, position: float) -> None: + """The player moved to another position. + + :param position: The new position in the track, in seconds + """ + if self.stopped: + return + self._update() + position = self._limit_position(position) + if self._resumed_at is not None: + self._add_action(ActionType.playback_stop, self._position) + self._add_action(ActionType.playback_start, position) + self._position = position + + def stop(self) -> PlaybackSession | None: + """The track stopped playing. + + :return: The finished play, to pass to :meth:`.Session.report_playback`, or None + if too little was listened to for TIDAL to count it + """ + if self.stopped: + return self.play + self._update() + self._resumed_at = None + self.stopped = True + + if self.start_timestamp is None or not self._counts(): + return None + self.play = PlaybackSession( + track_id=self.track.id, + track_duration=self.duration or None, + start_timestamp=self.start_timestamp, + end_timestamp=_now(), + start_position=self.start_position, + end_position=self._position, + source_type=self.source_type, + source_id=self.source_id, + actions=self.actions, + quality=self.quality, + audio_mode=self.audio_mode, + ) + return self.play + + def _counts(self) -> bool: + needed = float(MIN_LISTENED_SECONDS) + if self.duration: + needed = min(needed, self.duration - 1) + return self.listened_seconds >= needed + + def _update(self) -> None: + if self._resumed_at is None: + return + now = time.monotonic() + elapsed = now - self._resumed_at + self._resumed_at = now + self.listened_seconds += elapsed + self._position = self._limit_position(self._position + elapsed) + + def _limit_position(self, position: float) -> float: + position = max(0.0, position) + return min(position, self.duration) if self.duration else position + + def _add_action(self, action_type: ActionType, position: float) -> None: + self.actions.append(PlaybackAction(_now(), position, action_type)) + + +@dataclass +class PlaybackReport: + """What happened to the plays given to :meth:`.Session.report_playback`. + + :param accepted: Plays TIDAL took. Don't report them again + :param rejected: Plays that were refused for good, by the library or by TIDAL. They + would be refused again + :param unsent: Plays that couldn't be sent or weren't confirmed. Keep them and try + again later + :param error: What stopped the sending, if an exception did + """ + + accepted: list[PlaybackSession] = field(default_factory=list) + rejected: list[PlaybackSession] = field(default_factory=list) + unsent: list[PlaybackSession] = field(default_factory=list) + error: Exception | None = None + + +def get_token_claims(access_token: str) -> JsonObj: + """The claims of an access token, without verifying it. + + :param access_token: A TIDAL access token (a JWT) + :return: The claims, or an empty dict if the token can't be decoded + """ + try: + payload = access_token.split(".")[1] + payload += "=" * (-len(payload) % 4) + return json.loads(base64.urlsafe_b64decode(payload)) + except (IndexError, ValueError): + return {} + + +def _now() -> int: + return int(time.time() * 1000) + + +def _get_source(source: Any, track: media.Track) -> tuple[SourceType, int | str]: + if isinstance(source, Album): + return SourceType.album, source.id + if isinstance(source, Playlist): + return SourceType.playlist, source.id + if isinstance(source, (Mix, MixV2)): + return SourceType.mix, source.id + if isinstance(source, Artist): + return SourceType.artist, source.id + return SourceType.item, track.id + + +class PlayLog: + """Measures plays and reports them to TIDAL, see :meth:`.Session.start_playback` and + :meth:`.Session.report_playback`.""" + + def __init__(self, session: Session): + self.session = session + # The quality and audio mode of the streams fetched with this session, by track + self._streams: dict[int | str, tuple[str, str]] = {} + # When the plays this session reported happened, by id + self._reported: dict[str, tuple[int, int]] = {} + + def note_stream(self, track_id: int | str, quality: str, audio_mode: str) -> None: + """Remember that this session fetched a stream for a track.""" + self._streams[track_id] = (quality, audio_mode) + + def start( + self, + track: media.Track, + source: Any, + position: float, + paused: bool, + local_copy: bool, + ) -> PlaybackTracker: + """See :meth:`.Session.start_playback`.""" + stream = self._streams.get(track.id) + if stream is None and not local_copy: + raise ValueError( + f"This session didn't fetch a stream for track {track.id}, and it's " + "not playing from a local copy" + ) + quality, audio_mode = stream or ( + media.Quality.high_lossless, + media.AudioMode.stereo, + ) + source_type, source_id = _get_source(source, track) + return PlaybackTracker( + track, source_type, source_id, position, paused, quality, audio_mode + ) + + def report( + self, plays: Iterable[PlaybackSession], app_version: str + ) -> PlaybackReport: + """See :meth:`.Session.report_playback`.""" + report = PlaybackReport() + now = _now() + to_send: list[PlaybackSession] = [] + for play in plays: + problem = self._get_problem(play, to_send, now) + if problem: + log.warning("Not reporting play of %s: %s", play.track_id, problem) + report.rejected.append(play) + else: + to_send.append(play) + + for start in range(0, len(to_send), MAX_BATCH_SIZE): + batch = to_send[start : start + MAX_BATCH_SIZE] + # Like TIDAL's SDKs, stop at the first batch that doesn't go through + if report.unsent: + report.unsent += batch + continue + try: + accepted, rejected, unsent = self._post(batch, app_version) + except (requests.RequestException, TidalAPIError) as e: + log.info("Couldn't report plays to TIDAL: %s", e) + report.error = e + report.unsent += batch + continue + for play in accepted: + self._reported[play.playback_session_id] = ( + play.start_timestamp, + play.end_timestamp, + ) + report.accepted += accepted + report.rejected += rejected + report.unsent += unsent + return report + + def _get_problem( + self, play: PlaybackSession, earlier: list[PlaybackSession], now: int + ) -> str | None: + problem = play.get_problem(now) + if problem: + return problem + earlier_ids = {other.playback_session_id for other in earlier} + if ( + play.playback_session_id in self._reported + or play.playback_session_id in earlier_ids + ): + return "it was reported already" + + times = list(self._reported.values()) + times += [(other.start_timestamp, other.end_timestamp) for other in earlier] + for start, end in times: + if ( + play.start_timestamp < end - CLOCK_TOLERANCE_MS + and start < play.end_timestamp - CLOCK_TOLERANCE_MS + ): + return "it overlaps another play" + return None + + def _post( + self, plays: list[PlaybackSession], app_version: str + ) -> tuple[list[PlaybackSession], list[PlaybackSession], list[PlaybackSession]]: + """Send one batch. + + :return: The plays TIDAL accepted, rejected, and those to send again + :raises: :class:`requests.RequestException` or a :class:`.TidalAPIError` if the + batch couldn't be sent, which TIDAL's SDKs retry whatever the error + """ + requests_ = self.session.request + if requests_._access_token_expired(): + try: + requests_._refresh_access_token(self.session.access_token) + except AuthenticationError as e: + log.warning("Access token refresh failed: %s", e) + + sent_access_token = self.session.access_token + response = self._send(plays, app_version) + if response.status_code == 401 and requests_._refresh_access_token( + sent_access_token + ): + response = self._send(plays, app_version) + + try: + response.raise_for_status() + except requests.HTTPError as e: + if err := http_error_to_tidal_error(e): + raise err from e + raise + return self._read_reply(plays, response.text) + + def _read_reply( + self, plays: list[PlaybackSession], reply: str + ) -> tuple[list[PlaybackSession], list[PlaybackSession], list[PlaybackSession]]: + """Sort a batch by the reply. + + As in TIDAL's SDKs a play is only done when the reply lists it, as sent or as + malformed (``SenderFault``). + + Anything else answering with a success, like a Wi-Fi login page, confirms + nothing. + """ + accepted_ids: set[str | None] = set() + rejected_ids: set[str | None] = set() + try: + root = ElementTree.fromstring(reply) + except ElementTree.ParseError: + log.info("Couldn't read TIDAL's reply: %s", reply[:500]) + else: + for entry in root.iterfind(".//{*}SendMessageBatchResultEntry"): + accepted_ids.add(entry.findtext("{*}Id")) + for entry in root.iterfind(".//{*}BatchResultErrorEntry"): + if entry.findtext("{*}SenderFault") == "true": + log.warning("TIDAL rejected a reported play: %s", reply[:500]) + rejected_ids.add(entry.findtext("{*}Id")) + + accepted = [play for play in plays if play.event_id in accepted_ids] + rejected = [play for play in plays if play.event_id in rejected_ids] + done = accepted_ids | rejected_ids + return accepted, rejected, [p for p in plays if p.event_id not in done] + + def _send( + self, plays: list[PlaybackSession], app_version: str + ) -> requests.Response: + form = self.get_form(plays, app_version) + return self.session.request._send( + "POST", self.session.config.event_producer_location, {}, form, {} + ) + + def get_form( + self, plays: list[PlaybackSession], app_version: str + ) -> dict[str, str]: + """The request body for a batch of plays: an SQS ``SendMessageBatch`` form, with + one ``playback_session`` event per play. + + :param plays: At most :data:`MAX_BATCH_SIZE` plays + :param app_version: The version of the application that played them + :return: The form fields + """ + access_token = self.session.access_token or "" + claims = get_token_claims(access_token) + os_name = platform.system() + client_id = ( + self.session.config.client_id_pkce + if self.session.is_pkce + else self.session.config.client_id + ) + + sent_timestamp = str(_now()) + + form = {} + for index, play in enumerate(plays, 1): + body = { + "group": "play_log", + "version": 2, + "ts": play.end_timestamp, + "uuid": play.event_id, + "user": { + "id": claims.get("uid"), + "clientId": claims.get("cid"), + "sessionId": claims.get("sid"), + }, + "client": { + "token": str(claims.get("cid", "")), + "deviceType": "desktop", + "version": app_version, + "platform": os_name.lower(), + }, + "payload": play.to_json(), + } + headers = { + "app-version": app_version, + "os-name": os_name, + "os-version": platform.release(), + "consent-category": "NECESSARY", + "requested-sent-timestamp": sent_timestamp, + "authorization": access_token, + "client-id": client_id, + } + entry = f"SendMessageBatchRequestEntry.{index}." + form[entry + "Id"] = play.event_id + form[entry + "MessageBody"] = json.dumps(body) + form[entry + "MessageAttribute.1.Name"] = "Name" + form[entry + "MessageAttribute.1.Value.StringValue"] = "playback_session" + form[entry + "MessageAttribute.1.Value.DataType"] = "String" + form[entry + "MessageAttribute.2.Name"] = "Headers" + form[entry + "MessageAttribute.2.Value.StringValue"] = json.dumps(headers) + form[entry + "MessageAttribute.2.Value.DataType"] = "String" + return form diff --git a/tidalapi/session.py b/tidalapi/session.py index 71050fb..bcc6b22 100644 --- a/tidalapi/session.py +++ b/tidalapi/session.py @@ -27,7 +27,7 @@ import random import time import uuid -from collections.abc import Callable +from collections.abc import Callable, Iterable from dataclasses import dataclass from enum import Enum from pathlib import Path @@ -52,7 +52,7 @@ ) from tidalapi.types import JsonObj -from . import album, artist, genre, media, mix, page, playlist, request, user +from . import album, artist, genre, media, mix, page, playlist, playlog, request, user if TYPE_CHECKING: from tidalapi.user import FetchedUser, LoggedInUser, PlaylistCreator @@ -112,6 +112,8 @@ class Config: api_v1_location: str = "https://api.tidal.com/v1/" api_v2_location: str = "https://api.tidal.com/v2/" openapi_v2_location: str = "https://openapi.tidal.com/v2/" + #: Where plays are reported, see :meth:`Session.report_playback` + event_producer_location: str = "https://ec.tidal.com/api/event-batch" client_id: str client_secret: str image_url: str = "https://resources.tidal.com/images/%s/%ix%i.jpg" @@ -243,6 +245,7 @@ def __init__(self, config: Config | None = None): # Objects for keeping the session across all modules. self.request = request.Requests(session=self) + self.playlog = playlog.PlayLog(session=self) self.genre = genre.Genre(session=self) self.parse_user = user.User(self, None).parse @@ -896,6 +899,69 @@ def search( return result + def start_playback( + self, + track: media.Track, + source: Any = None, + position: float = 0.0, + paused: bool = False, + local_copy: bool = False, + ) -> playlog.PlaybackTracker: + """Start measuring a play of a track, to report it to TIDAL afterwards. + + Example:: + + stream = track.get_stream() + tracker = session.start_playback(track, source=album) + # When the player does: tracker.pause(), tracker.resume(), tracker.seek(90) + play = tracker.stop() + if play: + report = session.report_playback([play], app_version="My Player 1.0") + + :param track: The track that started playing. This session must have fetched a + stream for it + :param source: The :class:`.Album`, :class:`.Playlist`, :class:`.Mix` or + :class:`.Artist` it's played from, if any. Recently played lists the source + :param position: The position it starts from, in seconds + :param paused: The player isn't playing yet. The play starts at the first + :meth:`~tidalapi.playlog.PlaybackTracker.resume` + :param local_copy: It plays from a copy kept earlier (e.g. offline), so this + session may not have fetched a stream for it + :return: A :class:`~tidalapi.playlog.PlaybackTracker` + :raises: :class:`ValueError` if this session didn't fetch a stream for the track + """ + return self.playlog.start(track, source, position, paused, local_copy) + + def report_playback( + self, + plays: Iterable[playlog.PlaybackSession], + app_version: str | None = None, + ) -> playlog.PlaybackReport: + """Report finished plays to TIDAL, so they show up in Recently played on all + devices and count for recommendations. + + Fetching and playing a stream doesn't count as a play on its own. Measure each + play with :meth:`start_playback` and report it once it ends. + + Plays that can't have happened are refused: ones that end before they start, in + the future or too long ago, with a position past the end of the track, with more + listening than time passed, overlapping another play of this session, or that + this session reported already. + + :param plays: The plays to report, see + :class:`~tidalapi.playlog.PlaybackSession` + :param app_version: The name and version of the application that played them. + Defaults to the tidalapi version + :return: A :class:`~tidalapi.playlog.PlaybackReport` saying which plays were + accepted, which were rejected for good, and which to try again later. Being + offline or failing to reach TIDAL doesn't raise, the plays are left unsent + """ + if app_version is None: + from tidalapi import __version__ + + app_version = f"tidalapi {__version__}" + return self.playlog.report(plays, app_version) + def check_login(self) -> bool: """Returns true if current session is valid, false otherwise.""" if self.user is None or not self.user.id or not self.session_id: