Changes to episode retrieval and model to align with the new mobile API endpoint

Author: bendikrb

podme_api/auth/mobile_client.py

@@ -206,5 +206,6 @@ async def refresh_token(self, credentials: SchibstedCredentials | None = None):
 
         return self._credentials
 
+    @property
     def credentials_filename(self):
         return "credentials_mobile.json"

podme_api/cli/cli.py

@@ -4,6 +4,7 @@
 import asyncio
 import contextlib
 import logging
+from typing import AsyncGenerator
 
 from rich.console import Console
 from rich.live import Live
@@ -19,7 +20,7 @@
 from rich.table import Table
 
 from podme_api.__version__ import __version__
-from podme_api.auth import PodMeDefaultAuthClient
+from podme_api.auth import PodMeDefaultAuthClient, PodMeMobileAuthClient
 from podme_api.auth.models import PodMeUserCredentials
 from podme_api.cli.utils import bold_star, is_valid_writable_dir, pretty_dataclass, pretty_dataclass_list
 from podme_api.client import PodMeClient
@@ -119,6 +120,11 @@ def _add_default_arguments(parser: argparse.ArgumentParser):
     parser.add_argument("-V", "--version", action="version", version=f"%(prog)s v{__version__}")
     parser.add_argument("-v", "--verbose", action="count", default=0, help="Logging verbosity level")
     parser.add_argument("--debug", action="store_true", help="Enable debug mode")
+    parser.add_argument(
+        "--legacy",
+        action="store_true",
+        help="Use the old web auth (won't work for much more than retrieving metadata)",
+    )
 
 
 def _add_paging_arguments(parser: argparse.ArgumentParser):
@@ -129,7 +135,7 @@ def _add_paging_arguments(parser: argparse.ArgumentParser):
         default=20,
         help="Number of results to return per page (default=20).",
     )
-    parser.add_argument("--pages", type=int, default=1, help="Maxium number of pages to fetch (default=1)")
+    parser.add_argument("--pages", type=int, default=1, help="Maximum number of pages to fetch (default=1)")
 
 
 async def login(args):
@@ -399,14 +405,17 @@ async def search(args) -> None:
 
 
 @contextlib.asynccontextmanager
-async def _get_client(args) -> PodMeClient:
+async def _get_client(args) -> AsyncGenerator[PodMeClient, None]:
     """Return PodMeClient based on args."""
     if hasattr(args, "username") and hasattr(args, "password"):
         user_creds = PodMeUserCredentials(args.username, args.password)
     else:
         user_creds = None
-    auth_client = PodMeDefaultAuthClient(user_credentials=user_creds)
-    client = PodMeClient(auth_client=auth_client)
+    if args.legacy:
+        auth_client = PodMeDefaultAuthClient(user_credentials=user_creds)
+    else:
+        auth_client = PodMeMobileAuthClient(user_credentials=user_creds)
+    client = PodMeClient(auth_client=auth_client, api_platform="web" if args.legacy else "mobile")
     try:
         await client.__aenter__()
         yield client

podme_api/client.py

@@ -39,12 +39,14 @@
     PodMeApiStreamUrlNotFoundError,
     PodMeApiUnauthorizedError,
 )
