diff --git a/tests/test_mix.py b/tests/test_mix.py index 8ce6a61..8e04831 100644 --- a/tests/test_mix.py +++ b/tests/test_mix.py @@ -45,11 +45,9 @@ def test_mixv2_unavailable(session): mix = session.mixv2("12345678") -@pytest.mark.skip(reason="Cannot test against user specific mixes") def test_mix_available(session): - mix = session.mix("016edb91bc504e618de6918b11b25b") + mix = session.mix("001cb879e927219fc3322cb03aed01") -@pytest.mark.skip(reason="Cannot test against user specific mixes") def test_mixv2_available(session): - mix = session.mixv2("016edb91bc504e618de6918b11b25b") + mix = session.mixv2("001cb879e927219fc3322cb03aed01") diff --git a/tidalapi/mix.py b/tidalapi/mix.py index 3d52186..c6718c3 100644 --- a/tidalapi/mix.py +++ b/tidalapi/mix.py @@ -176,16 +176,22 @@ class TextInfo: class MixV2: """A mix from TIDALs v2 api endpoint.""" + mix_type: Optional[MixType] = None + country_code: Optional[str] = None date_added: Optional[datetime] = None - title: Optional[str] = None id: Optional[str] = None - mix_type: Optional[MixType] = None + artifact_id_type: Optional[str] = None + content_behavior: Optional[str] = None images: Optional[ImageResponse] = None detail_images: Optional[ImageResponse] = None master = False + is_stable_id = False + title: Optional[str] = None + sub_title: Optional[str] = None + short_subtitle: Optional[str] = None title_text_info: Optional[TextInfo] = None sub_title_text_info: Optional[TextInfo] = None - sub_title: Optional[str] = None + short_subtitle_text_info: Optional[TextInfo] = None updated: Optional[datetime] = None _retrieved = False _items: Optional[List[Union["Video", "Track"]]] = None @@ -232,38 +238,84 @@ def parse(self, json_obj: JsonObj) -> "MixV2": :param json_obj: The json of a mix to be parsed :return: A copy of the parsed mix """ - date_added = json_obj.get("dateAdded") - self.date_added = dateutil.parser.isoparse(date_added) if date_added else None - self.title = json_obj["title"] + self.id = json_obj["id"] - self.title = json_obj["title"] - self.mix_type = MixType(json_obj["mixType"]) - images = json_obj["images"] - self.images = ImageResponse( - small=images["SMALL"]["url"], - medium=images["MEDIUM"]["url"], - large=images["LARGE"]["url"], - ) - detail_images = json_obj["detailImages"] - self.detail_images = ImageResponse( - small=detail_images["SMALL"]["url"], - medium=detail_images["MEDIUM"]["url"], - large=detail_images["LARGE"]["url"], - ) - self.master = json_obj["master"] - title_text_info = json_obj["titleTextInfo"] - self.title_text_info = TextInfo( - text=title_text_info["text"], - color=title_text_info["color"], - ) - sub_title_text_info = json_obj["subTitleTextInfo"] - self.sub_title_text_info = TextInfo( - text=sub_title_text_info["text"], - color=sub_title_text_info["color"], - ) - self.sub_title = json_obj["subTitle"] - updated = json_obj.get("updated") - self.date_added = dateutil.parser.isoparse(updated) if date_added else None + if json_obj.get("mixType"): + date_added = json_obj.get("dateAdded") + self.date_added = ( + dateutil.parser.isoparse(date_added) if date_added else None + ) + self.title = json_obj["title"] + self.sub_title = json_obj["subTitle"] + images = json_obj["images"] + self.images = ImageResponse( + small=images["SMALL"]["url"], + medium=images["MEDIUM"]["url"], + large=images["LARGE"]["url"], + ) + detail_images = json_obj["detailImages"] + self.detail_images = ImageResponse( + small=detail_images["SMALL"]["url"], + medium=detail_images["MEDIUM"]["url"], + large=detail_images["LARGE"]["url"], + ) + self.master = json_obj["master"] + title_text_info = json_obj["titleTextInfo"] + self.title_text_info = TextInfo( + text=title_text_info["text"], + color=title_text_info["color"], + ) + sub_title_text_info = json_obj["subTitleTextInfo"] + self.sub_title_text_info = TextInfo( + text=sub_title_text_info["text"], + color=sub_title_text_info["color"], + ) + updated = json_obj.get("updated") + self.updated = dateutil.parser.isoparse(updated) if updated else None + elif json_obj.get("type"): + # Certain mix types (e.g. when returned from Page) must be parsed differently. Why, TIDAL? + self.country_code = json_obj.get("countryCode", None) + self.is_stable_id = json_obj.get("isStableId", False) + self.artifact_id_type = json_obj.get("trackGroupId", None) + self.content_behavior = json_obj.get("contentBehavior", None) + + images = json_obj["mixImages"] + self.images = ImageResponse( + small=images[0]["url"], + medium=images[1]["url"], + large=images[0]["url"], + ) + + detail_images = json_obj["detailMixImages"] + self.detail_images = ImageResponse( + small=detail_images[0]["url"], + medium=detail_images[1]["url"], + large=detail_images[2]["url"], + ) + + title_text_info = json_obj["titleTextInfo"] + self.title_text_info = TextInfo( + text=title_text_info["text"], + color=title_text_info["color"], + ) + self.title = title_text_info["text"] + + sub_title_text_info = json_obj["subtitleTextInfo"] + self.sub_title_text_info = TextInfo( + text=sub_title_text_info["text"], + color=sub_title_text_info["color"], + ) + self.sub_title = sub_title_text_info["text"] + + short_subtitle_text_info = json_obj["shortSubtitleTextInfo"] + self.short_subtitle_text_info = TextInfo( + text=sub_title_text_info["text"], + color=sub_title_text_info["color"], + ) + self.short_subtitle = short_subtitle_text_info["text"] + + if json_obj.get("updated"): + self.updated = datetime.fromtimestamp(json_obj["updated"] / 1000) return copy.copy(self) diff --git a/tidalapi/page.py b/tidalapi/page.py index 3696cf4..a5c3bc9 100644 --- a/tidalapi/page.py +++ b/tidalapi/page.py @@ -42,6 +42,7 @@ from tidalapi.request import Requests from tidalapi.session import Session + PageCategories = Union[ "Album", "PageLinks", @@ -54,6 +55,15 @@ AllCategories = Union["Artist", PageCategories] +PageCategoriesV2 = Union[ + "TrackList", + "ShortcutList", + "HorizontalList", + "HorizontalListWithContext", +] + +AllCategoriesV2 = Union[PageCategoriesV2] + class Page: """ @@ -64,10 +74,13 @@ class Page: """ title: str = "" - categories: Optional[List["AllCategories"]] = None - _categories_iter: Optional[Iterator["AllCategories"]] = None + categories: Optional[List[Union["AllCategories", "AllCategoriesV2"]]] = None + _categories_iter: Optional[Iterator[Union["AllCategories", "AllCategoriesV2"]]] = ( + None + ) _items_iter: Optional[Iterator[Callable[..., Any]]] = None page_category: "PageCategory" + page_category_v2: "PageCategoryV2" request: "Requests" def __init__(self, session: "Session", title: str): @@ -75,6 +88,7 @@ def __init__(self, session: "Session", title: str): self.categories = None self.title = title self.page_category = PageCategory(session) + self.page_category_v2 = PageCategoryV2(session) def __iter__(self) -> "Page": if self.categories is None: @@ -106,11 +120,17 @@ def parse(self, json_obj: JsonObj) -> "Page": """Goes through everything in the page, and gets the title and adds all the rows to the categories field :param json_obj: The json to be parsed :return: A copy of the Page that you can use to browse all the items.""" - self.title = json_obj["title"] self.categories = [] - for row in json_obj["rows"]: - page_item = self.page_category.parse(row["modules"][0]) - self.categories.append(page_item) + + if json_obj.get("rows"): + self.title = json_obj["title"] + for row in json_obj["rows"]: + page_item = self.page_category.parse(row["modules"][0]) + self.categories.append(page_item) + else: + for item in json_obj["items"]: + page_item = self.page_category_v2.parse_item(item) + self.categories.append(page_item) return copy.copy(self) @@ -140,10 +160,13 @@ class More: @classmethod def parse(cls, json_obj: JsonObj) -> Optional["More"]: show_more = json_obj.get("showMore") - if show_more is None: - return None - else: + view_all = json_obj.get("viewAll") + if show_more is not None: return cls(api_path=show_more["apiPath"], title=show_more["title"]) + elif view_all is not None: + return cls(api_path=view_all, title=json_obj.get("title")) + else: + return None class PageCategory: @@ -215,6 +238,147 @@ def show_more(self) -> Optional[Page]: ) +class PageCategoryV2: + """Base class for all V2 homepage page categories (e.g., TRACK_LIST, SHORTCUT_LIST). + + Handles shared fields and parsing logic, and automatically dispatches to the correct + subclass based on the 'type' field in the JSON object. + """ + + # Registry mapping 'type' strings to subclass types + _type_map: Dict[str, Type["PageCategoryV2"]] = {} + + # Common metadata fields for all category types + type: Optional[str] = None + module_id: Optional[str] = None + title: Optional[str] = None + subtitle: Optional[str] = None + description: Optional[str] = "" + _more: Optional["More"] = None + + def __init__(self, session: "Session"): + """Store the shared session object and initialize common fields. + + Subclasses should implement their own `parse()` method but not override + __init__. + """ + self.session = session + self.request = session.request + + # Common item parsers by type (can be used by subclasses like SimpleList) + self.item_types: Dict[str, Callable[..., Any]] = { + "PLAYLIST": self.session.parse_playlist, + "VIDEO": self.session.parse_video, + "TRACK": self.session.parse_track, + "ARTIST": self.session.parse_artist, + "ALBUM": self.session.parse_album, + "MIX": self.session.parse_v2_mix, + } + + @classmethod + def register_subclass(cls, category_type: str): + """Decorator to register subclasses in the _type_map. + + Usage: + @PageCategoryV2.register_subclass("TRACK_LIST") + class TrackList(PageCategoryV2): + ... + """ + + def decorator(subclass): + cls._type_map[category_type] = subclass + subclass.category_type = category_type + return subclass + + return decorator + + def parse_item(self, list_item: Dict) -> "PageCategoryV2": + """Factory method that creates the correct subclass instance based on the 'type' + field in item Dict, parses base fields, and then calls subclass parse().""" + category_type = list_item.get("type") + cls = self._type_map.get(category_type) + if cls is None: + raise NotImplementedError(f"Category {category_type} not implemented") + instance = cls(self.session) + instance._parse_base(list_item) + instance.parse(list_item) + return instance + + def _parse_base(self, list_item: Dict): + """Parse fields common to all categories.""" + self.type = list_item.get("type") + self.module_id = list_item.get("moduleId") + self.title = list_item.get("title") + self.subtitle = list_item.get("subtitle") + self.description = list_item.get("description") + self._more = More.parse(list_item) + + def parse(self, json_obj: JsonObj): + """Subclasses implement this method to parse category-specific data.""" + raise NotImplementedError("Subclasses must implement parse()") + + def view_all(self) -> Optional[Page]: + """View all items in a Get the full list of items on their own :class:`.Page` + from a :class:`.PageCategory` + + :return: A :class:`.Page` more of the items in the category, None if there aren't any + """ + api_path = self._more.api_path if self._more else None + return self.session.view_all(api_path) if api_path and self._more else None + + +class SimpleList(PageCategoryV2): + """A generic list of items (tracks, albums, playlists, etc.) using the shared + self.item_types parser dictionary.""" + + def __init__(self, session: "Session"): + super().__init__(session) + self.items: List[Any] = [] + + def parse(self, json_obj: "JsonObj"): + self.items = [self.get_item(item) for item in json_obj["items"]] + return self + + def get_item(self, json_obj: "JsonObj") -> Any: + item_type = json_obj.get("type") + if item_type not in self.item_types: + raise NotImplementedError(f"Item type '{item_type}' not implemented") + + return self.item_types[item_type](json_obj["data"]) + + +@PageCategoryV2.register_subclass("SHORTCUT_LIST") +class ShortcutList(SimpleList): + """A list of "shortcut" links (typically small horizontally scrollable rows).""" + + +@PageCategoryV2.register_subclass("HORIZONTAL_LIST") +class HorizontalList(SimpleList): + """A horizontal scrollable row of items.""" + + +@PageCategoryV2.register_subclass("HORIZONTAL_LIST_WITH_CONTEXT") +class HorizontalListWithContext(HorizontalList): + """A horizontal list of items with additional context.""" + + +@PageCategoryV2.register_subclass("TRACK_LIST") +class TrackList(PageCategoryV2): + """A category that represents a list of tracks, each one parsed with + parse_track().""" + + def __init__(self, session: "Session"): + super().__init__(session) + self.items: List[Any] = [] + + def parse(self, json_obj: "JsonObj"): + self.items = [ + self.session.parse_track(item["data"]) for item in json_obj["items"] + ] + + return self + + class FeaturedItems(PageCategory): """Items that have been featured by TIDAL.""" @@ -347,7 +511,9 @@ def __init__(self, session: "Session", json_obj: JsonObj): self.text = json_obj["text"] self.featured = bool(json_obj["featured"]) - def get(self) -> Union["Artist", "Playlist", "Track", "UserPlaylist", "Video"]: + def get( + self, + ) -> Union["Artist", "Playlist", "Track", "UserPlaylist", "Video", "Album"]: """Retrieve the PageItem with the artifact_id matching the type. :return: The fully parsed item, e.g. :class:`.Playlist`, :class:`.Video`, :class:`.Track` diff --git a/tidalapi/request.py b/tidalapi/request.py index e6bee54..d584830 100644 --- a/tidalapi/request.py +++ b/tidalapi/request.py @@ -57,6 +57,7 @@ class Requests(object): def __init__(self, session: "Session"): # More Android User-Agents here: https://user-agents.net/browsers/android self.user_agent = "Mozilla/5.0 (Linux; Android 12; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/91.0.4472.114 Safari/537.36" + self.client_version = "2025.7.16" self.session = session self.config = session.config self.latest_err_response = requests.Response() @@ -85,6 +86,9 @@ def basic_request( if not headers: headers = {} + if "x-tidal-client-version" not in headers: + headers["x-tidal-client-version"] = self.client_version + if "User-Agent" not in headers: headers["User-Agent"] = self.user_agent @@ -170,18 +174,22 @@ def request( return request def get_latest_err_response(self) -> dict: - """Get the latest request Response that resulted in an Exception :return: The - request Response that resulted in the Exception, returned as a dict An empty - dict will be returned, if no response was returned.""" + """Get the latest request Response that resulted in an Exception. + + :return: The request Response that resulted in the Exception, returned as a dict + An empty dict will be returned, if no response was returned. + """ if self.latest_err_response.content: return self.latest_err_response.json() else: return {} def get_latest_err_response_str(self) -> str: - """Get the latest request response message as a string :return: The contents of - the (detailed) error response Response, returned as a string An empty str will - be returned, if no response was returned.""" + """Get the latest request response message as a string. + + :return: The contents of the (detailed) error response, returned as a string An + empty str will be returned, if no response was returned. + """ if self.latest_err_response.content: resp = self.latest_err_response.json() return resp["errors"][0]["detail"] diff --git a/tidalapi/session.py b/tidalapi/session.py index 910b110..9ee5401 100644 --- a/tidalapi/session.py +++ b/tidalapi/session.py @@ -268,9 +268,10 @@ class Session: refresh_token: Optional[str] = None #: The type of access token, e.g. Bearer token_type: Optional[str] = None - #: The id for a TIDAL session, you also need this to use load_oauth_session + #: The session id for a TIDAL session, you also need this to use load_oauth_session session_id: Optional[str] = None country_code: Optional[str] = None + locale: Optional[str] = None #: A :class:`.User` object containing the currently logged in user. user: Optional[Union["FetchedUser", "LoggedInUser", "PlaylistCreator"]] = None @@ -282,15 +283,6 @@ def __init__(self, config: Config = Config()): self.request = request.Requests(session=self) self.genre = genre.Genre(session=self) - # self.parse_artists = self.artist().parse_artists - # self.parse_playlist = self.playlist().parse - - # self.parse_track = self.track().parse_track - # self.parse_video = self.video().parse_video - # self.parse_media = self.track().parse_media - # self.parse_mix = self.mix().parse - # self.parse_v2_mix = self.mixv2().parse - self.parse_user = user.User(self, None).parse self.page = page.Page(self, "") self.parse_page = self.page.parse @@ -453,6 +445,7 @@ def load_oauth_session( self.session_id = json["sessionId"] self.country_code = json["countryCode"] + self.locale = "en_US" # TODO Get locale from system configuration self.user = user.User(self, user_id=json["userId"]).factory() return True @@ -719,6 +712,7 @@ def process_auth_token( json = session.json() self.session_id = json["sessionId"] self.country_code = json["countryCode"] + self.locale = "en_US" # TODO Set locale from system configuration self.user = user.User(self, user_id=json["userId"]).factory() self.is_pkce = is_pkce_token @@ -1094,7 +1088,15 @@ def home(self) -> page.Page: :return: A :class:`.Page` object with the :class:`.PageCategory` list from the home page """ - return self.page.get("pages/home") + params = {"deviceType": "BROWSER", "locale": self.locale, "platform": "WEB"} + + json_obj = self.request.request( + "GET", + "home/feed/static", + base_url=self.config.api_v2_location, + params=params, + ).json() + return self.page.parseV2(json_obj) def explore(self) -> page.Page: """