diff --git a/lib/streaming_providers/base/models/vod.py b/lib/streaming_providers/base/models/vod.py index 39a4216..3f64c47 100644 --- a/lib/streaming_providers/base/models/vod.py +++ b/lib/streaming_providers/base/models/vod.py @@ -157,7 +157,13 @@ class VodItem(Content): # Classification rating: Optional[str] = None # e.g. "FSK 12", "PG-13", "TV-MA" - genre: Optional[str] = None # already on Content but repeated for clarity + genre: Optional[str] = None # primary genre (mainGenre) + genres: Optional[List[str]] = None # full genre list + + # Extended descriptions + # description (short) is inherited from Content + long_description: Optional[str] = None # longDescription from contentInformation + original_title: Optional[str] = None # originalTitle — important for localised content # People cast: Optional[List[str]] = None @@ -166,6 +172,9 @@ class VodItem(Content): # Series / episode context (None for standalone movies / documentaries) season_number: Optional[int] = None episode_number: Optional[int] = None + # Back-references for episodes — lets the UI navigate up without re-parsing IDs + series_id: Optional[str] = None # e.g. "GN_SERIES_184925" + series_title: Optional[str] = None # e.g. "Two and a Half Men" # Promotional trailer_url: Optional[str] = None @@ -215,10 +224,16 @@ class VodItem(Content): "duration_minutes": self.duration_minutes, "release_year": self.release_year, "rating": self.rating, + "genre": self.genre, + "genres": self.genres, + "long_description": self.long_description, + "original_title": self.original_title, "cast": self.cast, "director": self.director, "season_number": self.season_number, "episode_number": self.episode_number, + "series_id": self.series_id, + "series_title": self.series_title, "trailer_url": self.trailer_url, }) return result diff --git a/lib/streaming_providers/providers/magenta2/config_models.py b/lib/streaming_providers/providers/magenta2/config_models.py index 0d18b14..3a55dab 100644 --- a/lib/streaming_providers/providers/magenta2/config_models.py +++ b/lib/streaming_providers/providers/magenta2/config_models.py @@ -10,6 +10,7 @@ class BootstrapConfig: client_model: str device_model: str + subscriber_type: str = "FTV_OTT_DT" # resolved from platform via SUBSCRIBER_TYPES sam3_client_id: Optional[str] = None taa_url: Optional[str] = None device_tokens_url: Optional[str] = None @@ -19,16 +20,23 @@ class BootstrapConfig: account_base_url: Optional[str] = None consumer_accounts_url: Optional[str] = None login_qr_code_url: Optional[str] = None + # VOD / personal-bar fields from baseSettings + home_url: Optional[str] = None # baseSettings.homeUrl + profile_name: Optional[str] = None # baseSettings.profileName + theme_id: Optional[str] = None # baseSettings.themeId raw_data: Dict[str, Any] = field(default_factory=dict) @classmethod def from_api_response(cls, bootstrap_data: Dict[str, Any], platform: str) -> "BootstrapConfig": """Create BootstrapConfig from API response""" + from .constants import SUBSCRIBER_TYPES + base_settings = bootstrap_data.get("baseSettings", {}) return cls( client_model=base_settings.get("clientModel", f"ftv-{platform}"), device_model=base_settings.get("deviceModel", f"{platform.upper()}_FTV"), + subscriber_type=SUBSCRIBER_TYPES.get(platform, "FTV_OTT_DT"), sam3_client_id=base_settings.get("sam3ClientId"), taa_url=base_settings.get("taaUrl"), device_tokens_url=base_settings.get("deviceTokensUrl"), @@ -38,6 +46,9 @@ class BootstrapConfig: account_base_url=base_settings.get("accountBaseUrl"), consumer_accounts_url=base_settings.get("consumerAccountsBaseUrl"), login_qr_code_url=base_settings.get("loginQrCodeUrl"), + home_url=base_settings.get("homeUrl"), + profile_name=base_settings.get("profileName"), + theme_id=base_settings.get("themeId"), raw_data=bootstrap_data, ) @@ -441,6 +452,32 @@ class ProviderConfig: resolved_client_model = client_model or self.bootstrap.client_model return hub_template.replace("{clientModel}", resolved_client_model) + def get_tvhubs_base_url(self, client_model: Optional[str] = None) -> Optional[str]: + """ + Return the tvhubs root URL (scheme + host + /v3/{clientModel}) with + the client model already substituted. + + Derived from the first available hub URL in the manifest by stripping + everything after /v3/{clientModel}/. Used by VodManager to build + StructuredGrid / UnstructuredGrid / VodDetails endpoints without + hardcoding the host. + """ + if not self.manifest: + return None + + resolved_client_model = client_model or self.bootstrap.client_model + + for hub_template in self.manifest.tv_hubs.base_urls.values(): + resolved = hub_template.replace("{clientModel}", resolved_client_model) + # All tvhubs URLs follow: https://host/v3/{clientModel}/... + # Split off everything after the client-model path segment. + marker = f"/v3/{resolved_client_model}" + idx = resolved.find(marker) + if idx != -1: + return resolved[: idx + len(marker)] + + return None + def get_mpx_account_uri(self) -> Optional[str]: """NEW: Get MPX account URI for persona token composition""" if not self.manifest or not self.manifest.mpx: @@ -457,4 +494,4 @@ class ProviderConfig: """NEW: Get authorize tokens URL from manifest""" if not self.manifest: return None - return self.manifest.get_authorize_tokens_url() + return self.manifest.get_authorize_tokens_url() \ No newline at end of file diff --git a/lib/streaming_providers/providers/magenta2/constants.py b/lib/streaming_providers/providers/magenta2/constants.py index 6702178..ff8cbb3 100644 --- a/lib/streaming_providers/providers/magenta2/constants.py +++ b/lib/streaming_providers/providers/magenta2/constants.py @@ -266,3 +266,35 @@ TAA_REQUEST_TEMPLATE = { BOOTSTRAP_PARAMS = {"$redirect": "false"} # REMOVED: Old BOOTSTRAP_KEYS - now handled in config_models.py + +# ============================================================================ +# VOD Configuration +# ============================================================================ + +# tvhubs base URL template — {client_model} is resolved at runtime from +# CLIENT_MODEL_MAPPINGS. This is a fallback; the live value should come +# from the manifest's tv_hubs.base_urls["ftv"] (or equivalent) via +# ProviderConfig.get_resolved_tvhub_url(). +TVHUBS_BASE_URL = "https://tvhubs.t-online.de/v3/{client_model}" + +# Flex IDs for the VOD catalogue. +# FLEX_ID_VOD_HOME is only used when personal-bar discovery fails (i.e. no +# homeUrl was obtained from bootstrap/manifest). +# FLEX_ID_VOD_DETAILS is stable across platforms — no discovery equivalent. +VOD_FLEX_ID_HOME = "164035" # StructuredGrid fallback for VOD home screen +VOD_FLEX_ID_DETAILS = "202887" # VodDetails flex ID +VOD_FLEX_ID_PLAYER = "202889" # VodPlayer flex ID + +# Pagination default for UnstructuredGrid lane requests +VOD_DEFAULT_PAGE_SIZE = 36 + +# Content-ID prefixes — used to route get_children() to the correct handler +VOD_PREFIX_SERIES = "GN_SERIES_" +VOD_PREFIX_SEASON = "GN_SEASON_" +VOD_PREFIX_EPISODE = "GN_EP" +VOD_PREFIX_MOVIE_MV = "GN_MV" +VOD_PREFIX_MOVIE_SH = "GN_SH" + +# Name of the personal-bar tile that leads to the VOD StructuredGrid. +# Looked up by exact title match in the primary.tiles list returned by homeUrl. +VOD_STREAMING_TILE_TITLE = "Streaming" \ No newline at end of file diff --git a/lib/streaming_providers/providers/magenta2/provider.py b/lib/streaming_providers/providers/magenta2/provider.py index 3a2bce5..3046d00 100644 --- a/lib/streaming_providers/providers/magenta2/provider.py +++ b/lib/streaming_providers/providers/magenta2/provider.py @@ -15,6 +15,7 @@ from ...base.models.proxy_models import ProxyConfig from ...base.network import HTTPManagerFactory, ProxyConfigManager from ...base.provider import StreamingProvider from ...base.utils.logger import logger +from .vod_manager import VodManager from .auth import Magenta2Authenticator, Magenta2Credentials, Magenta2UserCredentials from .config_models import ProviderConfig from .constants import ( @@ -113,6 +114,14 @@ class Magenta2Provider(StreamingProvider): self.endpoint_manager: Optional[EndpointManager] = None self.provider_config: Optional[ProviderConfig] = None + self._vod_manager = VodManager( + http_manager=self.http_manager, + provider_name=self.provider_name, + bootstrap=self.endpoint_manager.config.bootstrap, + provider_config=self.endpoint_manager.config, + session_id=self.authenticator._session_id, + ) + # 🚨 INITIALIZE AUTHENTICATOR FIRST (with minimal config) # Use fallback client IDs initially fallback_client_id = MAGENTA2_CLIENT_IDS.get( @@ -1100,6 +1109,13 @@ class Magenta2Provider(StreamingProvider): return successful_channels + def get_vod_category(self, category_path, **kwargs): + """Delegate VOD catalogue browsing to VodManager.""" + return self._vod_manager.get_children( + category_path=category_path, + **kwargs, + ) + def enrich_channel_data( self, channel: StreamingChannel, **kwargs ) -> Optional[StreamingChannel]: diff --git a/lib/streaming_providers/providers/magenta2/vod_manager.py b/lib/streaming_providers/providers/magenta2/vod_manager.py new file mode 100644 index 0000000..53033d3 --- /dev/null +++ b/lib/streaming_providers/providers/magenta2/vod_manager.py @@ -0,0 +1,619 @@ +# streaming_providers/providers/magenta2/vod_manager.py +""" +Magenta2 VOD Manager + +Handles VOD catalogue browsing for the Magenta2 provider. + +Browsing API hierarchy +----------------------- + StructuredGrid/{VOD_FLEX_ID_HOME} → top-level lanes + UnstructuredGrid/{flex_id} → paginated lane items + VodDetails/{VOD_FLEX_ID_DETAILS}/{gn_id} → series / movie / episode detail + VodDetails/{VOD_FLEX_ID_DETAILS}/GN_SEASON_{id}_DE_{n} → season detail + +Playback chain (implemented in provider.get_manifest) +------------------------------------------------------- + 1. VodDetails response → content.productInformationLink.href + (wcps .../uspip/.../vodproductinformation/.../GN_...) + Stored as VodItem.manifest_script at browse time. + + 2. Fetch productInformationLink → buttons.primary[] + Pick entry where rel=="player" AND instantUsable==true + (skip rel=="launchApp" -- those are external apps, not streams) + → href = VodPlayer URL (follow directly, do not construct) + → partnerId already in the URL as a query param + + 3. Fetch VodPlayer href (append $redirect=false + sid from _base_params) + → content.playbackUrls[].href + (link.theplatform.eu/s/mdeprod/media/{mpx_id}) + That MPX selector URL is the manifest -- identical format to live channels. + + Bonus: buttons.secondary[] where rel=="trailer" + → href = trailer URL (populate VodItem.trailer_url if desired) + +tvhubs base URL resolution order +--------------------------------- +1. Manifest tv_hubs.base_urls (via ProviderConfig.get_tvhubs_base_url) +2. TVHUBS_BASE_URL constant (fallback) + +Public interface +----------------- + vod_manager.get_children(category_path, **kwargs) + -> List[VodCategory | VodItem] +""" + +import time +import logging +from typing import Dict, List, Optional, Union + +from ...base.models.vod import VodCategory, VodItem + +from .constants import ( + SUBSCRIBER_TYPES, + TVHUBS_BASE_URL, + VOD_DEFAULT_PAGE_SIZE, + VOD_FLEX_ID_DETAILS, + VOD_FLEX_ID_HOME, + VOD_PREFIX_EPISODE, + VOD_PREFIX_SEASON, + VOD_PREFIX_SERIES, + VOD_STREAMING_TILE_TITLE, +) + +logger = logging.getLogger(__name__) + + +class VodManager: + """ + Manages Magenta2 VOD catalogue traversal. + + Args: + http_manager: An HTTPManager instance (from the parent provider). + provider_name: Provider identifier string (e.g. "magenta2"). + bootstrap: BootstrapConfig obtained from discovery. Supplies all + per-platform values (device_model, subscriber_type, + client_model, profile_name, theme_id, home_url) so + that platform identity does not need to be carried + through this class at all. + provider_config: Optional ProviderConfig; when supplied the tvhubs base + URL is resolved from the manifest instead of the + TVHUBS_BASE_URL fallback constant. + """ + + def __init__( + self, + http_manager, + provider_name: str, + bootstrap=None, + provider_config=None, + session_id: Optional[str] = None, + ): + self._http = http_manager + self._provider = provider_name + self._provider_config = provider_config + self._session_id: str = session_id or "" + + # All content values resolved from BootstrapConfig -- no platform needed. + self._home_url: Optional[str] = getattr(bootstrap, "home_url", None) + self._client_model: str = getattr(bootstrap, "client_model", None) or "ftv-web" + self._device_model: str = getattr(bootstrap, "device_model", None) or "WEB2_FTV" + self._profile_name: str = getattr(bootstrap, "profile_name", None) or "stageExt" + self._theme_id: str = getattr(bootstrap, "theme_id", None) or "hdr-ui2" + # subscriber_type is not in bootstrap; derive from platform once at init. + _platform = getattr(bootstrap, "platform", "") + self._subscriber_type: str = SUBSCRIBER_TYPES.get(_platform, "FTV_OTT_DT") + + # ========================================================================= + # Public API + # ========================================================================= + + def get_children( + self, + category_path: List[str], + *, + page_size: int = VOD_DEFAULT_PAGE_SIZE, + offset: int = 0, + ) -> List[Union[VodCategory, VodItem]]: + """ + Return the children of a VOD tree node. + + Args: + category_path: Ordered list of content_ids from root to the + requested node, matching the provider's contract: + [] -> VOD home (top-level lanes) + ["GN_SERIES_123"] -> series (seasons) + ["GN_SERIES_123", + "GN_SEASON_123_DE_2"] -> season (episodes) + ["lane:326619"] -> UnstructuredGrid lane + page_size: Number of items to fetch per page (UnstructuredGrid). + offset: Pagination offset (UnstructuredGrid). + + Returns: + Mixed list of VodCategory and VodItem objects. + """ + params = self._base_params() + + if not category_path: + return self._fetch_home_lanes(params) + + node_id = category_path[-1] + + if node_id.startswith("lane:"): + flex_id = node_id[5:] + return self._fetch_lane_items(flex_id, params, page_size=page_size, offset=offset) + + if node_id.startswith(VOD_PREFIX_SEASON): + return self._fetch_season_episodes(node_id, params) + + if node_id.startswith(VOD_PREFIX_SERIES): + return self._fetch_series_seasons(node_id, params) + + if node_id.startswith(VOD_PREFIX_EPISODE): + return self._fetch_single_episode(node_id, params) + + # Movies (GN_MV, GN_SH) and any other leaf ID + return self._fetch_single_item(node_id, params) + + # ========================================================================= + # Private helpers – HTTP layer + # ========================================================================= + + def _base_params(self) -> Dict[str, str]: + """Build query parameters common to every VOD API request.""" + return { + "$deviceModel": self._device_model, + "$profile": self._profile_name, + "$subscriberType": self._subscriber_type, + "$theme": self._theme_id, + "$redirect": "false", + "sid": self._session_id, + "t": str(int(time.time() * 1000)), + } + + def _base_url(self) -> str: + """ + Resolve the tvhubs root URL (https://host/v3/{clientModel}) with the + client model already substituted. + + Resolution order: + 1. ProviderConfig.get_tvhubs_base_url() — derives the root from hub + URLs already in the manifest, reusing data fetched for live TV. + 2. TVHUBS_BASE_URL constant — fallback when manifest is unavailable. + """ + if self._provider_config is not None: + resolved = self._provider_config.get_tvhubs_base_url(self._client_model) + if resolved: + return resolved + + return TVHUBS_BASE_URL.format(client_model=self._client_model) + + def _get(self, url: str, params: Dict) -> Optional[Dict]: + """Perform a GET request and return the parsed JSON body, or None on error.""" + try: + response = self._http.get(url, params=params) + if response and response.status_code == 200: + return response.json() + logger.warning( + f"{self._provider}: VOD request failed " + f"[{response.status_code if response else 'no response'}] {url}" + ) + except Exception as exc: + logger.error(f"{self._provider}: VOD request exception for {url}: {exc}") + return None + + # ========================================================================= + # Private helpers – Personal Bar Discovery + # ========================================================================= + + def _get_streaming_grid_url(self) -> Optional[str]: + """ + Fetch the personal bar from homeUrl and extract the Streaming tile's + onFocus screen href, which contains the StructuredGrid URL for VOD. + """ + if not self._home_url: + logger.debug( + f"{self._provider}: No homeUrl available; " + "skipping personal-bar VOD discovery" + ) + return None + + params = self._base_params() + # homeUrl needs the un-prefixed parameter names alongside the standard ones + params.update({ + "deviceModel": self._device_model, + "profileName": self._profile_name, + "themeId": self._theme_id, + }) + + data = self._get(self._home_url, params) + if not data: + logger.error(f"{self._provider}: Failed to fetch personal bar from homeUrl") + return None + + tiles = data.get("primary", {}).get("tiles", []) + for tile in tiles: + if tile.get("title") == VOD_STREAMING_TILE_TITLE: + href = tile.get("onFocus", {}).get("screen", {}).get("href") + if href: + logger.debug(f"{self._provider}: Found Streaming grid URL: {href}") + return href + + logger.warning( + f"{self._provider}: '{VOD_STREAMING_TILE_TITLE}' tile not found in personal bar" + ) + return None + + # ========================================================================= + # Private helpers – StructuredGrid (home) + # ========================================================================= + + def _fetch_home_lanes(self, params: Dict) -> List[VodCategory]: + """ + Fetch the VOD home StructuredGrid and return each browsable lane as a + VodCategory. External-partner lanes (Disney+, RTL+, …) are skipped. + """ + streaming_grid_url = self._get_streaming_grid_url() + + if streaming_grid_url: + url = streaming_grid_url + logger.debug(f"{self._provider}: Using discovered VOD home URL: {url}") + else: + url = f"{self._base_url()}/StructuredGrid/{VOD_FLEX_ID_HOME}" + logger.warning(f"{self._provider}: Falling back to hardcoded VOD home URL: {url}") + + data = self._get(url, params) + if not data: + return [] + + lanes = data.get("content", {}).get("lanes", []) + categories: List[VodCategory] = [] + + for lane in lanes: + lane_type = lane.get("type", "") + title = lane.get("title", "").strip() + flex_id = lane.get("flexId", "") + + # Skip external partner lanes (e.g. action == "ChannelTuneOpenApp") + show_all = lane.get("showAllUrl", {}) + action = show_all.get("action", "") if isinstance(show_all, dict) else "" + if action == "ChannelTuneOpenApp": + logger.debug(f"{self._provider}: Skipping external partner lane '{title}'") + continue + + if lane_type != "UnstructuredGrid" or not flex_id or not title: + continue + + categories.append( + VodCategory( + name=title, + content_id=f"lane:{flex_id}", + provider=self._provider, + child_count=lane.get("totalCount"), + ) + ) + + logger.debug(f"{self._provider}: Found {len(categories)} VOD home lanes") + return categories + + # ========================================================================= + # Private helpers – UnstructuredGrid (lane items) + # ========================================================================= + + def _fetch_lane_items( + self, + flex_id: str, + params: Dict, + page_size: int = VOD_DEFAULT_PAGE_SIZE, + offset: int = 0, + ) -> List[Union[VodCategory, VodItem]]: + """ + Fetch items from an UnstructuredGrid lane. Movies → VodItem, + series → VodCategory (seasons/episodes require further drill-down). + """ + url = f"{self._base_url()}/UnstructuredGrid/{flex_id}" + paged_params = dict(params) + paged_params["$size"] = str(page_size) + paged_params["$offset"] = str(offset) + + data = self._get(url, paged_params) + if not data: + return [] + + content = data.get("content", {}) + results: List[Union[VodCategory, VodItem]] = [] + for item in content.get("items", []): + node = self._map_unstructured_item(item) + if node is not None: + results.append(node) + + total = content.get("page", {}).get("total", len(results)) + logger.debug( + f"{self._provider}: Lane {flex_id} – fetched {len(results)}/{total} items " + f"(offset={offset})" + ) + return results + + def _map_unstructured_item( + self, item: Dict + ) -> Optional[Union[VodCategory, VodItem]]: + """Map a single UnstructuredGrid item dict to a VodCategory or VodItem.""" + content_id: str = item.get("id", "") + title: str = (item.get("title") or "").strip() + vod_type: str = item.get("vodType", "") + image_url: Optional[str] = (item.get("image") or {}).get("href") + description: Optional[str] = item.get("description") + year_raw = item.get("yearOfProduction") + release_year: Optional[int] = int(year_raw) if year_raw else None + genre: Optional[str] = item.get("mainGenre") + genres: Optional[List[str]] = item.get("genres") or None + duration_raw = item.get("duration") + seasons_available: Optional[int] = item.get("seasonsAvailable") + rating: Optional[str] = item.get("childProtectionId") + + if not content_id or not title: + return None + + if vod_type == "Series": + return VodCategory( + name=title, + content_id=content_id, + provider=self._provider, + logo_url=image_url, + description=description, + child_count=seasons_available, + ) + + duration_seconds: Optional[int] = int(duration_raw) * 60 if duration_raw else None + return VodItem.create_movie( + name=title, + content_id=content_id, + provider=self._provider, + logo_url=image_url, + description=description, + release_year=release_year, + genre=genre, + genres=genres, + duration_seconds=duration_seconds, + rating=rating, + ) + + # ========================================================================= + # Private helpers – VodDetails (series → seasons) + # ========================================================================= + + def _fetch_series_seasons(self, content_id: str, params: Dict) -> List[VodCategory]: + """Fetch a series VodDetails page and return each season as a VodCategory.""" + url = f"{self._base_url()}/VodDetails/{VOD_FLEX_ID_DETAILS}/{content_id}" + data = self._get(url, params) + if not data: + return [] + + content = data.get("content", {}) + info = content.get("contentInformation", {}) + series_title: str = info.get("seriesTitle") or info.get("title") or "" + series_num = content_id.replace(VOD_PREFIX_SERIES, "") + + seasons: List[VodCategory] = [] + for lane in content.get("lanes", []): + if lane.get("type") != "Season": + continue + season_num = lane.get("seasonNumber") + if season_num is None: + continue + season_title = lane.get("title") or f"Staffel {season_num}" + season_id = f"{VOD_PREFIX_SEASON}{series_num}_DE_{season_num}" + seasons.append( + VodCategory( + name=season_title, + content_id=season_id, + provider=self._provider, + description=f"{series_title} – {season_title}", + child_count=lane.get("episodeCount") or lane.get("totalCount"), + ) + ) + + # Fallback: build from seasonsAvailable count + if not seasons: + seasons_count = info.get("seasonsAvailable") or 0 + for s in range(1, int(seasons_count) + 1): + season_id = f"{VOD_PREFIX_SEASON}{series_num}_DE_{s}" + seasons.append( + VodCategory( + name=f"Staffel {s}", + content_id=season_id, + provider=self._provider, + description=f"{series_title} – Staffel {s}", + ) + ) + + logger.debug(f"{self._provider}: Series {content_id} → {len(seasons)} seasons") + return seasons + + # ========================================================================= + # Private helpers – VodDetails (season → episodes) + # ========================================================================= + + def _fetch_season_episodes(self, season_id: str, params: Dict) -> List[VodItem]: + """ + Fetch episodes for a season. + + The season VodDetails response does not embed episode items inline. + Instead each season lane carries a laneContentLink href that points to + an UnstructuredGrid endpoint listing the actual episodes. We fetch + that endpoint for every Episode-type lane we find. + """ + url = f"{self._base_url()}/VodDetails/{VOD_FLEX_ID_DETAILS}/{season_id}" + episode_params = dict(params) + episode_params["autofocus"] = "videoload" + + data = self._get(url, episode_params) + if not data: + return [] + + content = data.get("content", {}) + season_number: Optional[int] = self._extract_season_number(season_id) + + episodes: List[VodItem] = [] + for lane in content.get("lanes", []): + if lane.get("type") != "Episode": + continue + lane_url = (lane.get("laneContentLink") or {}).get("href") + if not lane_url: + continue + lane_data = self._get(lane_url, params) + if not lane_data: + continue + for ep in lane_data.get("content", {}).get("items", []): + node = self._map_episode_item(ep, season_number) + if node is not None: + episodes.append(node) + + logger.debug(f"{self._provider}: Season {season_id} → {len(episodes)} episodes") + return episodes + + def _map_episode_item( + self, + ep: Dict, + season_number: Optional[int], + ) -> Optional[VodItem]: + """ + Map an episode item from a season lane content response to a VodItem. + + manifest_script is not set here — lane listings do not include + productInformationLink. The provider's get_manifest(content_id) must + fetch VodDetails for the episode to obtain it. + """ + content_id: str = ep.get("id", "") + if not content_id: + return None + ep_num: Optional[int] = ep.get("episodeNumber") + title: str = (ep.get("title") or f"Episode {ep_num}").strip() + image_url: Optional[str] = (ep.get("image") or {}).get("href") + description: Optional[str] = ep.get("description") + duration_raw = ep.get("duration") + duration_seconds: Optional[int] = int(duration_raw) * 60 if duration_raw else None + genre: Optional[str] = ep.get("mainGenre") + genres: Optional[List[str]] = ep.get("genres") or None + rating: Optional[str] = ep.get("childProtectionId") + + return VodItem.create_episode( + name=title, + content_id=content_id, + provider=self._provider, + season_number=season_number or 0, + episode_number=ep_num or 0, + logo_url=image_url, + description=description, + duration_seconds=duration_seconds, + genre=genre, + genres=genres, + rating=rating, + ) + + def _fetch_single_episode(self, content_id: str, params: Dict) -> List[VodItem]: + """ + Fetch a single episode by content_id. + + The VodDetails response for an episode has the metadata directly in + contentInformation — there are no inline episode items in the lanes. + The lanes only contain Person and recommendation (UnstructuredGrid) + rows, neither of which are relevant here. + """ + episode_params = dict(params) + episode_params["autofocus"] = "videoload" + url = f"{self._base_url()}/VodDetails/{VOD_FLEX_ID_DETAILS}/{content_id}" + data = self._get(url, episode_params) + if not data: + return [] + + content_block = data.get("content", {}) + info = content_block.get("contentInformation", {}) + ep_num: Optional[int] = info.get("episodeNumber") + title: str = (info.get("title") or f"Episode {ep_num}").strip() + if not title: + return [] + + duration_seconds: Optional[int] = ( + int(info["runtime"]) * 60 if info.get("runtime") else None + ) + product_url: Optional[str] = ( + content_block.get("productInformationLink") or {} + ).get("href") + return [VodItem.create_episode( + name=title, + content_id=content_id, + provider=self._provider, + season_number=info.get("seasonNumber") or 0, + episode_number=ep_num or 0, + logo_url=(info.get("image") or {}).get("href"), + description=info.get("description"), + long_description=info.get("longDescription"), + original_title=info.get("originalTitle"), + genre=info.get("mainGenre"), + genres=info.get("genres") or None, + duration_seconds=duration_seconds, + rating=info.get("childProtectionId"), + series_id=info.get("seriesId"), + series_title=info.get("seriesTitle"), + manifest_script=product_url, + session_manifest=product_url is not None, + )] + + def _fetch_single_item(self, content_id: str, params: Dict) -> List[VodItem]: + """ + Fetch a movie (or unknown leaf) by content_id. + + Metadata comes directly from contentInformation. The lanes in the + real API response only contain Person and recommendation rows — there + are no playable child items to recurse into. + """ + url = f"{self._base_url()}/VodDetails/{VOD_FLEX_ID_DETAILS}/{content_id}" + data = self._get(url, params) + if not data: + return [] + + content_block = data.get("content", {}) + info = content_block.get("contentInformation", {}) + title: str = (info.get("title") or "").strip() + if not title: + return [] + + duration_seconds: Optional[int] = ( + int(info["runtime"]) * 60 if info.get("runtime") else None + ) + year_raw = info.get("yearFrom") + product_url: Optional[str] = ( + content_block.get("productInformationLink") or {} + ).get("href") + return [VodItem.create_movie( + name=title, + content_id=content_id, + provider=self._provider, + logo_url=(info.get("image") or {}).get("href"), + description=info.get("description"), + long_description=info.get("longDescription"), + original_title=info.get("originalTitle"), + release_year=int(year_raw) if year_raw else None, + genre=info.get("mainGenre"), + genres=info.get("genres") or None, + duration_seconds=duration_seconds, + rating=info.get("childProtectionId"), + manifest_script=product_url, + session_manifest=product_url is not None, + )] + + # ========================================================================= + # Utilities + # ========================================================================= + + @staticmethod + def _extract_season_number(season_id: str) -> Optional[int]: + """ + Parse the season number from a season content_id. + + e.g. "GN_SEASON_184925_DE_3" → 3 + """ + try: + return int(season_id.rsplit("_DE_", 1)[-1]) + except (ValueError, IndexError): + return None \ No newline at end of file