+from podme_api.helpers import async_cache
 from podme_api.models import (
     PodMeApiPlatform,
     PodMeCategory,
     PodMeCategoryPage,
     PodMeDownloadProgressTask,
     PodMeEpisode,
+    PodMeEpisodeData,
     PodMeHomeScreen,
     PodMeLanguage,
     PodMePodcast,
@@ -76,7 +78,7 @@ class PodMeClient:
     auth_client: PodMeAuthClient
     """auth_client (PodMeAuthClient): The authentication client."""
 
-    api_platform: PodMeApiPlatform = PodMeApiPlatform.MOBILE
+    api_platform: PodMeApiPlatform = "mobile"
     """api_platform (PodMeApiPlatform): The API platform to use."""
 
     disable_credentials_storage: bool = False
@@ -158,6 +160,7 @@ async def _request(  # noqa: C901
         uri: str,
         method: str = METH_GET,
         retry: int = 0,
+        api_platform: PodMeApiPlatform | None = None,
         **kwargs,
     ) -> str | dict | list | bool | None:
         """Make a request to the PodMe API.
@@ -166,6 +169,7 @@ async def _request(  # noqa: C901
             uri (str): The URI for the API endpoint.
             method (str): The HTTP method to use for the request.
             retry (int): The number of retries for the request.
+            api_platform (PodMeApiPlatform): The API platform to use.
             **kwargs: Additional keyword arguments for the request.
                 May include:
                 - params (dict): Query parameters for the request.
@@ -176,7 +180,9 @@ async def _request(  # noqa: C901
             The response data from the API.
 
         """
-        base_url = PODME_API_URL.format(platform=self.api_platform)
+        if api_platform is None:
+            api_platform = self.api_platform
+        base_url = PODME_API_URL.format(platform=api_platform)
         url = URL(f"{base_url.strip('/')}/").join(URL(uri))
 
         access_token = await self.auth_client.async_get_access_token()
@@ -224,7 +230,7 @@ async def _request(  # noqa: C901
             if response.status == HTTPStatus.TOO_MANY_REQUESTS:
                 raise PodMeApiRateLimitError("Rate limit error has occurred with the PodMe API")
             if response.status == HTTPStatus.NOT_FOUND:
-                raise PodMeApiNotFoundError("Resource not found")
+                raise PodMeApiNotFoundError(f"Resource not found: <{url}>")
             if response.status == HTTPStatus.BAD_REQUEST:
                 raise PodMeApiError("Bad request syntax or unsupported method")
             if response.status == HTTPStatus.UNAUTHORIZED:
@@ -281,6 +287,7 @@ async def _get_pages(
         page_size: int | None = None,
         params: dict | None = None,
         items_key: str | None = None,
+        **kwargs,
     ):
         """Retrieve multiple pages of data from the API.
 
@@ -291,6 +298,7 @@ async def _get_pages(
             page_size: The number of items per page.
             params: Additional parameters for the request.
             items_key: The key for the items in the response.
+            **kwargs: Additional keyword arguments which will be passed on to the request.
 
         Returns:
             list: The retrieved data.
@@ -311,6 +319,7 @@ async def _get_pages(
                         "getByOldest": "true" if get_by_oldest else None,
                         **params,
                     },
+                    **kwargs,
                 )
                 if not isinstance(new_results, list) and items_key is not None:
                     new_results = new_results.get(items_key, [])
@@ -484,11 +493,11 @@ async def download_files(
             on_finished=on_finished,
         )
 
-    async def get_episode_download_url(self, episode: PodMeEpisode | int) -> tuple[int, URL]:
+    async def get_episode_download_url(self, episode: PodMeEpisodeData | int) -> tuple[int, URL]:
         """Get the download URL for an episode.
 
         Args:
-            episode (PodMeEpisode | int): The episode object or episode ID to get the download URL for.
+            episode (PodMeEpisodeData | int): The episode object or episode ID to get the download URL for.
 
         Returns:
             tuple[int, URL]: The episode ID and the download URL.
@@ -500,22 +509,24 @@ async def get_episode_download_url(self, episode: PodMeEpisode | int) -> tuple[i
         episode_data = episode
         if isinstance(episode_data, int):
             episode_data = await self.get_episode_info(episode)
+        if episode_data.url is not None:
+            return episode_data.id, URL(episode_data.url)
         if episode_data.stream_url is None:
             raise PodMeApiStreamUrlError(f"No stream URL found for episode {episode_data.id}")
         info = await self.resolve_stream_url(URL(episode_data.stream_url))
         return episode_data.id, URL(info["url"])
 
     async def get_episode_download_url_bulk(
         self,
-        episodes: list[PodMeEpisode | int],
+        episodes: list[PodMeEpisodeData | int],
     ) -> list[tuple[int, URL]]:
         """Get download URLs for a list of episodes.
 
         This method fetches download URLs for multiple episodes concurrently and
         ensures that only unique episode IDs are included in the result.
 
         Args:
-            episodes (list[PodMeEpisode | int]): A list of PodMeEpisode objects
+            episodes (list[PodMeEpisodeData | int]): A list of PodMeEpisode objects
                 or episode IDs for which to fetch download URLs.
 
         Returns:
@@ -793,6 +804,7 @@ async def get_currently_playing(self) -> list[PodMeEpisode]:
         )
         return [PodMeEpisode.from_dict(data) for data in episodes]
 
+    @async_cache
     async def get_podcast_info(self, podcast_slug: str) -> PodMePodcast:
         """Get information about a podcast.
 
@@ -815,19 +827,19 @@ async def get_podcasts_info(self, podcast_slugs: list[str]) -> list[PodMePodcast
         podcasts = await asyncio.gather(*[self.get_podcast_info(slug) for slug in podcast_slugs])
         return list(podcasts)
 
-    async def get_episode_info(self, episode_id: int) -> PodMeEpisode:
+    async def get_episode_info(self, episode_id: int) -> PodMeEpisodeData:
         """Get information about an episode.
 
         Args:
             episode_id (int): The ID of the episode.
 
         """
         data = await self._request(
-            f"episode/{episode_id}",
+            f"episodes/{episode_id}",
         )
-        return PodMeEpisode.from_dict(data)
+        return PodMeEpisodeData.from_dict(data)
 
-    async def get_episodes_info(self, episode_ids: list[int]) -> list[PodMeEpisode]:
+    async def get_episodes_info(self, episode_ids: list[int]) -> list[PodMeEpisodeData]:
         """Get information about multiple episodes.
 
         Args:
@@ -862,22 +874,25 @@ async def search_podcast(
         )
         return [PodMeSearchResult.from_dict(data) for data in podcasts]
 
-    async def get_episode_list(self, podcast_slug: str) -> list[PodMeEpisode]:
+    async def get_episode_list(self, podcast_slug: str) -> list[PodMeEpisodeData]:
         """Get the full list of episodes for a podcast.
 
         Args:
             podcast_slug (str): The slug of the podcast.
 
         """
+        podcast_info = await self.get_podcast_info(podcast_slug)
         episodes = await self._get_pages(
-            f"episode/slug/{podcast_slug}",
+            f"episodes/podcast/{podcast_info.id}",
             get_by_oldest=True,
         )
         _LOGGER.debug("Retrieved full episode list, containing %s episodes", len(episodes))
 
-        return [PodMeEpisode.from_dict(data) for data in episodes]
+        return [PodMeEpisodeData.from_dict(data) for data in episodes]
 
-    async def get_latest_episodes(self, podcast_slug: str, episodes_limit: int = 20) -> list[PodMeEpisode]:
+    async def get_latest_episodes(
+        self, podcast_slug: str, episodes_limit: int = 20
+    ) -> list[PodMeEpisodeData]:
         """Get the latest episodes for a podcast.
 
         Args:
@@ -888,9 +903,10 @@ async def get_latest_episodes(self, podcast_slug: str, episodes_limit: int = 20)
         max_per_page = 50
         pages = math.ceil(episodes_limit / max_per_page)
         page_size = min(max_per_page, episodes_limit)
+        podcast_info = await self.get_podcast_info(podcast_slug)
 
         episodes = await self._get_pages(
-            f"episode/slug/{podcast_slug}",
+            f"episodes/podcast/{podcast_info.id}",
             get_pages=pages,
             page_size=page_size,
         )
@@ -900,7 +916,7 @@ async def get_latest_episodes(self, podcast_slug: str, episodes_limit: int = 20)
             episodes_limit,
             len(episodes),
         )
-        return [PodMeEpisode.from_dict(data) for data in episodes]
+        return [PodMeEpisodeData.from_dict(data) for data in episodes]
 
     async def get_episode_ids(self, podcast_slug) -> list[int]:
         """Get the IDs of all episodes for a podcast.

podme_api/helpers.py

@@ -0,0 +1,17 @@
+"""podme_api helper functions."""
+
+from functools import wraps
+
+
+def async_cache(func):
+    cache = {}
+
+    @wraps(func)
+    async def wrapper(self, slug):
+        if slug in cache:
+            return cache[slug]
+        result = await func(self, slug)
+        cache[slug] = result
+        return result
+
+    return wrapper

podme_api/models.py

@@ -5,7 +5,7 @@
 from dataclasses import dataclass, field
 from datetime import datetime, time
 from enum import IntEnum, StrEnum, auto
-from typing import TYPE_CHECKING, TypedDict
+from typing import TYPE_CHECKING, Literal, TypedDict
 
 from mashumaro import field_options
 from mashumaro.config import BaseConfig
@@ -16,6 +16,9 @@
     from yarl import URL
 
 
+PodMeApiPlatform = Literal["web", "mobile"]
+
+
 @dataclass
 class BaseDataClassORJSONMixin(DataClassORJSONMixin):
     class Config(BaseConfig):
@@ -46,13 +49,6 @@ def __repr__(self):
         return self.value.lower()
 
 
-class PodMeApiPlatform(StrEnum):
-    """Enumeration of PodMe API platforms."""
-
-    WEB = auto()
-    MOBILE = auto()
-
-
 class PodMeRegion(IntEnum):
     """Enumeration of PodMe regions."""
 
@@ -363,18 +359,18 @@ class PodMeEpisode(PodMeEpisodeBase):
     total_no_of_episodes: int | None = field(default=None, metadata=field_options(alias="totalNoOfEpisodes"))
 
 
-@dataclass
+@dataclass(kw_only=True)
 class PodMeEpisodeData(PodMeEpisode):
     """Represents detailed data for a PodMe episode."""
 
     number: int = field(metadata=field_options(alias="number"))
     byte_length: int = field(metadata=field_options(alias="byteLength"))
     url: str = field(metadata=field_options(alias="url"))
     type: str = field(metadata=field_options(alias="type"))
-    smooth_streaming_url: str = field(metadata=field_options(alias="smoothStreamingUrl"))
-    mpeg_dash_url: str = field(metadata=field_options(alias="mpegDashUrl"))
-    hls_v3_url: str = field(metadata=field_options(alias="hlsV3Url"))
-    hls_v4_url: str = field(metadata=field_options(alias="hlsV4Url"))
+    smooth_streaming_url: str | None = field(default=None, metadata=field_options(alias="smoothStreamingUrl"))
+    mpeg_dash_url: str | None = field(default=None, metadata=field_options(alias="mpegDashUrl"))
+    hls_v3_url: str | None = field(default=None, metadata=field_options(alias="hlsV3Url"))
+    hls_v4_url: str | None = field(default=None, metadata=field_options(alias="hlsV4Url"))
     publish_date: datetime = field(metadata=field_options(alias="publishDate"))
     has_played: bool = field(metadata=field_options(alias="hasPlayed"))
     episode_created_at: datetime = field(metadata=field_options(alias="episodeCreatedAt"))
@@ -403,15 +399,15 @@ class PodMeSubscriptionPlan(BaseDataClassORJSONMixin):
     next_plan_product_id: float | None = field(
         default=None, metadata=field_options(alias="nextPlanProductId")
     )
-    price: int | None = field(default=None)
+    region_id: int | None = field(default=None, metadata=field_options(alias="regionId"))
 
 
 @dataclass
 class PodMeSubscription(BaseDataClassORJSONMixin):
     """Represents a PodMe subscription."""
 
+    subscription_id: int = field(metadata=field_options(alias="subscriptionId"))
     subscription_state: int = field(metadata=field_options(alias="subscriptionState"))
-    subscription_type: int = field(metadata=field_options(alias="subscriptionType"))
     subscription_platform: int = field(metadata=field_options(alias="subscriptionPlatform"))
     expiration_date: datetime = field(
         metadata=field_options(