From a6078e98e2dca3efc01ad9094e44b5be0e3e674d Mon Sep 17 00:00:00 2001 From: Nirvana Date: Wed, 27 May 2026 20:12:29 +0200 Subject: [PATCH] Refactor Magenta2 --- .../providers/magenta2/auth_bridge.py | 320 +++ .../providers/magenta2/channel_manager.py | 648 ++++++ .../providers/magenta2/playback_manager.py | 221 ++ .../providers/magenta2/provider.py | 1866 ++++------------- 4 files changed, 1617 insertions(+), 1438 deletions(-) create mode 100644 lib/streaming_providers/providers/magenta2/auth_bridge.py create mode 100644 lib/streaming_providers/providers/magenta2/channel_manager.py create mode 100644 lib/streaming_providers/providers/magenta2/playback_manager.py diff --git a/lib/streaming_providers/providers/magenta2/auth_bridge.py b/lib/streaming_providers/providers/magenta2/auth_bridge.py new file mode 100644 index 0000000..2fe02bb --- /dev/null +++ b/lib/streaming_providers/providers/magenta2/auth_bridge.py @@ -0,0 +1,320 @@ +# streaming_providers/providers/magenta2/auth_bridge.py +# -*- coding: utf-8 -*- +""" +AuthBridge — token-consumer layer between Magenta2Provider and Magenta2Authenticator. + +Responsibilities: + - In-memory persona-token cache (TTL-based, 60 s safety margin) + - tvhubs-scoped Bearer token retrieval and refresh (via TokenFlowManager) + - Platform-aware auth-header construction for VOD and nPVR endpoints + - Auth-state and readiness inspection (moved from provider.py) + +What this class does NOT do: + - Token acquisition / OAuth flows → Magenta2Authenticator / TokenFlowManager + - Endpoint discovery → DiscoveryService / EndpointManager + - Any HTTP requests of its own + +Usage (from Magenta2Provider.__init__):: + + self._auth = AuthBridge( + authenticator=self.authenticator, + provider_name=self.provider_name, + country=self.country, + platform=self.platform, + platform_config=self.platform_config, + provider_config=self.provider_config, + session_id=self.session_id, + serial_number=self.serial_number, + generate_call_id=self._generate_call_id, + ) +""" + +import time +from typing import Any, Dict, Optional, Tuple + +from ...base.models.auth import AuthState +from ...base.utils.logger import logger +from .playback_manager import PlaybackManager +from .token_flow_manager import PersonaResult + + +class AuthBridge: + """ + Thin token-consumer and caching façade. + + The bridge holds a reference to the authenticator but never calls into + its private internals — all refresh work is delegated to the public + TokenFlowManager API. + """ + + def __init__( + self, + authenticator: Any, + provider_name: str, + country: str, + platform: str, + platform_config: Dict[str, Any], + provider_config: Any, + session_id: str, + serial_number: str, + generate_call_id, # callable() -> str + ) -> None: + self._authenticator = authenticator + self._provider_name = provider_name + self._country = country + self._platform = platform + self._platform_config = platform_config + self._provider_config = provider_config + self._session_id = session_id + self._serial_number = serial_number + self._generate_call_id = generate_call_id + + self._persona_cache: Optional[PersonaResult] = None + + # ------------------------------------------------------------------ # + # provider_config setter — updated after configuration discovery # + # ------------------------------------------------------------------ # + + def update_provider_config(self, provider_config: Any) -> None: + """Called by the provider after configuration discovery completes.""" + self._provider_config = provider_config + + # ------------------------------------------------------------------ # + # Persona token # + # ------------------------------------------------------------------ # + + def get_persona_token(self, force_refresh: bool = False) -> str: + """ + Return a valid persona token, using an in-memory TTL cache. + + Raises: + RuntimeError: If TokenFlowManager is not available. + Exception: If the underlying token flow fails. + """ + if not force_refresh and self._persona_cache and self._persona_cache.success: + expires_at: float = self._persona_cache.expires_at + if time.time() < (expires_at - 60): + logger.debug( + f"Using cached persona token (expires at {time.ctime(expires_at)})" + ) + return self._persona_cache.persona_token + self._persona_cache = None + logger.debug("In-memory persona cache expired") + + tfm = self._authenticator.token_flow_manager + if tfm is None: + raise RuntimeError("TokenFlowManager is not available on the authenticator") + + persona_result: PersonaResult = tfm.get_persona_token(force_refresh=force_refresh) + if not persona_result.success: + raise Exception(f"Failed to get persona token: {persona_result.error}") + + self._persona_cache = persona_result + logger.debug( + f"Cached persona token (expires at {time.ctime(persona_result.expires_at)})" + ) + return persona_result.persona_token + + def ensure_authenticated(self) -> str: + """Return a valid persona token (lazy, no forced refresh). Callable as a callback.""" + return self.get_persona_token(force_refresh=False) + + def clear_persona_cache(self) -> None: + """Discard the in-memory persona token cache.""" + self._persona_cache = None + logger.debug("Cleared in-memory persona cache") + + # ------------------------------------------------------------------ # + # tvhubs Bearer token # + # ------------------------------------------------------------------ # + + def get_tvhubs_bearer(self) -> Optional[str]: + """ + Return a valid tvhubs-scoped access token for use as a Bearer token. + + Delegates to TokenFlowManager.get_tvhubs_token() which owns the full + lifecycle: cache read → TTL check → refresh → full re-auth chain fallback. + Returns None if the token is unavailable after all attempts. + """ + tfm = self._authenticator.token_flow_manager + if tfm is None: + return None + + try: + return tfm.get_tvhubs_token() + except Exception as exc: + logger.debug(f"Could not obtain tvhubs token: {exc}") + return None + + # ------------------------------------------------------------------ # + # Auth-header builders # + # ------------------------------------------------------------------ # + + def vod_auth_headers(self) -> Dict[str, str]: + """ + Build auth headers for VOD endpoints, platform-aware. + + Strategy: + 1. Prefer tvhubs-scoped Bearer token. + 2. Fall back to persona JWT extracted from the composed persona token. + 3. Last resort: Basic + raw persona token. + """ + persona_token = self.ensure_authenticated() + tvhubs_token = self.get_tvhubs_bearer() + + if tvhubs_token: + auth_value = f"Bearer {tvhubs_token}" + logger.debug("VOD auth: using tvhubs token as Bearer") + else: + persona_jwt = PlaybackManager.extract_persona_jwt_from_token(persona_token) + auth_value = ( + f"Bearer {persona_jwt}" if persona_jwt else f"Basic {persona_token}" + ) + logger.debug("VOD auth: tvhubs token unavailable, falling back to persona_jwt") + + client_model: str = ( + self._provider_config.bootstrap.client_model + if self._provider_config and self._provider_config.bootstrap + else f"ftv-{self._platform}" + ) or f"ftv-{self._platform}" + is_web = client_model == "ftv-web" + + if is_web: + return { + "Authorization": auth_value, + "x-mpx-authorization": f"Basic {persona_token}", + "x-dt-session-id": self._session_id, + "x-dt-call-id": self._generate_call_id(), + "origin": "https://www.magenta.tv", + "referer": "https://www.magenta.tv/", + "user-agent": self._platform_config["user_agent"], + "accept": "*/*", + "accept-encoding": "gzip, deflate, br, zstd", + "accept-language": "de-DE,de;q=0.9", + "x-permissionflagpersonalizeduireco": "false", + } + else: + return { + "Authorization": auth_value, + "x-mpx-authorization": f"Basic {persona_token}", + "x-stbserialnumber": self._serial_number, + "dt-session-id": self._session_id, + "dt-call-id": self._generate_call_id(), + "user-agent": self._platform_config["user_agent"], + "accept-encoding": "gzip", + } + + def pvr_auth_headers(self) -> Dict[str, str]: + """Build auth headers for nPVR recording endpoints.""" + persona_token = self.ensure_authenticated() + return { + "Authorization": f"Basic {persona_token}", + "User-Agent": self._platform_config["user_agent"], + "Accept-Encoding": "gzip", + "CID": f"{self._session_id}::{self._generate_call_id()}", + } + + # ------------------------------------------------------------------ # + # Auth-state introspection (moved from provider.py) # + # ------------------------------------------------------------------ # + + def calculate_auth_state(self, context: Any) -> AuthState: + """Derive AuthState from the stored persona token.""" + persona_token = context.get_token(self._provider_name, "persona", self._country) + if not persona_token: + logger.debug("No persona token found") + return AuthState.NOT_AUTHENTICATED + if not isinstance(persona_token, dict) or "persona_token" not in persona_token: + logger.warning("Invalid persona token structure") + return AuthState.NOT_AUTHENTICATED + if "expires_at" in persona_token: + current_time = time.time() + expires_at: float = persona_token["expires_at"] + if current_time >= (expires_at - 300): + logger.debug( + f"Persona token expired (expires_at: {expires_at}, now: {current_time})" + ) + return AuthState.EXPIRED + logger.debug("Persona token is valid") + return AuthState.AUTHENTICATED + + def calculate_readiness(self, context: Any) -> Tuple[bool, str]: + """Return (is_ready, reason) based on token availability and expiry.""" + persona_token = context.get_token(self._provider_name, "persona", self._country) + if not persona_token: + return False, "No persona token available" + if not isinstance(persona_token, dict) or "persona_token" not in persona_token: + return False, "Invalid persona token structure" + + if "expires_at" in persona_token: + current_time = time.time() + expires_at: float = persona_token["expires_at"] + if current_time >= (expires_at - 300): + yo_token = context.get_token( + self._provider_name, "yo_digital", self._country + ) + if yo_token and "refresh_token" in yo_token: + if ( + "refresh_token_expires_in" in yo_token + and "refresh_token_issued_at" in yo_token + ): + refresh_expires_at: float = ( + yo_token["refresh_token_issued_at"] + + yo_token["refresh_token_expires_in"] + ) + if current_time < (refresh_expires_at - 300): + return ( + True, + "Persona token expired but can be refreshed via yo_digital", + ) + return False, f"Persona token expired (expired at {time.ctime(expires_at)})" + + return True, "Has valid persona token" + + def get_auth_details(self, token_scopes: list, context: Any) -> Dict[str, Any]: + """Return per-scope token status for all provider token scopes.""" + details: Dict[str, Any] = {} + for scope in token_scopes: + token = context.get_token(self._provider_name, scope, self._country) + if not token: + details[scope] = {"available": False} + continue + + scope_info: Dict[str, Any] = {"available": True} + + if scope == "persona": + if "expires_at" in token: + current_time = time.time() + expires_at: float = token["expires_at"] + scope_info["expires_at"] = expires_at + scope_info["is_expired"] = current_time >= expires_at + scope_info["time_remaining"] = int(max(0, expires_at - current_time)) + if "composed_at" in token: + scope_info["composed_at"] = token["composed_at"] + + elif scope == "yo_digital": + if "access_token_expires_in" in token and "access_token_issued_at" in token: + current_time = time.time() + at_expires: float = ( + token["access_token_issued_at"] + token["access_token_expires_in"] + ) + scope_info["access_token_expires_at"] = at_expires + scope_info["access_token_is_expired"] = current_time >= at_expires + if "refresh_token_expires_in" in token and "refresh_token_issued_at" in token: + current_time = time.time() + rt_expires: float = ( + token["refresh_token_issued_at"] + token["refresh_token_expires_in"] + ) + scope_info["refresh_token_expires_at"] = rt_expires + scope_info["refresh_token_is_expired"] = current_time >= rt_expires + scope_info["has_refresh_token"] = True + + else: + if "expires_in" in token and "issued_at" in token: + current_time = time.time() + std_expires: float = token["issued_at"] + token["expires_in"] + scope_info["expires_at"] = std_expires + scope_info["is_expired"] = current_time >= std_expires + + details[scope] = scope_info + return details \ No newline at end of file diff --git a/lib/streaming_providers/providers/magenta2/channel_manager.py b/lib/streaming_providers/providers/magenta2/channel_manager.py new file mode 100644 index 0000000..74ec471 --- /dev/null +++ b/lib/streaming_providers/providers/magenta2/channel_manager.py @@ -0,0 +1,648 @@ +# streaming_providers/providers/magenta2/channel_manager.py +# -*- coding: utf-8 -*- +""" +Manages channel discovery, metadata enrichment, entitlement, and streaming-data +population for the Magenta2 provider. + +Responsibilities +---------------- +- Fetch and cache the unauthenticated channel-stations feed (_fetch_station_metadata) +- Build StreamingChannel objects from theplatform feed entries +- Fetch the entitled-channels feed and merge it with station metadata (get_channels) +- Request entitlement tokens and playlist data per channel +- Populate streaming data (manifest URL, DRM) for a set of channels + +The live-manifest / live-pid caches live here because they are produced by +get_channels() and consumed by PlaybackManager via the provider's cache +attributes (_live_manifest_cache, _live_pid_cache). +""" +import json +import time +from typing import Callable, Dict, List, Optional + +from ...base.models import StreamingChannel +from ...base.utils.logger import logger +from .constants import ( + CONTENT_TYPE_LIVE, + DEFAULT_EPG_WINDOW_HOURS, + DEFAULT_MAX_RETRIES, + DEFAULT_REQUEST_TIMEOUT, + DISTRIBUTION_PACKAGE_NAMES, + DRM_SYSTEM_WIDEVINE, + ERROR_CODES, + MODE_LIVE, + QUALITY_RANK, +) +from .endpoint_manager import EndpointManager +from .config_models import ProviderConfig +from .models import Magenta2Channel, Magenta2PlaybackRestrictedException +from ..lib_theplatform import ( + TheplatformChannel, + fetch_distribution_rights, + fetch_entitled_channels_feed, +) + + +class ChannelManager: + """ + Handles all channel-related operations for Magenta2. + + Parameters + ---------- + http_manager: + Shared HTTP manager (created by the provider). + provider_name: + Provider name string used when building StreamingChannel objects. + country: + Two-letter country code. + platform_config: + Platform-specific dict from MAGENTA2_PLATFORMS (user_agent, etc.). + session_id: + Session UUID shared with the provider instance. + serial_number: + Device serial UUID shared with the provider instance. + endpoint_manager: + Populated EndpointManager after discovery. + provider_config: + ProviderConfig after discovery. + auth_callback: + Callable[[], str] — returns a valid persona token (Basic-auth value). + Provided by the provider as ``self._ensure_authenticated``. + build_scaled_image_url_callback: + Callable[[str], Optional[str]] — scales a logo URL. + Provided by the provider. + """ + + def __init__( + self, + http_manager, + provider_name: str, + country: str, + platform_config: Dict, + session_id: str, + serial_number: str, + endpoint_manager: Optional[EndpointManager], + provider_config: Optional[ProviderConfig], + auth_callback: Callable[[], str], + build_scaled_image_url_callback: Callable[[str], Optional[str]], + ): + self._http = http_manager + self._provider_name = provider_name + self._country = country + self._platform_config = platform_config + self._session_id = session_id + self._serial_number = serial_number + self._endpoint_manager = endpoint_manager + self._provider_config = provider_config + self._ensure_authenticated = auth_callback + self._build_scaled_image_url = build_scaled_image_url_callback + + # Populated on first get_channels() call; returned directly on subsequent calls. + self._cached_channels: Optional[List[StreamingChannel]] = None + + # release_pid → mpd_url, keyed by release_pid (= channel_id after get_channels). + self._live_manifest_cache: Dict[str, str] = {} + # release_pid → release_pid (used as a fast membership test by PlaybackManager). + self._live_pid_cache: Dict[str, str] = {} + + # Populated eagerly at provider init (unauthenticated call). + self.station_metadata: Dict[str, Dict] = self._fetch_station_metadata() + + # ------------------------------------------------------------------ # + # Public API # + # ------------------------------------------------------------------ # + + def get_channels( + self, + time_window_hours: int = DEFAULT_EPG_WINDOW_HOURS, + fetch_manifests: bool = False, + populate_streaming: bool = True, + prefer_highest_quality: bool = True, + **kwargs, + ) -> List[StreamingChannel]: + """ + Fetch available channels via the entitled-channels flow. + + Uses lib_theplatform to: + 1. Call getApplicableDistributionRights (license_service_url from manifest). + 2. Fetch the entitled-channels feed filtered by those rights. + 3. Merge with station metadata pre-fetched at init time (unauthenticated). + 4. Convert each TheplatformChannel to a StreamingChannel. + + Results are cached after the first successful call; subsequent calls + return from cache without hitting the network. + """ + if self._cached_channels: + logger.debug("get_channels: returning from cache (no network calls)") + return list(self._cached_channels) + + try: + import uuid + cid = f"{self._session_id}::{str(uuid.uuid4())}" + user_agent = self._platform_config["user_agent"] + + # ── Step 1: distribution rights ────────────────────────────────── + rights_url = ( + self._provider_config.manifest.mpx.license_service_url + if self._provider_config and self._provider_config.manifest + else None + ) + if not rights_url: + raise RuntimeError( + "No license_service_url available – configuration discovery may have failed" + ) + + persona_token = self._ensure_authenticated() + auth_headers = { + "Authorization": f"Basic {persona_token}", + "Origin": "https://www.magenta.tv", + "Referer": "https://www.magenta.tv/", + } + + distribution_rights = fetch_distribution_rights( + http_manager=self._http, + rights_url=rights_url, + cid=cid, + user_agent=user_agent, + timeout=DEFAULT_REQUEST_TIMEOUT, + extra_headers=auth_headers, + ) + + def _dist_uri_to_int(uri) -> Optional[int]: + try: + return int(str(uri).rstrip("/").rsplit("/", 1)[-1]) + except (ValueError, AttributeError): + return None + + package_names = [ + DISTRIBUTION_PACKAGE_NAMES.get(numeric_id, f"Unknown package ({dist_id})") + for dist_id in distribution_rights + if (numeric_id := _dist_uri_to_int(dist_id)) is not None + ] + logger.info(f"Active subscription packages ({len(package_names)}):") + for pkg in package_names: + logger.info(f" · {pkg}") + + # ── Step 2: entitled-channels feed ─────────────────────────────── + feed_url = ( + self._endpoint_manager.get_endpoint("mpx_feed_entitledChannelsFeed") + if self._endpoint_manager + else None + ) or "https://feed.entertainment.tv.theplatform.eu/f/mdeprod/mdeprod-entitled-channels" + + tp_channels: List[TheplatformChannel] = fetch_entitled_channels_feed( + http_manager=self._http, + feed_url=feed_url, + distribution_rights=distribution_rights, + cid=cid, + user_agent=user_agent, + timeout=DEFAULT_REQUEST_TIMEOUT, + ) + + # ── Step 3: merge with station metadata (pre-fetched at init) ──── + channels: List[StreamingChannel] = [] + for tp_ch in tp_channels: + try: + meta = self.station_metadata.get(tp_ch.station_id, {}) + name = meta.get("title") or tp_ch.station_id + logo_url = meta.get("logo_url") + quality = meta.get("quality") + channel_number = ( + meta["channel_number"] + if meta.get("channel_number") is not None + else tp_ch.channel_number + ) + + magenta2_channel = Magenta2Channel( + name=name, + channel_id=tp_ch.release_pid, + logo_url=logo_url, + mode=MODE_LIVE, + content_type=CONTENT_TYPE_LIVE, + country=self._country, + raw_data=tp_ch.extra, + ) + streaming_channel = magenta2_channel.to_streaming_channel( + provider_name=self._provider_name + ) + streaming_channel.channel_number = channel_number + streaming_channel.quality = quality + streaming_channel.manifest = tp_ch.mpd_url + if tp_ch.hls_url: + streaming_channel.hls_url = tp_ch.hls_url + + self._live_manifest_cache[tp_ch.release_pid] = tp_ch.mpd_url + self._live_pid_cache[tp_ch.release_pid] = tp_ch.release_pid + + channels.append(streaming_channel) + except Exception as exc: + logger.warning(f"get_channels: skipping channel {tp_ch.station_id}: {exc}") + + logger.info( + f"Successfully fetched {len(channels)} entitled channels " + f"for country {self._country} " + f"({len(self.station_metadata)} stations with metadata)" + ) + channels.sort(key=lambda ch: (ch.channel_number is None, ch.channel_number or 0)) + self._cached_channels = channels + return channels + + except Exception as e: + raise Exception(f"Error fetching channels from Magenta2 API: {e}") + + def get_entitlement_token( + self, content_id: str, content_type: str = CONTENT_TYPE_LIVE + ) -> str: + """ + Request an entitlement token for *content_id* using the persona token + (Basic auth). + """ + self._ensure_authenticated() + headers = self._get_api_headers(require_auth=True) + payload = {"content_id": content_id, "content_type": content_type} + + url = ( + self._endpoint_manager.get_endpoint("entitlement") + if self._endpoint_manager + else "https://entitlement.p7s1.io/api/user/entitlement-token" + ) + + try: + logger.debug(f"Requesting entitlement token for: {content_id}") + response = self._http.post( + url, + operation="auth", + headers=headers, + json_data=payload, + timeout=DEFAULT_REQUEST_TIMEOUT, + ) + + if response.status_code == 400: + try: + error_data = response.json() + error_list = error_data if isinstance(error_data, list) else [error_data] + if error_list: + error = error_list[0] + code = error.get("code", error.get("errorCode", "UNKNOWN")) + msg = error.get("msg", error.get("message", "No error message provided")) + if code == ERROR_CODES["PLAYBACK_RESTRICTED"]: + raise Magenta2PlaybackRestrictedException( + f"Playback restricted for {content_id}: {msg}" + ) + else: + raise Exception(f"Entitlement error for {content_id} ({code}): {msg}") + except (json.JSONDecodeError, KeyError, IndexError) as e: + raise Exception( + f"Bad response for {content_id} (400), failed to parse error: {e}" + ) + + response.raise_for_status() + data = response.json() + + if "entitlement_token" in data: + return data["entitlement_token"] + elif "entitlementToken" in data: + return data["entitlementToken"] + elif "token" in data: + return data["token"] + else: + raise KeyError("No entitlement token found in response") + + except Magenta2PlaybackRestrictedException: + raise + except KeyError as e: + logger.error(f"No entitlement token in response for {content_id}: {e}") + raise Exception(f"No entitlement token in response for {content_id}: {e}") + except Exception as e: + logger.error(f"Error getting entitlement token for {content_id}: {e}") + raise Exception(f"Error getting entitlement token for {content_id}: {e}") + + def get_channel_playlist(self, channel_id: str, entitlement_token: str) -> Dict: + """Fetch playlist data (manifest URL, licence URL, format) for a channel.""" + if self._endpoint_manager and self._endpoint_manager.has_endpoint("channel_playlist"): + url = self._endpoint_manager.get_endpoint("channel_playlist").format( + channel_id=channel_id + ) + else: + url = f"https://api.magentatv.de/v1/channel/{channel_id}/playlist" + + headers = { + "Authorization": f"Bearer {entitlement_token}", + "User-Agent": self._platform_config["user_agent"], + "Accept": "application/json", + } + + try: + response = self._http.get( + url, + operation="manifest", + headers=headers, + timeout=DEFAULT_REQUEST_TIMEOUT, + ) + response.raise_for_status() + return response.json() + except Exception as e: + raise Exception(f"Error getting playlist for {channel_id}: {e}") + + def populate_streaming_data( + self, + channels: List[StreamingChannel], + max_retries: int = DEFAULT_MAX_RETRIES, + ) -> List[StreamingChannel]: + """ + Populate manifest URL, DRM config, and streaming format for each channel + by fetching an entitlement token and playlist. + + Channels that are restricted or repeatedly fail are silently dropped from + the returned list. + """ + self._ensure_authenticated() + successful_channels = [] + + for channel in channels: + retries = 0 + success = False + is_restricted = False + + while retries < max_retries and not success and not is_restricted: + try: + logger.debug( + f"Getting entitlement token for: {channel.name} (attempt {retries + 1})" + ) + entitlement_token = self.get_entitlement_token( + content_id=channel.channel_id, content_type=channel.content_type + ) + logger.debug(f"Getting playlist data for: {channel.name}") + playlist_data = self.get_channel_playlist( + channel.channel_id, entitlement_token + ) + + manifest_url = playlist_data.get( + "manifestUrl", playlist_data.get("manifest") + ) + license_url = playlist_data.get( + "licenseUrl", playlist_data.get("license") + ) + certificate_url = playlist_data.get( + "certificateUrl", playlist_data.get("certificate") + ) + streaming_format = playlist_data.get( + "streamingFormat", playlist_data.get("format", "dash") + ) + + if manifest_url: + channel.manifest = manifest_url + channel.cdm_type = DRM_SYSTEM_WIDEVINE + channel.cdm = f"pid={channel.channel_id}" + channel.license_url = license_url + channel.certificate_url = certificate_url + channel.streaming_format = streaming_format + logger.info(f"Streaming data populated for: {channel.name}") + successful_channels.append(channel) + success = True + else: + raise Exception("No manifest URL in response") + + except Magenta2PlaybackRestrictedException as e: + logger.warning(f"Playback restricted for {channel.name}: {e}") + is_restricted = True + + except Exception as e: + retries += 1 + if retries < max_retries: + logger.debug(f"Retry {retries}/{max_retries} for {channel.name}: {e}") + time.sleep(1) + else: + logger.error(f"Failed to get streaming data for {channel.name}: {e}") + + logger.info( + f"Streaming data population complete: " + f"{len(successful_channels)} successful, " + f"{len(channels) - len(successful_channels)} failed/restricted, " + f"{len(channels)} total" + ) + return successful_channels + + def invalidate_cache(self) -> None: + """Clear the in-memory channel and live-manifest caches.""" + self._cached_channels = None + self._live_manifest_cache.clear() + self._live_pid_cache.clear() + logger.debug("ChannelManager: caches cleared") + + # ------------------------------------------------------------------ # + # Internal helpers # + # ------------------------------------------------------------------ # + + def _fetch_station_metadata(self) -> Dict[str, Dict]: + """ + Fetch the unauthenticated channel-stations feed and return a lookup map + keyed by the theplatform Station URI. + + The URI is the key of the ``stations`` dict in each feed entry, e.g. + ``http://data.entertainment.tv.theplatform.eu/…/Station/265808936224``. + This matches ``listings[0].stationId`` in the entitled-channels feed, + which is what ``TheplatformChannel.station_id`` contains after parsing. + + Note: ``era$mediaPids["urn:theplatform:tv:location:any"]`` is a short + opaque PID used for other purposes — it is NOT the mapping key. + + Each value dict contains: + title – display name (" - Main" suffix stripped) + logo_url – scaled logo URL or None + quality – "HD", "SD", etc. + channel_number – display channel number or None + """ + metadata: Dict[str, Dict] = {} + try: + url = None + if self._endpoint_manager: + url = ( + self._endpoint_manager.get_endpoint("channel_stations") + or self._endpoint_manager.get_endpoint("channel_list") + ) + url = ( + url + or "https://feed.entertainment.tv.theplatform.eu/f/mdeprod/mdeprod-channel-stations-main" + ) + url += "?lang=short-de&sort=dt%24displayChannelNumber&range=1-1000" + + headers = self._get_api_headers(require_auth=False) + response = self._http.get( + url, operation="api", headers=headers, timeout=DEFAULT_REQUEST_TIMEOUT + ) + response.raise_for_status() + data = response.json() + + for entry in data.get("entries", []): + try: + stations = entry.get("stations", {}) + if not stations: + continue + station_uri = next(iter(stations.keys())) + station_info = stations[station_uri] + + title = ( + station_info.get("title") or entry.get("title", "") + ).replace(" - Main", "") + quality = station_info.get("dt$quality", "SD") + + logo_url = None + thumbnails = station_info.get("thumbnails", {}) + for logo_type in ["stationLogo", "stationLogoColored"]: + if logo_type in thumbnails: + original_url = thumbnails[logo_type].get("url") + if original_url: + logo_url = self._build_scaled_image_url(original_url) + break + + channel_number = entry.get("dt$displayChannelNumber") + + existing = metadata.get(station_uri) + if not existing or QUALITY_RANK.get(quality, 1) > QUALITY_RANK.get( + existing["quality"], 1 + ): + metadata[station_uri] = { + "title": title, + "logo_url": logo_url, + "quality": quality, + "channel_number": channel_number, + } + except Exception as exc: + logger.debug(f"_fetch_station_metadata: skipping entry: {exc}") + + logger.debug( + f"_fetch_station_metadata: built metadata for {len(metadata)} stations" + ) + except Exception as exc: + logger.warning( + f"_fetch_station_metadata: failed, channels will have no names: {exc}" + ) + + return metadata + + @staticmethod + def _extract_channel_id_from_entry(entry: Dict) -> Optional[str]: + """Extract the correct channel ID from ``era$mediaPids``.""" + try: + stations = entry.get("stations", {}) + if not stations: + return None + station_id = next(iter(stations.keys())) + station_info = stations[station_id] + era_media_pids = station_info.get("era$mediaPids", {}) + channel_id = era_media_pids.get("urn:theplatform:tv:location:any") + if channel_id: + logger.debug(f"Extracted channel ID from era$mediaPids: {channel_id}") + return channel_id + fallback_id = entry.get("guid") + if fallback_id: + logger.warning(f"Using fallback channel ID from guid: {fallback_id}") + return fallback_id + logger.warning("No channel ID found in entry") + return None + except Exception as e: + logger.warning(f"Error extracting channel ID from entry: {e}") + return None + + def _create_channel_from_entry( + self, entry: Dict, station_info: Dict, display_number + ) -> Optional[StreamingChannel]: + """Build a StreamingChannel from a raw feed entry.""" + try: + title = station_info.get("title") or entry.get("title", "Unknown Channel") + title = title.replace(" - Main", "") + + channel_id = self._extract_channel_id_from_entry(entry) + if not channel_id: + return None + + logo_url = None + thumbnails = station_info.get("thumbnails", {}) + for logo_type in ["stationLogo", "stationLogoColored"]: + if logo_type in thumbnails: + original_url = thumbnails[logo_type].get("url") + if original_url: + logo_url = self._build_scaled_image_url(original_url) + break + + magenta2_channel = Magenta2Channel( + name=title, + channel_id=channel_id, + logo_url=logo_url, + mode=MODE_LIVE, + content_type=CONTENT_TYPE_LIVE, + country=self._country, + raw_data=entry, + ) + streaming_channel = magenta2_channel.to_streaming_channel( + provider_name=self._provider_name + ) + streaming_channel.channel_number = display_number + streaming_channel.quality = station_info.get("dt$quality", "SD") + return streaming_channel + + except Exception as e: + logger.warning(f"Error creating channel from entry: {e}") + return None + + def _process_channel_stations_response_optimized( + self, response_data: Dict, prefer_highest_quality: bool = True + ) -> List[StreamingChannel]: + """Single-pass deduplication and channel construction from a raw feed response.""" + if "entries" not in response_data: + return [] + + best_entries: Dict = {} + channels: List[StreamingChannel] = [] + + for entry in response_data["entries"]: + try: + stations = entry.get("stations", {}) + if not stations: + continue + station_info = next(iter(stations.values())) + display_number = entry.get("dt$displayChannelNumber") + + if display_number is None: + channel = self._create_channel_from_entry( + entry, station_info, display_number + ) + if channel: + channels.append(channel) + continue + + quality = station_info.get("dt$quality", "SD") + current_rank = QUALITY_RANK.get(quality, 1) + + existing = best_entries.get(display_number) + if not existing: + best_entries[display_number] = (entry, station_info, current_rank) + else: + _, _, existing_rank = existing + if (prefer_highest_quality and current_rank > existing_rank) or ( + not prefer_highest_quality and current_rank < existing_rank + ): + best_entries[display_number] = (entry, station_info, current_rank) + + except Exception: + continue + + for display_number, (entry, station_info, _) in best_entries.items(): + channel = self._create_channel_from_entry(entry, station_info, display_number) + if channel: + channels.append(channel) + + return channels + + def _get_api_headers(self, require_auth: bool = False) -> Dict[str, str]: + """Build standard API request headers.""" + headers = { + "User-Agent": self._platform_config["user_agent"], + "Accept": "application/json", + "Content-Type": "application/json", + } + if require_auth: + persona_token = self._ensure_authenticated() + headers["Authorization"] = f"Basic {persona_token}" + return headers \ No newline at end of file diff --git a/lib/streaming_providers/providers/magenta2/playback_manager.py b/lib/streaming_providers/providers/magenta2/playback_manager.py new file mode 100644 index 0000000..6d3d76d --- /dev/null +++ b/lib/streaming_providers/providers/magenta2/playback_manager.py @@ -0,0 +1,221 @@ +# streaming_providers/providers/magenta2/playback_manager.py +# -*- coding: utf-8 -*- +""" +Routes manifest and DRM requests for the Magenta2 provider. + +Responsibilities +---------------- +- Serve live-channel manifests directly from the ChannelManager cache + (avoiding a SMIL round-trip for channels already fetched by get_channels). +- Build Widevine licence URLs directly for live channels using lib_theplatform. +- Delegate VOD and recording manifest/DRM requests to SmilManager. +- Inject ``smil_base_url`` from the recording-URL cache so callers (e.g. + DRMOperations) only need to pass a ``content_id``. +- Provide ``get_catchup_manifest`` via SmilManager. + +The class holds NO state of its own beyond references to the managers and +callbacks passed at construction time. All caches belong to ChannelManager. +""" +import base64 +from typing import Callable, Dict, List, Optional + +from ...base.models import DRMConfig +from ...base.utils.logger import logger +from .constants import CONTENT_TYPE_LIVE, MAGENTA2_FALLBACK_ACCOUNT_URI +from .endpoint_manager import EndpointManager +from .config_models import ProviderConfig +from .smil_manager import SmilManager +from .channel_manager import ChannelManager +from ..lib_theplatform import ( + extract_persona_jwt, + build_licence_url, + build_widevine_drm_config, +) + + +class PlaybackManager: + """ + Routes manifest and DRM requests for Magenta2 content. + + Parameters + ---------- + channel_manager: + The provider's ChannelManager instance (owns live-manifest/pid caches). + smil_manager: + The provider's SmilManager instance (handles VOD / recording SMIL). + endpoint_manager: + Populated EndpointManager after discovery. + provider_config: + ProviderConfig after discovery. + platform_config: + Platform-specific dict from MAGENTA2_PLATFORMS (user_agent, etc.). + auth_callback: + Callable[[], str] — returns a valid persona token (Basic-auth value). + recording_url_cache: + Shared dict (owned by the provider) mapping content_id → manifest_script + for recordings fetched via RecordingsManager. + """ + + def __init__( + self, + channel_manager: ChannelManager, + smil_manager: Optional[SmilManager], + endpoint_manager: Optional[EndpointManager], + provider_config: Optional[ProviderConfig], + platform_config: Dict, + auth_callback: Callable[[], str], + recording_url_cache: Dict[str, str], + ): + self._channel_manager = channel_manager + self._smil_manager = smil_manager + self._endpoint_manager = endpoint_manager + self._provider_config = provider_config + self._platform_config = platform_config + self._ensure_authenticated = auth_callback + self._recording_url_cache = recording_url_cache + + # ------------------------------------------------------------------ # + # Public API # + # ------------------------------------------------------------------ # + + def get_manifest( + self, content_id: str, content_type: str = CONTENT_TYPE_LIVE, **kwargs + ) -> Optional[str]: + """ + Return the MPD manifest URL for *content_id*. + + Live channels whose manifest was already fetched by get_channels() are + served directly from the ChannelManager cache — no SMIL round-trip. + VOD and recordings fall through to SmilManager. + """ + if content_type == CONTENT_TYPE_LIVE: + self._ensure_live_cache() + if content_id in self._channel_manager._live_manifest_cache: + logger.debug(f"get_manifest: cache hit for live channel {content_id}") + return self._channel_manager._live_manifest_cache[content_id] + + if not self._smil_manager: + raise RuntimeError("SmilManager not available") + self._inject_smil_base_url(content_id, kwargs) + return self._smil_manager.get_manifest(content_id, content_type, **kwargs) + + def get_drm( + self, content_id: str, content_type: str = CONTENT_TYPE_LIVE, **kwargs + ) -> List[DRMConfig]: + """ + Return DRM configuration for *content_id*. + + For live channels whose releasePid is cached, the Widevine licence URL + is built directly using lib_theplatform — no SMIL fetch needed. + VOD and recordings fall through to SmilManager. + """ + if content_type == CONTENT_TYPE_LIVE: + self._ensure_live_cache() + if content_id in self._channel_manager._live_pid_cache: + return self._build_live_drm(content_id) + + if not self._smil_manager: + raise RuntimeError("SmilManager not available") + self._inject_smil_base_url(content_id, kwargs) + return self._smil_manager.get_drm(content_id, content_type, **kwargs) + + def get_catchup_manifest( + self, channel_id: str, start_time: int, end_time: int, **kwargs + ) -> Optional[str]: + """Return a catchup manifest URL via SmilManager.""" + if not self._smil_manager: + raise RuntimeError("SmilManager not available") + return self._smil_manager.get_catchup_manifest( + channel_id, start_time, end_time, **kwargs + ) + + # ------------------------------------------------------------------ # + # Internal helpers # + # ------------------------------------------------------------------ # + + def _ensure_live_cache(self) -> None: + """ + Bootstrap the live-manifest / live-pid caches on demand by calling + ChannelManager.get_channels() if they are empty. + """ + if not self._channel_manager._live_manifest_cache: + logger.debug("Live channel cache is empty — auto-populating via get_channels()") + self._channel_manager.get_channels() + + def _build_live_drm(self, content_id: str) -> List[DRMConfig]: + """ + Build a Widevine DRMConfig directly for a live channel, bypassing SMIL. + """ + release_pid = content_id + logger.debug( + f"get_drm: building licence directly for live channel (releasePid: {release_pid})" + ) + try: + persona_token = self._ensure_authenticated() + raw_jwt = extract_persona_jwt(persona_token) + if not raw_jwt: + logger.error("get_drm: failed to extract persona JWT") + return [] + + widevine_endpoint = ( + self._endpoint_manager.get_endpoint("widevine_license") + if self._endpoint_manager + else None + ) + if not widevine_endpoint: + logger.error("get_drm: no widevine_license endpoint available") + return [] + + account_uri = ( + self._provider_config.manifest.mpx.get_account_uri() + if self._provider_config and self._provider_config.manifest + else None + ) or MAGENTA2_FALLBACK_ACCOUNT_URI + + licence_url = build_licence_url( + widevine_endpoint=widevine_endpoint, + release_pid=release_pid, + persona_jwt=raw_jwt, + account_uri=account_uri, + ) + return [ + build_widevine_drm_config( + licence_url=licence_url, + user_agent=self._platform_config["user_agent"], + ) + ] + except Exception as exc: + logger.error(f"get_drm: direct licence build failed for {content_id}: {exc}") + return [] + + def _inject_smil_base_url(self, content_id: str, kwargs: Dict) -> None: + """ + If a recording manifest_script URL is cached for *content_id*, inject it + as ``smil_base_url`` into *kwargs* so SmilManager can use it without the + caller needing to know about it. + """ + if "smil_base_url" not in kwargs and content_id in self._recording_url_cache: + kwargs["smil_base_url"] = self._recording_url_cache[content_id] + logger.debug(f"Injected smil_base_url from recording cache for {content_id}") + + @staticmethod + def extract_persona_jwt_from_token(persona_token: str) -> Optional[str]: + """ + Decode a Base64-encoded persona token and extract the raw JWT portion. + + The persona token format is: ``Base64(account_uri + ":" + persona_jwt)``. + """ + try: + decoded = base64.b64decode(persona_token).decode("utf-8") + last_colon_index = decoded.rfind(":") + if last_colon_index == -1: + logger.error("No colon found in decoded persona token") + return None + persona_jwt = decoded[last_colon_index + 1:] + if not persona_jwt.startswith("eyJ"): + logger.error("Extracted token doesn't look like a JWT") + return None + return persona_jwt + except Exception as e: + logger.error(f"Error extracting persona JWT token: {e}") + return None \ 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 7116fa0..99814ec 100644 --- a/lib/streaming_providers/providers/magenta2/provider.py +++ b/lib/streaming_providers/providers/magenta2/provider.py @@ -1,12 +1,21 @@ # streaming_providers/providers/magenta2/provider.py # -*- coding: utf-8 -*- -import base64 -import json -import time +""" +Magenta2 streaming provider. + +This module contains only lifecycle, authentication, and the thin public API +that delegates to the three domain managers: + + ChannelManager – channel discovery, entitlement, streaming-data population + PlaybackManager – manifest / DRM routing (live fast-path + SMIL fallback) + VodManager – VOD catalogue browsing + RecordingsManager – nPVR (list / delete / manifest) + SmilManager – SMIL-based manifest and DRM for VOD / recordings +""" import uuid from datetime import datetime, timedelta +from typing import Any, ClassVar, Dict, List, Optional, Tuple, cast from urllib.parse import quote -from typing import Any, ClassVar, Dict, List, Optional, Tuple from ...base.models import DRMConfig, StreamingChannel, Event from ...base.models.auth import AuthState @@ -18,7 +27,9 @@ from .recordings_manager import RecordingsManager from .smil_manager import SmilManager from .vod_manager import VodManager from .auth import Magenta2Authenticator, Magenta2Credentials, Magenta2UserCredentials -from .config_models import ProviderConfig +from .channel_manager import ChannelManager +from .playback_manager import PlaybackManager +from .config_models import BootstrapConfig, ProviderConfig from .constants import ( CONTENT_TYPE_LIVE, DEFAULT_COUNTRY, @@ -26,36 +37,22 @@ from .constants import ( DEFAULT_MAX_RETRIES, DEFAULT_PLATFORM, DEFAULT_REQUEST_TIMEOUT, - DISTRIBUTION_PACKAGE_NAMES, - DRM_SYSTEM_WIDEVINE, - ERROR_CODES, MAGENTA2_CLIENT_IDS, - MAGENTA2_FALLBACK_ACCOUNT_URI, MAGENTA2_LOGO, MAGENTA2_PLATFORMS, - MODE_LIVE, - QUALITY_RANK, SUPPORTED_COUNTRIES, ) from .discovery import DiscoveryService -from ..lib_theplatform import ( - TheplatformChannel, - fetch_distribution_rights, - fetch_entitled_channels_feed, - extract_persona_jwt, - build_licence_url, - build_widevine_drm_config, -) from .endpoint_manager import EndpointManager -from .models import Magenta2Channel, Magenta2PlaybackRestrictedException -from .token_flow_manager import PersonaResult +from .models import Magenta2PlaybackRestrictedException # noqa: F401 – re-exported +from .auth_bridge import AuthBridge class Magenta2Provider(StreamingProvider): PROVIDER_LABEL: ClassVar[str] = "Magenta TV 2.0" PROVIDER_LOGO: ClassVar[str] = MAGENTA2_LOGO """ - Magenta2 streaming provider implementation with enhanced dynamic discovery + Magenta2 streaming provider implementation with enhanced dynamic discovery. """ def __init__( @@ -68,9 +65,6 @@ class Magenta2Provider(StreamingProvider): username: Optional[str] = None, password: Optional[str] = None, ): - """ - Initialize Magenta2 provider with enhanced discovery - """ super().__init__(country=country) if country not in SUPPORTED_COUNTRIES: @@ -84,25 +78,23 @@ class Magenta2Provider(StreamingProvider): ) self.terminal_type = self.platform_config["terminal_type"] - # Generate session ID, device ID, and serial number. - # serial_number is stable for the lifetime of this provider instance. + # Stable UUIDs for the lifetime of this provider instance. self.session_id = self._generate_uuid() self.device_id = self._generate_uuid() self.serial_number = self._generate_uuid() - # Setup proxy configuration + # ── Proxy ──────────────────────────────────────────────────────────── self.proxy_config = ( proxy_config or (ProxyConfig.from_url(proxy_url) if proxy_url else None) or self._load_proxy_from_manager(config_dir) ) - if self.proxy_config: logger.info("Using proxy configuration for Magenta2") else: logger.debug("No proxy configuration found for Magenta2") - # Create HTTP manager + # ── HTTP manager ───────────────────────────────────────────────────── self.http_manager = HTTPManagerFactory.create_for_provider( provider_name="magenta2", proxy_config=self.proxy_config, @@ -111,7 +103,7 @@ class Magenta2Provider(StreamingProvider): max_retries=DEFAULT_MAX_RETRIES, ) - # Initialize discovery service + # ── Discovery service ──────────────────────────────────────────────── self.discovery_service = DiscoveryService( platform=platform, terminal_type=self.terminal_type, @@ -121,238 +113,161 @@ class Magenta2Provider(StreamingProvider): proxy_config=self.proxy_config, ) - # Initialize endpoint manager (will be populated after discovery) - self.endpoint_manager: Optional[EndpointManager] = None - self.provider_config: Optional[ProviderConfig] = None - self._vod_manager: Optional[VodManager] = None - self._recordings_manager: Optional[RecordingsManager] = None - self._smil_manager: Optional[SmilManager] = None + # Both are assigned concrete values by _perform_configuration_discovery() + # (or _create_fallback_configuration()) before __init__ returns. We use + # cast(None) as a typed sentinel so the class-level annotation stays + # non-Optional -- callers and other methods see EndpointManager / + # ProviderConfig directly, with no Optional unwrapping needed. + self.endpoint_manager = cast(EndpointManager, cast(object, None)) + self.provider_config = cast(ProviderConfig, cast(object, None)) - # 🚨 INITIALIZE AUTHENTICATOR FIRST (with minimal config) - # Use fallback client IDs initially + # ── Authenticator (minimal config; updated after discovery) ────────── fallback_client_id = MAGENTA2_CLIENT_IDS.get( platform, MAGENTA2_CLIENT_IDS[DEFAULT_PLATFORM] ) if username and password: - # Use user credentials for complete authentication flow - credentials = Magenta2UserCredentials( - client_id=fallback_client_id, # Use fallback initially - platform=platform, - country=country, - device_id=self.device_id, - username=username, - password=password, + credentials: Magenta2Credentials | Magenta2UserCredentials = ( + Magenta2UserCredentials( + client_id=fallback_client_id, + platform=platform, + country=country, + device_id=self.device_id, + username=username, + password=password, + ) ) logger.info("Using user credentials for authentication") else: - # Use client credentials for TAA-only flow credentials = Magenta2Credentials( - client_id=fallback_client_id, # Use fallback initially + client_id=fallback_client_id, platform=platform, country=country, device_id=self.device_id, ) logger.info("Using client credentials for authentication") - # Create authenticator with minimal configuration self.authenticator = Magenta2Authenticator( country=country, platform=platform, config_dir=config_dir, http_manager=self.http_manager, credentials=credentials, - endpoints={}, # Empty initially, will be updated after discovery - client_model=f"ftv-{platform}", # Use fallback initially - device_model=f"{platform.upper()}_FTV", # Use fallback initially - sam3_client_id=fallback_client_id, # Use fallback initially + endpoints={}, + client_model=f"ftv-{platform}", + device_model=f"{platform.upper()}_FTV", + sam3_client_id=fallback_client_id, session_id=self.session_id, device_id=self.device_id, provider_config=None, ) - # 🚨 NOW PERFORM CONFIGURATION DISCOVERY (authenticator exists) + # ── Configuration discovery ────────────────────────────────────────── + # _perform_configuration_discovery always assigns self.endpoint_manager + # and self.provider_config (either from discovery or from the fallback). + # Re-raise so callers see the error; do NOT silently swallow it. try: self._perform_configuration_discovery() except Exception as e: logger.error(f"Configuration discovery failed: {e}") raise - # 🚨 UPDATE AUTHENTICATOR WITH DISCOVERED CONFIG - if self.provider_config: - # Update authenticator with discovered client_id and models - self.authenticator.provider_config = self.provider_config # ✅ Store the config - logger.info("✓ ProviderConfig stored in authenticator") + # ── Update authenticator with discovered config ─────────────────────── + self._configure_authenticator_from_discovery(self.provider_config, self.endpoint_manager) - # Also update TokenFlowManager if it exists - if ( - hasattr(self.authenticator, "token_flow_manager") - and self.authenticator.token_flow_manager - ): - self.authenticator.token_flow_manager.provider_config = self.provider_config - logger.info("✓ ProviderConfig also stored in TokenFlowManager") - - if self.provider_config.bootstrap.sam3_client_id: - # Use public method if available, otherwise update directly - if hasattr(self.authenticator, "update_sam3_client_id"): - self.authenticator.update_sam3_client_id( - self.provider_config.bootstrap.sam3_client_id - ) - else: - self.authenticator._sam3_client_id = ( - self.provider_config.bootstrap.sam3_client_id - ) - - if self.authenticator.credentials: - self.authenticator.credentials.client_id = ( - self.provider_config.bootstrap.sam3_client_id - ) - logger.debug( - f"Updated authenticator with SAM3 client ID: {self.provider_config.bootstrap.sam3_client_id}" - ) - - if self.provider_config.bootstrap.client_model: - if hasattr(self.authenticator, "update_client_model"): - self.authenticator.update_client_model( - self.provider_config.bootstrap.client_model - ) - else: - self.authenticator._client_model = self.provider_config.bootstrap.client_model - logger.debug( - f"Updated authenticator with client model: {self.provider_config.bootstrap.client_model}" - ) - - if self.provider_config.bootstrap.device_model: - if hasattr(self.authenticator, "update_device_model"): - self.authenticator.update_device_model( - self.provider_config.bootstrap.device_model - ) - else: - self.authenticator._device_model = self.provider_config.bootstrap.device_model - logger.debug( - f"Updated authenticator with device model: {self.provider_config.bootstrap.device_model}" - ) - - # Update authenticator with device token and MPX account if available - if self.provider_config.manifest: - device_token = self.provider_config.get_device_token() - authorize_tokens_url = self.provider_config.get_authorize_tokens_url() - - if device_token: - self.authenticator.set_device_token(device_token, authorize_tokens_url) - logger.debug("Device token configured in authenticator") - - # CRITICAL: Pass MPX account PID for account URI construction - if self.provider_config.manifest.mpx.account_pid: - self.authenticator.set_mpx_account_pid( - self.provider_config.manifest.mpx.account_pid - ) - logger.debug( - f"MPX account PID configured: {self.provider_config.manifest.mpx.account_pid}" - ) - - # Pass OpenID configuration if available - if self.provider_config.openid: - self.authenticator.set_openid_config(self.provider_config.openid.raw_data) - - # Update authenticator with discovered endpoints using public methods - if self.endpoint_manager: - # Get all endpoints - all_endpoints = { - name: info.url - for name, info in self.endpoint_manager.get_all_endpoints().items() - } - - # Update authenticator with discovered endpoints using public method - if hasattr(self.authenticator, "update_dynamic_endpoints"): - self.authenticator.update_dynamic_endpoints(all_endpoints) - logger.info(f"✓ Updated authenticator with {len(all_endpoints)} endpoints") - elif hasattr(self.authenticator, "update_endpoints"): - self.authenticator.update_endpoints(all_endpoints) - logger.info(f"✓ Updated authenticator with {len(all_endpoints)} endpoints") - else: - logger.warning("No public method available to update endpoints") - - # Specifically update SAM3 client with QR code URL - qr_url = self.endpoint_manager.get_endpoint("login_qr_code") - if qr_url and hasattr(self.authenticator, "update_sam3_qr_code_url"): - success = self.authenticator.update_sam3_qr_code_url(qr_url) - if success: - logger.info("✓ Successfully updated SAM3 client with QR code URL") - else: - logger.warning("✗ Failed to update SAM3 client with QR code URL") - - # Initialize VodManager now that config is discovered - if self.provider_config and self.endpoint_manager: - 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.session_id, - serial_number=self.serial_number, - auth_headers_callback=self._vod_auth_headers, - ) - logger.info("✓ VodManager initialized with discovered config") - - self._recordings_manager = RecordingsManager( - http_manager=self.http_manager, - provider_name=self.provider_name, - provider_config=self.endpoint_manager.config, - auth_headers_callback=self._pvr_auth_headers, - ) - logger.info("✓ RecordingsManager initialized with discovered config") - - self._smil_manager = SmilManager( - http_manager=self.http_manager, - provider_name=self.provider_name, - session_id=self.session_id, - call_id_callback=self._generate_call_id, - auth_callback=self._ensure_authenticated, - platform_config=self.platform_config, - endpoint_manager=self.endpoint_manager, - provider_config=self.endpoint_manager.config, - vod_manager=self._vod_manager, - ) - logger.info("✓ SmilManager initialized with discovered config") - - # Initialize auth tokens (lazy - populated on first use) - self.device_token = None - self._persona_cache: Optional[PersonaResult] = None - # GUID → manifest_script URL; populated by get_recordings() so that - # get_manifest() / get_drm() can inject smil_base_url automatically - # without requiring callers (e.g. DRMOperations) to know about it. + # ── recording content_id → manifest_script; shared with PlaybackManager ── self._recording_url_cache: Dict[str, str] = {} - # release_pid → mpd_url / release_pid; keyed by release_pid because - # channel_id is now set to tp_ch.release_pid in get_channels(), so - # content_id arriving in get_manifest() / get_drm() is a release_pid. - self._live_manifest_cache: Dict[str, str] = {} - self._live_pid_cache: Dict[str, str] = {} - # Full StreamingChannel list from the last successful get_channels() call. - # None until get_channels() has run at least once. - self._cached_channels: Optional[List[StreamingChannel]] = None - # station_uri → {title, logo_url, quality}; fetched once at init time - # from the unauthenticated channel-stations feed. get_channels() reads - # this dict instead of making a fresh network call every time. - # channel_id (= release_pid) comes from the entitled-channels feed - # (authenticated), NOT from this call — this call only supplies display - # metadata (name, logo, quality, channel number). - self._station_metadata: Dict[str, Dict] = self._fetch_station_metadata() + # ── Domain managers ────────────────────────────────────────────────── + self._vod_manager: Optional[VodManager] = None + self._recordings_manager: Optional[RecordingsManager] = None + self._smil_manager: Optional[SmilManager] = 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.session_id, + serial_number=self.serial_number, + auth_headers_callback=self._vod_auth_headers, + ) + logger.info("✓ VodManager initialized") + + self._recordings_manager = RecordingsManager( + http_manager=self.http_manager, + provider_name=self.provider_name, + provider_config=self.endpoint_manager.config, + auth_headers_callback=self._pvr_auth_headers, + ) + logger.info("✓ RecordingsManager initialized") + + self._smil_manager = SmilManager( + http_manager=self.http_manager, + provider_name=self.provider_name, + session_id=self.session_id, + call_id_callback=self._generate_call_id, + auth_callback=self._ensure_authenticated, + platform_config=self.platform_config, + endpoint_manager=self.endpoint_manager, + provider_config=self.endpoint_manager.config, + vod_manager=self._vod_manager, + ) + logger.info("✓ SmilManager initialized") + + self._channel_manager = ChannelManager( + http_manager=self.http_manager, + provider_name=self.provider_name, + country=country, + platform_config=self.platform_config, + session_id=self.session_id, + serial_number=self.serial_number, + endpoint_manager=self.endpoint_manager, + provider_config=self.provider_config, + auth_callback=self._ensure_authenticated, + build_scaled_image_url_callback=self._build_scaled_image_url, + ) + logger.info("✓ ChannelManager initialized") + + self._playback_manager = PlaybackManager( + channel_manager=self._channel_manager, + smil_manager=self._smil_manager, + endpoint_manager=self.endpoint_manager, + provider_config=self.provider_config, + platform_config=self.platform_config, + auth_callback=self._ensure_authenticated, + recording_url_cache=self._recording_url_cache, + ) + logger.info("✓ PlaybackManager initialized") + + # ── Auth bridge ─────────────────────────────────────────────────────── + self.device_token = None + self._auth = AuthBridge( + authenticator=self.authenticator, + provider_name=self.provider_name, + country=self.country, + platform=self.platform, + platform_config=self.platform_config, + provider_config=self.provider_config, + session_id=self.session_id, + serial_number=self.serial_number, + generate_call_id=self._generate_call_id, + ) logger.info("Magenta2 provider initialization completed successfully") + # ------------------------------------------------------------------ # + # Static / utility # + # ------------------------------------------------------------------ # + @staticmethod def _generate_uuid() -> str: - """Generate a UUID string. Used for session IDs, device IDs, and call IDs.""" return str(uuid.uuid4()) def _generate_call_id(self) -> str: - """Generate a fresh call ID for each individual request.""" return self._generate_uuid() def _load_proxy_from_manager(self, config_dir: Optional[str]) -> Optional[ProxyConfig]: - """Load proxy configuration from ProxyConfigManager""" try: proxy_manager = ProxyConfigManager(config_dir) return proxy_manager.get_proxy_config("magenta2", self.country) @@ -360,121 +275,9 @@ class Magenta2Provider(StreamingProvider): logger.warning(f"Could not load proxy from ProxyConfigManager: {e}") return None - def _perform_configuration_discovery(self) -> None: - """ - Perform complete configuration discovery using discovery service - """ - logger.info("Performing Magenta2 configuration discovery") - - try: - # Perform discovery - self.provider_config = self.discovery_service.discover_provider_config() - - if not self.provider_config or not self.provider_config.is_complete: - logger.warning("Configuration discovery incomplete, some features may not work") - - # Initialize endpoint manager with discovered configuration - self.endpoint_manager = EndpointManager(self.provider_config) - - # DEBUG: Check if QR code endpoint is discovered - if self.endpoint_manager: - qr_url = self.endpoint_manager.get_endpoint("login_qr_code") - if qr_url: - logger.info(f"✓ QR code endpoint discovered in endpoint manager: {qr_url}") - - # PROPER FIX: Use public method to update SAM3 client - if hasattr(self.authenticator, "update_sam3_qr_code_url"): - success = self.authenticator.update_sam3_qr_code_url(qr_url) - if success: - logger.info("✓ Successfully updated SAM3 client with QR code URL") - else: - logger.warning("✗ Failed to update SAM3 client with QR code URL") - - # Also debug the current status - if hasattr(self.authenticator, "get_sam3_client_status"): - status = self.authenticator.get_sam3_client_status() - logger.debug(f"SAM3 client status after update: {status}") - else: - logger.warning("✗ QR code endpoint NOT found in endpoint manager") - - if self.provider_config and self.provider_config.manifest: - device_token = self.provider_config.get_device_token() - authorize_tokens_url = self.provider_config.get_authorize_tokens_url() - - if device_token: - logger.info(f"✓ Device token discovered (length: {len(device_token)})") - else: - logger.warning("⚠️ No device token found in manifest") - - if authorize_tokens_url: - logger.info(f"✓ Line auth endpoint discovered: {authorize_tokens_url}") - else: - logger.warning("⚠️ No authorize tokens URL found in manifest") - - if self.provider_config.manifest.mpx.account_pid: - logger.info( - f"✓ MPX account PID discovered: {self.provider_config.manifest.mpx.account_pid}" - ) - - # Validate critical endpoints - missing_endpoints = self.endpoint_manager.validate_critical_endpoints() - if missing_endpoints: - logger.warning(f"Missing critical endpoints: {missing_endpoints}") - else: - logger.info("All critical endpoints available") - - # Log discovery statistics - stats = self.endpoint_manager.get_stats() - logger.info( - f"Discovery complete: {stats['dynamic_endpoints']} dynamic endpoints, " - f"{stats['fallback_endpoints']} fallback endpoints, " - f"complete: {stats['is_complete']}" - ) - - except Exception as e: - logger.error(f"Configuration discovery failed: {e}") - self._create_fallback_configuration() - raise - - def _create_fallback_configuration(self) -> None: - """Create fallback configuration when discovery fails""" - logger.warning("Creating fallback configuration") - - from .config_models import BootstrapConfig, ProviderConfig - - bootstrap_config = BootstrapConfig( - client_model=f"ftv-{self.platform}", - device_model=f"{self.platform.upper()}_FTV", - ) - - self.provider_config = ProviderConfig(bootstrap=bootstrap_config) - self.endpoint_manager = EndpointManager(self.provider_config) - - logger.info("Fallback configuration created") - - def _get_dcm_headers(self) -> Dict[str, str]: - """Get headers for DCM requests""" - return { - "User-Agent": self.platform_config["user_agent"], - "Content-Type": "application/json", - "Accept": "application/json", - "x-dt-session-id": self.session_id, - "x-dt-call-id": self._generate_call_id(), - } - - def _get_api_headers(self, require_auth: bool = False) -> Dict[str, str]: - """Get headers for API requests""" - headers = { - "User-Agent": self.platform_config["user_agent"], - "Accept": "application/json", - "Content-Type": "application/json", - } - - if require_auth: - persona_token = self._ensure_authenticated() - headers["Authorization"] = f"Basic {persona_token}" - - return headers + # ------------------------------------------------------------------ # + # Provider properties # + # ------------------------------------------------------------------ # @property def provider_name(self) -> str: @@ -502,7 +305,6 @@ class Magenta2Provider(StreamingProvider): @property def catchup_window(self) -> int: - # This provider offers 4 hours of catchup return 4 @property @@ -517,47 +319,203 @@ class Magenta2Provider(StreamingProvider): def token_scopes(self) -> List[str]: return ["yo_digital", "tvhubs", "taa", "persona"] + # ------------------------------------------------------------------ # + # Configuration discovery # + # ------------------------------------------------------------------ # + + def _perform_configuration_discovery(self) -> None: + """ + Run discovery and initialise EndpointManager. + + Always sets both self.provider_config and self.endpoint_manager — either + from the live discovery result or from the fallback (via + _create_fallback_configuration). Callers may therefore assert both are + non-None after this method returns without raising. + """ + logger.info("Performing Magenta2 configuration discovery") + try: + self.provider_config = self.discovery_service.discover_provider_config() + + if not self.provider_config or not self.provider_config.is_complete: + logger.warning("Configuration discovery incomplete, some features may not work") + + self.endpoint_manager = EndpointManager(self.provider_config) + + qr_url = self.endpoint_manager.get_endpoint("login_qr_code") + if qr_url: + logger.info(f"✓ QR code endpoint discovered: {qr_url}") + if hasattr(self.authenticator, "update_sam3_qr_code_url"): + success = self.authenticator.update_sam3_qr_code_url(qr_url) + logger.info( + "✓ SAM3 client updated with QR code URL" + if success + else "✗ Failed to update SAM3 client with QR code URL" + ) + if hasattr(self.authenticator, "get_sam3_client_status"): + logger.debug( + f"SAM3 client status: {self.authenticator.get_sam3_client_status()}" + ) + else: + logger.warning("✗ QR code endpoint NOT found") + + if self.provider_config and self.provider_config.manifest: + device_token = self.provider_config.get_device_token() + authorize_tokens_url = self.provider_config.get_authorize_tokens_url() + if device_token: + logger.info(f"✓ Device token discovered (length: {len(device_token)})") + else: + logger.warning("⚠️ No device token found in manifest") + if authorize_tokens_url: + logger.info(f"✓ Line auth endpoint discovered: {authorize_tokens_url}") + else: + logger.warning("⚠️ No authorize tokens URL found in manifest") + if self.provider_config.manifest.mpx.account_pid: + logger.info( + f"✓ MPX account PID discovered: " + f"{self.provider_config.manifest.mpx.account_pid}" + ) + + missing_endpoints = self.endpoint_manager.validate_critical_endpoints() + if missing_endpoints: + logger.warning(f"Missing critical endpoints: {missing_endpoints}") + else: + logger.info("All critical endpoints available") + + stats = self.endpoint_manager.get_stats() + logger.info( + f"Discovery complete: {stats['dynamic_endpoints']} dynamic endpoints, " + f"{stats['fallback_endpoints']} fallback endpoints, " + f"complete: {stats['is_complete']}" + ) + + except Exception as e: + logger.error(f"Configuration discovery failed: {e}") + self._create_fallback_configuration() + raise + + def _create_fallback_configuration(self) -> None: + """Create minimal fallback configuration when discovery fails.""" + logger.warning("Creating fallback configuration") + bootstrap_config = BootstrapConfig( + client_model=f"ftv-{self.platform}", + device_model=f"{self.platform.upper()}_FTV", + ) + self.provider_config = ProviderConfig(bootstrap=bootstrap_config) + self.endpoint_manager = EndpointManager(self.provider_config) + logger.info("Fallback configuration created") + + def _configure_authenticator_from_discovery( + self, + cfg: ProviderConfig, + endpoint_manager: EndpointManager, + ) -> None: + """ + Push discovered config values (client_id, models, device token, MPX PID, + endpoints) into the authenticator and its TokenFlowManager. + + Parameters are passed explicitly (not read from self) so the type checker + knows they are non-None. + """ + self.authenticator.provider_config = cfg + logger.info("✓ ProviderConfig stored in authenticator") + + if ( + hasattr(self.authenticator, "token_flow_manager") + and self.authenticator.token_flow_manager + ): + self.authenticator.token_flow_manager.provider_config = cfg + logger.info("✓ ProviderConfig stored in TokenFlowManager") + + if cfg.bootstrap.sam3_client_id: + if hasattr(self.authenticator, "update_sam3_client_id"): + self.authenticator.update_sam3_client_id(cfg.bootstrap.sam3_client_id) + else: + self.authenticator._sam3_client_id = cfg.bootstrap.sam3_client_id + if self.authenticator.credentials: + self.authenticator.credentials.client_id = cfg.bootstrap.sam3_client_id + logger.debug( + f"Updated authenticator SAM3 client ID: {cfg.bootstrap.sam3_client_id}" + ) + + if cfg.bootstrap.client_model: + if hasattr(self.authenticator, "update_client_model"): + self.authenticator.update_client_model(cfg.bootstrap.client_model) + else: + self.authenticator._client_model = cfg.bootstrap.client_model + logger.debug(f"Updated authenticator client model: {cfg.bootstrap.client_model}") + + if cfg.bootstrap.device_model: + if hasattr(self.authenticator, "update_device_model"): + self.authenticator.update_device_model(cfg.bootstrap.device_model) + else: + self.authenticator._device_model = cfg.bootstrap.device_model + logger.debug(f"Updated authenticator device model: {cfg.bootstrap.device_model}") + + if cfg.manifest: + device_token = cfg.get_device_token() + authorize_tokens_url = cfg.get_authorize_tokens_url() + if device_token: + self.authenticator.set_device_token(device_token, authorize_tokens_url) + logger.debug("Device token configured in authenticator") + if cfg.manifest.mpx.account_pid: + self.authenticator.set_mpx_account_pid(cfg.manifest.mpx.account_pid) + logger.debug(f"MPX account PID configured: {cfg.manifest.mpx.account_pid}") + if cfg.openid: + self.authenticator.set_openid_config(cfg.openid.raw_data) + + all_endpoints = { + name: info.url + for name, info in endpoint_manager.get_all_endpoints().items() + } + if hasattr(self.authenticator, "update_dynamic_endpoints"): + self.authenticator.update_dynamic_endpoints(all_endpoints) + logger.info(f"✓ Updated authenticator with {len(all_endpoints)} endpoints") + elif hasattr(self.authenticator, "update_endpoints"): + self.authenticator.update_endpoints(all_endpoints) + logger.info(f"✓ Updated authenticator with {len(all_endpoints)} endpoints") + else: + logger.warning("No public method available to update endpoints") + + qr_url = endpoint_manager.get_endpoint("login_qr_code") + if qr_url and hasattr(self.authenticator, "update_sam3_qr_code_url"): + success = self.authenticator.update_sam3_qr_code_url(qr_url) + logger.info( + "✓ SAM3 client updated with QR code URL" + if success + else "✗ Failed to update SAM3 client with QR code URL" + ) + def get_discovery_status(self) -> Dict[str, Any]: - """Get discovery and configuration status""" + """Return discovery and endpoint statistics.""" if not self.discovery_service: return {"error": "Discovery service not initialized"} - status = self.discovery_service.get_discovery_status() - - if self.endpoint_manager: - status["endpoints"] = self.endpoint_manager.get_stats() - + status["endpoints"] = self.endpoint_manager.get_stats() return status def refresh_configuration(self, force: bool = False) -> bool: - """ - Refresh provider configuration - - Args: - force: Force refresh even if cache is valid - - Returns: - bool: True if refresh successful - """ + """Re-run configuration discovery.""" try: logger.info("Refreshing provider configuration") - new_config = self.discovery_service.discover_provider_config(force_refresh=force) if new_config and new_config.is_complete: self.provider_config = new_config self.endpoint_manager = EndpointManager(new_config) - # Update authenticator with new config if new_config.manifest: - device_token = new_config.manifest.raw_data.get("deviceToken") - authorize_tokens_url = new_config.manifest.raw_data.get("authorizeTokensUrl") + device_token: Any = new_config.manifest.raw_data.get("deviceToken") + authorize_tokens_url: Any = new_config.manifest.raw_data.get( + "authorizeTokensUrl" + ) if device_token: self.authenticator.set_device_token(device_token, authorize_tokens_url) - if new_config.manifest.mpx.account_pid: - self.authenticator.set_mpx_account_pid(new_config.manifest.mpx.account_pid) + self.authenticator.set_mpx_account_pid( + new_config.manifest.mpx.account_pid + ) + self._auth.update_provider_config(new_config) logger.info("Configuration refresh successful") return True else: @@ -569,18 +527,9 @@ class Magenta2Provider(StreamingProvider): return False def register_device(self) -> bool: - """ - Perform device registration and authentication - Useful for initial setup or device token refresh - """ + """Perform device registration / authentication.""" try: logger.info("Performing device registration") - - if not self.authenticator: - logger.error("Authenticator not available for device registration") - return False - - # PROPER: Use public method instead of checking protected attribute if hasattr(self.authenticator, "perform_device_authentication"): success = self.authenticator.perform_device_authentication() if success: @@ -592,303 +541,92 @@ class Magenta2Provider(StreamingProvider): else: logger.warning("Device authentication not supported in current authenticator") return False - except Exception as e: logger.error(f"Device registration failed: {e}") return False + # ------------------------------------------------------------------ # + # Authentication — thin delegates to AuthBridge # + # ------------------------------------------------------------------ # + def get_persona_token(self, force_refresh: bool = False) -> str: - """Get persona token with accurate expiry-based caching""" - # Check in-memory cache first - if not force_refresh and self._persona_cache and self._persona_cache.success: - current_time = time.time() - # Check if cached token is still valid (with 1-minute buffer) - if current_time < (self._persona_cache.expires_at - 60): - logger.debug( - f"Using in-memory cached persona token (expires at {time.ctime(self._persona_cache.expires_at)})" - ) - return self._persona_cache.persona_token - else: - # Cache expired - self._persona_cache = None - logger.debug("In-memory persona cache expired") - - # Get from TokenFlowManager (now returns PersonaResult with expiry) - persona_result = self.authenticator.token_flow_manager.get_persona_token( - force_refresh=force_refresh - ) - - if not persona_result.success: - raise Exception(f"Failed to get persona token: {persona_result.error}") - - # Cache the entire PersonaResult with expiry - self._persona_cache = persona_result - logger.debug( - f"Cached persona token in memory (expires at {time.ctime(persona_result.expires_at)})" - ) - - return persona_result.persona_token + """Return a valid persona token (cached). Delegates to AuthBridge.""" + return self._auth.get_persona_token(force_refresh=force_refresh) def _ensure_authenticated(self) -> str: - """Ensure we have a valid persona token with accurate caching""" - return self.get_persona_token(force_refresh=False) + """Return a valid persona token (lazy auth). Delegates to AuthBridge.""" + return self._auth.ensure_authenticated() - def clear_persona_cache(self): - """Clear in-memory persona cache""" - self._persona_cache = None - logger.debug("Cleared in-memory persona cache") + def clear_persona_cache(self) -> None: + """Discard the in-memory persona token cache. Delegates to AuthBridge.""" + self._auth.clear_persona_cache() - def get_dynamic_manifest_params(self, channel: StreamingChannel, **kwargs) -> Optional[str]: - return None + def _vod_auth_headers(self) -> Dict[str, str]: + """Build auth headers for VOD endpoints. Delegates to AuthBridge.""" + return self._auth.vod_auth_headers() - @staticmethod - def _extract_channel_id_from_entry(entry: Dict) -> Optional[str]: - """Extract the correct channel ID from era$mediaPids""" - try: - stations = entry.get("stations", {}) - if not stations: - return None + def _pvr_auth_headers(self) -> Dict[str, str]: + """Build auth headers for nPVR endpoints. Delegates to AuthBridge.""" + return self._auth.pvr_auth_headers() - # Get the first station - station_id = next(iter(stations.keys())) - station_info = stations[station_id] - - # Extract from era$mediaPids (this is the correct ID for API calls) - era_media_pids = station_info.get("era$mediaPids", {}) - channel_id = era_media_pids.get("urn:theplatform:tv:location:any") - - if channel_id: - logger.debug(f"Extracted channel ID from era$mediaPids: {channel_id}") - return channel_id - - # Fallback to guid if era$mediaPids not available - fallback_id = entry.get("guid") - if fallback_id: - logger.warning(f"Using fallback channel ID from guid: {fallback_id}") - return fallback_id - - logger.warning("No channel ID found in entry") - return None - - except Exception as e: - logger.warning(f"Error extracting channel ID from entry: {e}") - return None + # ------------------------------------------------------------------ # + # Image scaling # + # ------------------------------------------------------------------ # def _build_scaled_image_url(self, original_url: str) -> Optional[str]: - """Build scaled image URL using image scaling service""" + """Return a scaled logo URL using the image scaling service.""" if not original_url: return None - if not self.provider_config or not self.provider_config.manifest: - return original_url # Fallback to original URL + return original_url image_config = self.provider_config.manifest.image_config - - # Check if we have the required scaling parameters if not image_config.scaling_base_url or not image_config.scaling_call_parameter: return original_url - # Parse the call parameter (e.g., "client=ftp22") - call_params = {} + call_params: Dict[str, str] = {} for param in image_config.scaling_call_parameter.split("&"): if "=" in param: key, value = param.split("=", 1) call_params[key] = value - # Build the scaling URL base_url = image_config.scaling_base_url.rstrip("/") - - # Add required parameters - params = { - **call_params, # client=ftp22 - "x": "120", - "y": "42", - "ar": "keep", # aspect ratio - "src": original_url, # original image URL - } - - # Build query string + params = {**call_params, "x": "120", "y": "42", "ar": "keep", "src": original_url} query_string = "&".join([f"{k}={quote(v, safe='')}" for k, v in params.items()]) - return f"{base_url}/iss?{query_string}" - def _process_channel_stations_response_optimized( - self, response_data: Dict, prefer_highest_quality: bool = True - ) -> List[StreamingChannel]: - """Ultra-optimized single-pass processing""" + # ------------------------------------------------------------------ # + # Header helpers # + # ------------------------------------------------------------------ # - if "entries" not in response_data: - return [] + def _get_dcm_headers(self) -> Dict[str, str]: + return { + "User-Agent": self.platform_config["user_agent"], + "Content-Type": "application/json", + "Accept": "application/json", + "x-dt-session-id": self.session_id, + "x-dt-call-id": self._generate_call_id(), + } - best_entries = {} - channels = [] + def _get_api_headers(self, require_auth: bool = False) -> Dict[str, str]: + headers: Dict[str, str] = { + "User-Agent": self.platform_config["user_agent"], + "Accept": "application/json", + "Content-Type": "application/json", + } + if require_auth: + persona_token = self._ensure_authenticated() + headers["Authorization"] = f"Basic {persona_token}" + return headers - for entry in response_data["entries"]: - try: - stations = entry.get("stations", {}) - if not stations: - continue + # ------------------------------------------------------------------ # + # Public StreamingProvider API # + # ------------------------------------------------------------------ # - station_info = next(iter(stations.values())) - display_number = entry.get("dt$displayChannelNumber") - - if display_number is None: - # Process immediately if no number (no filtering needed) - channel = self._create_channel_from_entry(entry, station_info, display_number) - if channel: - channels.append(channel) - continue - - # Extract quality for comparison using the shared QUALITY_RANK constant - quality = station_info.get("dt$quality", "SD") - current_rank = QUALITY_RANK.get(quality, 1) - - # Check if we need to replace existing entry - existing = best_entries.get(display_number) - if not existing: - best_entries[display_number] = (entry, station_info, current_rank) - else: - _, _, existing_rank = existing - if (prefer_highest_quality and current_rank > existing_rank) or ( - not prefer_highest_quality and current_rank < existing_rank - ): - best_entries[display_number] = ( - entry, - station_info, - current_rank, - ) - - except Exception: - continue - - # Convert best entries to channels - for display_number, (entry, station_info, _) in best_entries.items(): - channel = self._create_channel_from_entry(entry, station_info, display_number) - if channel: - channels.append(channel) - - return channels - - def _create_channel_from_entry(self, entry, station_info, display_number): - """Helper to create StreamingChannel from entry data""" - try: - title = station_info.get("title") or entry.get("title", "Unknown Channel") - title = title.replace(" - Main", "") - - channel_id = self._extract_channel_id_from_entry(entry) - if not channel_id: - return None - - # Logo processing - logo_url = None - thumbnails = station_info.get("thumbnails", {}) - for logo_type in ["stationLogo", "stationLogoColored"]: - if logo_type in thumbnails: - original_url = thumbnails[logo_type].get("url") - if original_url: - logo_url = self._build_scaled_image_url(original_url) - break - - magenta2_channel = Magenta2Channel( - name=title, - channel_id=channel_id, - logo_url=logo_url, - mode=MODE_LIVE, - content_type=CONTENT_TYPE_LIVE, - country=self.country, - raw_data=entry, - ) - - streaming_channel = magenta2_channel.to_streaming_channel( - provider_name=self.provider_name - ) - streaming_channel.channel_number = display_number - streaming_channel.quality = station_info.get("dt$quality", "SD") - - return streaming_channel - - except Exception as e: - logger.warning(f"Error creating channel from entry: {e}") - return None - - def _fetch_station_metadata(self) -> Dict[str, Dict]: - """ - Fetch the channel-stations feed and return a lookup map keyed by the - theplatform Station URI (the key of the stations dict, e.g. - "http://data.entertainment.tv.theplatform.eu/…/Station/265808936224"). - - This matches listings[0].stationId in the entitled-channels feed, which - is what TheplatformChannel.station_id contains after parsing. - - Note: era$mediaPids["urn:theplatform:tv:location:any"] is a short opaque - PID used for other purposes — it is NOT the mapping key. - - Each value is a dict with: - title – display name (" - Main" suffix already stripped) - logo_url – scaled logo URL or None - quality – "HD", "SD", etc. - """ - metadata: Dict[str, Dict] = {} - try: - url = None - if self.endpoint_manager: - url = ( - self.endpoint_manager.get_endpoint("channel_stations") - or self.endpoint_manager.get_endpoint("channel_list") - ) - url = url or "https://feed.entertainment.tv.theplatform.eu/f/mdeprod/mdeprod-channel-stations-main" - url += "?lang=short-de&sort=dt%24displayChannelNumber&range=1-1000" - - headers = self._get_api_headers(require_auth=False) - response = self.http_manager.get( - url, operation="api", headers=headers, timeout=DEFAULT_REQUEST_TIMEOUT - ) - response.raise_for_status() - data = response.json() - - for entry in data.get("entries", []): - try: - stations = entry.get("stations", {}) - if not stations: - continue - # The stations dict key IS the Station URI, matching - # listings[0].stationId in the entitled-channels feed. - # era$mediaPids["urn:theplatform:tv:location:any"] is a - # short opaque PID — a different identifier entirely. - station_uri = next(iter(stations.keys())) - station_info = stations[station_uri] - - title = (station_info.get("title") or entry.get("title", "")).replace(" - Main", "") - quality = station_info.get("dt$quality", "SD") - - logo_url = None - thumbnails = station_info.get("thumbnails", {}) - for logo_type in ["stationLogo", "stationLogoColored"]: - if logo_type in thumbnails: - original_url = thumbnails[logo_type].get("url") - if original_url: - logo_url = self._build_scaled_image_url(original_url) - break - - channel_number = entry.get("dt$displayChannelNumber") - - # Keep highest-quality entry when duplicates exist - existing = metadata.get(station_uri) - if not existing or QUALITY_RANK.get(quality, 1) > QUALITY_RANK.get(existing["quality"], 1): - metadata[station_uri] = { - "title": title, - "logo_url": logo_url, - "quality": quality, - "channel_number": channel_number, - } - except Exception as exc: - logger.debug(f"_fetch_station_metadata: skipping entry: {exc}") - - logger.debug(f"_fetch_station_metadata: built metadata for {len(metadata)} stations") - except Exception as exc: - logger.warning(f"_fetch_station_metadata: failed, channels will have no names: {exc}") - - return metadata + def get_dynamic_manifest_params( + self, channel: StreamingChannel, **kwargs: Any + ) -> Optional[str]: + return None def get_channels( self, @@ -896,513 +634,50 @@ class Magenta2Provider(StreamingProvider): fetch_manifests: bool = False, populate_streaming_data: bool = True, prefer_highest_quality: bool = True, - **kwargs, + **kwargs: Any, ) -> List[StreamingChannel]: - """ - Fetch available channels from Magenta2 via the entitled-channels flow. - - Uses lib_theplatform to: - 1. Call getApplicableDistributionRights (license_service_url from manifest). - 2. Fetch the entitled-channels feed filtered by those rights. - 3. Merge with station metadata pre-fetched at init time (unauthenticated). - 4. Convert each TheplatformChannel to a StreamingChannel, enriched with - metadata from the stations feed. - - Results are cached after the first successful call. Subsequent calls - return directly from the cache without hitting the network again. - - Note: channel_id (= release_pid) comes exclusively from the - entitled-channels feed (step 2, authenticated). The station-metadata - dict fetched at init time supplies only display data (name, logo, quality). - """ - # ── Cache guard ─────────────────────────────────────────────────────── - # After the first successful call _cached_channels holds the full list. - # Return it directly — no network calls, no re-parsing. - if hasattr(self, "_cached_channels") and self._cached_channels: - logger.debug("get_channels: returning from cache (no network calls)") - return list(self._cached_channels) - - try: - cid = f"{self.session_id}::{self._generate_call_id()}" - user_agent = self.platform_config["user_agent"] - - # ── Step 1: resolve distribution rights ────────────────────────── - rights_url = ( - self.provider_config.manifest.mpx.license_service_url - if self.provider_config and self.provider_config.manifest - else None - ) - if not rights_url: - raise RuntimeError( - "No license_service_url available – configuration discovery may have failed" - ) - - persona_token = self._ensure_authenticated() - auth_headers = { - "Authorization": f"Basic {persona_token}", - "Origin": "https://www.magenta.tv", - "Referer": "https://www.magenta.tv/", - } - - distribution_rights = fetch_distribution_rights( - http_manager=self.http_manager, - rights_url=rights_url, - cid=cid, - user_agent=user_agent, - timeout=DEFAULT_REQUEST_TIMEOUT, - extra_headers=auth_headers, - ) - - # Log subscription packages by resolving distribution IDs to names. - # Rights arrive as full URIs like - # "http://data.entitlement.theplatform.eu/.../DistributionRight/376449962" - # so we extract just the trailing integer segment before the lookup. - def _dist_uri_to_int(uri) -> int | None: - try: - return int(str(uri).rstrip("/").rsplit("/", 1)[-1]) - except (ValueError, AttributeError): - return None - - package_names = [ - DISTRIBUTION_PACKAGE_NAMES.get(numeric_id, f"Unknown package ({dist_id})") - for dist_id in distribution_rights - if (numeric_id := _dist_uri_to_int(dist_id)) is not None - ] - logger.info(f"Active subscription packages ({len(package_names)}):") - for pkg in package_names: - logger.info(f" · {pkg}") - - # ── Step 2: fetch entitled-channels feed ───────────────────────── - feed_url = ( - self.endpoint_manager.get_endpoint("mpx_feed_entitledChannelsFeed") - if self.endpoint_manager - else None - ) or "https://feed.entertainment.tv.theplatform.eu/f/mdeprod/mdeprod-entitled-channels" - - tp_channels: List[TheplatformChannel] = fetch_entitled_channels_feed( - http_manager=self.http_manager, - feed_url=feed_url, - distribution_rights=distribution_rights, - cid=cid, - user_agent=user_agent, - timeout=DEFAULT_REQUEST_TIMEOUT, - ) - - # ── Step 3: use station metadata pre-fetched at init time ──────── - # _station_metadata was populated from the unauthenticated - # channel-stations feed during __init__. No network call needed. - station_metadata = self._station_metadata - - # ── DIAGNOSTIC ─────────────────────────────────────────────────── - #logger.info(f"get_channels: station_metadata has {len(station_metadata)} entries") - #if station_metadata: - # sample_key = next(iter(station_metadata)) - # logger.debug(f"get_channels: sample station_metadata key: {sample_key!r}") - #if tp_channels: - # logger.debug(f"get_channels: sample tp_ch.station_id: {tp_channels[0].station_id!r}") - # ── END DIAGNOSTIC ─────────────────────────────────────────────── - - # ── Step 4: convert to StreamingChannel ────────────────────────── - channels: List[StreamingChannel] = [] - for tp_ch in tp_channels: - try: - meta = station_metadata.get(tp_ch.station_id, {}) - name = meta.get("title") or tp_ch.station_id - logo_url = meta.get("logo_url") - quality = meta.get("quality") - # Prefer the channel number from the stations feed (fetched at - # init time) over the one from the entitled-channels feed, as - # the stations feed is the authoritative source for display numbers. - channel_number = ( - meta["channel_number"] - if meta.get("channel_number") is not None - else tp_ch.channel_number - ) - - magenta2_channel = Magenta2Channel( - name=name, - channel_id=tp_ch.release_pid, - logo_url=logo_url, - mode=MODE_LIVE, - content_type=CONTENT_TYPE_LIVE, - country=self.country, - raw_data=tp_ch.extra, - ) - streaming_channel = magenta2_channel.to_streaming_channel( - provider_name=self.provider_name - ) - streaming_channel.channel_number = channel_number - streaming_channel.quality = quality - streaming_channel.manifest = tp_ch.mpd_url - if tp_ch.hls_url: - streaming_channel.hls_url = tp_ch.hls_url - - # Cache manifest URL and releasePid keyed by release_pid, - # which is now the channel's channel_id (content_id). - self._live_manifest_cache[tp_ch.release_pid] = tp_ch.mpd_url - self._live_pid_cache[tp_ch.release_pid] = tp_ch.release_pid - - channels.append(streaming_channel) - except Exception as exc: - logger.warning(f"get_channels: skipping channel {tp_ch.station_id}: {exc}") - - logger.info( - f"Successfully fetched {len(channels)} entitled channels " - f"for country {self.country} " - f"({len(station_metadata)} stations with metadata)" - ) - # Persist the full channel list so subsequent get_channels() calls - # return immediately from cache without any network requests. - # Sort by channel number; channels without a number go to the end - channels.sort(key=lambda ch: (ch.channel_number is None, ch.channel_number or 0)) - self._cached_channels = channels - return channels - - except Exception as e: - raise Exception(f"Error fetching channels from Magenta2 API: {e}") + return self._channel_manager.get_channels( + time_window_hours=time_window_hours, + fetch_manifests=fetch_manifests, + populate_streaming=populate_streaming_data, + prefer_highest_quality=prefer_highest_quality, + **kwargs, + ) def get_events( - self, - start_time: Optional[datetime] = None, - end_time: Optional[datetime] = None, - **kwargs, + self, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + **kwargs: Any, ) -> List[Event]: return [] - def get_entitlement_token(self, content_id: str, content_type: str = CONTENT_TYPE_LIVE) -> str: - """ - Get entitlement token using persona_token Basic auth - """ - # Ensure we're authenticated with persona token - self._ensure_authenticated() + def get_manifest( + self, content_id: str, content_type: str = CONTENT_TYPE_LIVE, **kwargs: Any + ) -> Optional[str]: + return self._playback_manager.get_manifest(content_id, content_type, **kwargs) - # Use persona token (Basic auth) for entitlement - headers = self._get_api_headers(require_auth=True) + def get_drm( + self, content_id: str, content_type: str = CONTENT_TYPE_LIVE, **kwargs: Any + ) -> List[DRMConfig]: + return self._playback_manager.get_drm(content_id, content_type, **kwargs) - payload = {"content_id": content_id, "content_type": content_type} - - url = ( - self.endpoint_manager.get_endpoint("entitlement") - if self.endpoint_manager - else "https://entitlement.p7s1.io/api/user/entitlement-token" + def get_catchup_manifest( + self, channel_id: str, start_time: int, end_time: int, **kwargs: Any + ) -> Optional[str]: + return self._playback_manager.get_catchup_manifest( + channel_id, start_time, end_time, **kwargs ) - try: - logger.debug(f"Requesting entitlement token with persona token for: {content_id}") - response = self.http_manager.post( - url, - operation="auth", - headers=headers, - json_data=payload, - timeout=DEFAULT_REQUEST_TIMEOUT, - ) - - if response.status_code == 400: - try: - error_data = response.json() - error_list = error_data if isinstance(error_data, list) else [error_data] - - if len(error_list) > 0: - error = error_list[0] - code = error.get("code", error.get("errorCode", "UNKNOWN")) - msg = error.get("msg", error.get("message", "No error message provided")) - - if code == ERROR_CODES["PLAYBACK_RESTRICTED"]: - raise Magenta2PlaybackRestrictedException( - f"Playback restricted for {content_id}: {msg}" - ) - else: - raise Exception(f"Entitlement error for {content_id} ({code}): {msg}") - except (json.JSONDecodeError, KeyError, IndexError) as e: - raise Exception( - f"Bad response for {content_id} (400), failed to parse error: {e}" - ) - - response.raise_for_status() - data = response.json() - - if "entitlement_token" in data: - return data["entitlement_token"] - elif "entitlementToken" in data: - return data["entitlementToken"] - elif "token" in data: - return data["token"] - else: - raise KeyError("No entitlement token found in response") - - except Magenta2PlaybackRestrictedException: - raise - except KeyError as e: - logger.error(f"No entitlement token in response for {content_id}: {e}") - logger.debug(f"Auth state: {self.authenticator.debug_authentication_state()}") - raise Exception(f"No entitlement token in response for {content_id}: {e}") - except Exception as e: - logger.error(f"Error getting entitlement token for {content_id}: {e}") - logger.debug(f"Auth state: {self.authenticator.debug_authentication_state()}") - raise Exception(f"Error getting entitlement token for {content_id}: {e}") - - def get_channel_playlist(self, channel_id: str, entitlement_token: str) -> Dict: - """Get channel playlist data""" - if self.endpoint_manager and self.endpoint_manager.has_endpoint("channel_playlist"): - url = self.endpoint_manager.get_endpoint("channel_playlist").format( - channel_id=channel_id - ) - else: - url = f"https://api.magentatv.de/v1/channel/{channel_id}/playlist" - - headers = { - "Authorization": f"Bearer {entitlement_token}", - "User-Agent": self.platform_config["user_agent"], - "Accept": "application/json", - } - - try: - response = self.http_manager.get( - url, - operation="manifest", - headers=headers, - timeout=DEFAULT_REQUEST_TIMEOUT, - ) - response.raise_for_status() - return response.json() - - except Exception as e: - raise Exception(f"Error getting playlist for {channel_id}: {e}") - - def populate_streaming_data( - self, channels: List[StreamingChannel], max_retries: int = DEFAULT_MAX_RETRIES - ) -> List[StreamingChannel]: - """Populate streaming data (manifest, DRM) for all channels""" - self._ensure_authenticated() - - successful_channels = [] - - for channel in channels: - retries = 0 - success = False - is_restricted = False - - while retries < max_retries and not success and not is_restricted: - try: - logger.debug( - f"Getting entitlement token for: {channel.name} (attempt {retries + 1})" - ) - - entitlement_token = self.get_entitlement_token( - content_id=channel.channel_id, content_type=channel.content_type - ) - - logger.debug(f"Getting playlist data for: {channel.name}") - - playlist_data = self.get_channel_playlist(channel.channel_id, entitlement_token) - - manifest_url = playlist_data.get("manifestUrl", playlist_data.get("manifest")) - license_url = playlist_data.get("licenseUrl", playlist_data.get("license")) - certificate_url = playlist_data.get( - "certificateUrl", playlist_data.get("certificate") - ) - streaming_format = playlist_data.get( - "streamingFormat", playlist_data.get("format", "dash") - ) - - if manifest_url: - channel.manifest = manifest_url - channel.cdm_type = DRM_SYSTEM_WIDEVINE - channel.cdm = f"pid={channel.channel_id}" - channel.license_url = license_url - channel.certificate_url = certificate_url - channel.streaming_format = streaming_format - - logger.info(f"Streaming data populated for: {channel.name}") - successful_channels.append(channel) - success = True - else: - raise Exception("No manifest URL in response") - - except Magenta2PlaybackRestrictedException as e: - logger.warning(f"Playback restricted for {channel.name}: {e}") - is_restricted = True - - except Exception as e: - retries += 1 - if retries < max_retries: - logger.debug(f"Retry {retries}/{max_retries} for {channel.name}: {e}") - time.sleep(1) - else: - logger.error(f"Failed to get streaming data for {channel.name}: {e}") - - logger.info(f"Streaming data population complete:") - logger.info(f" Successful: {len(successful_channels)}") - logger.info(f" Failed/Restricted: {len(channels) - len(successful_channels)}") - logger.info(f" Total: {len(channels)}") - - return successful_channels - - def _get_tvhubs_bearer(self) -> Optional[str]: - """ - Return a valid tvhubs-scoped access_token for use as Bearer. - - The tvhubs token has audience "https://tvhubs.telekom.de" and is what - the real device sends in Authorization: Bearer for all tvhubs/wcps calls. - - If the stored token is expired, refreshes it automatically using the - shared refresh_token via sam3_client.refresh_access_token("tvhubs"). - Falls back to None if unavailable (caller falls back to persona_jwt). - """ - import time as _time - try: - tfm = self.authenticator.token_flow_manager - token_data = tfm.session_manager.load_scoped_token( - self.provider_name, "tvhubs", self.country - ) - if not token_data or not token_data.get("access_token"): - return None - - # Check if expired - issued_at = token_data.get("issued_at", 0) - expires_in = token_data.get("expires_in", 7200) - if _time.time() < issued_at + expires_in - 60: - # Still valid (with 60s safety margin) - return token_data["access_token"] - - # Expired — refresh using the shared refresh_token. - # The refresh_token lives in top-level session data (shared across - # all subordinate tokens: tvhubs, taa, yo_digital). - logger.debug("tvhubs token expired, refreshing via sam3_client") - sam3 = tfm.sam3_client - - # Load shared refresh_token from session storage if not on instance - shared_rt = ( - sam3.refresh_token - if sam3 and sam3.refresh_token - else ( - (tfm.session_manager.load_session(self.provider_name, self.country) or {}) - .get("refresh_token") - ) - ) - if not shared_rt: - logger.warning("Cannot refresh tvhubs token: no shared refresh_token found") - return None - - # Inject into sam3_client so refresh_access_token can use it - sam3.refresh_token = shared_rt - new_token = sam3.refresh_access_token("tvhubs") - if new_token: - # Persist the refreshed token - tfm._save_tvhubs_token({ - "access_token": new_token, - "token_type": "Bearer", - "expires_in": 7200, - }) - logger.debug("tvhubs token refreshed successfully") - return new_token - - logger.warning("tvhubs token refresh failed") - except Exception as exc: - logger.debug(f"Could not load/refresh tvhubs token: {exc}") - return None - - def _vod_auth_headers(self) -> Dict[str, str]: - """ - Build authentication headers for VOD endpoints, platform-aware. - - Authorization Bearer uses the tvhubs-scoped token (audience - https://tvhubs.telekom.de) — this is what the real device sends - and what the server uses for partner entitlement decisions. - Falls back to persona_jwt if tvhubs token is unavailable. - - ftv-web: x-dt-session-id / x-dt-call-id, origin, referer, - x-permissionflagpersonalizeduireco - ftv-android: x-stbserialnumber, dt-session-id / dt-call-id (no x- prefix), - accept-encoding: gzip - """ - persona_token = self._ensure_authenticated() - # Prefer tvhubs token as Bearer — it carries correct entitlement scope. - tvhubs_token = self._get_tvhubs_bearer() - if tvhubs_token: - auth_value = f"Bearer {tvhubs_token}" - logger.debug("VOD auth: using tvhubs token as Bearer") - else: - persona_jwt = self._extract_persona_jwt_from_token(persona_token) - auth_value = ( - f"Bearer {persona_jwt}" if persona_jwt else f"Basic {persona_token}" - ) - logger.debug("VOD auth: tvhubs token unavailable, falling back to persona_jwt") - - client_model: str = ( - self.provider_config.bootstrap.client_model - if self.provider_config and self.provider_config.bootstrap - else f"ftv-{self.platform}" - ) - is_web = client_model == "ftv-web" - - if is_web: - return { - "Authorization": auth_value, - "x-mpx-authorization": f"Basic {persona_token}", - "x-dt-session-id": self.session_id, - "x-dt-call-id": self._generate_call_id(), - "origin": "https://www.magenta.tv", - "referer": "https://www.magenta.tv/", - "user-agent": self.platform_config["user_agent"], - "accept": "*/*", - "accept-encoding": "gzip, deflate, br, zstd", - "accept-language": "de-DE,de;q=0.9", - "x-permissionflagpersonalizeduireco": "false", - } - else: - return { - "Authorization": auth_value, - "x-mpx-authorization": f"Basic {persona_token}", - "x-stbserialnumber": self.serial_number, - "dt-session-id": self.session_id, - "dt-call-id": self._generate_call_id(), - "user-agent": self.platform_config["user_agent"], - "accept-encoding": "gzip", - } - - - def _pvr_auth_headers(self) -> Dict[str, str]: - """ - Build authentication headers for nPVR (Audience) recording endpoints. - - The nPVR Audience API authenticates with ``Authorization: Basic - {persona_token}`` — the same Base64-encoded persona token used for - MPX licence and entitlement requests, *not* a Bearer JWT. - - A ``CID`` correlation header (session::call) is included to match - real-device request patterns observed in the API. - """ - persona_token = self._ensure_authenticated() - return { - "Authorization": f"Basic {persona_token}", - "User-Agent": self.platform_config["user_agent"], - "Accept-Encoding": "gzip", - "CID": f"{self.session_id}::{self._generate_call_id()}", - } - - def get_vod_category(self, content_id: str = "", **kwargs): - """ - Return the children of a VOD node. - - Args: - content_id: Opaque node identifier produced by a previous - get_vod_category call (e.g. "lane:322341", - "series:GN_SERIES_20914057"). Empty string → root. - cursor: Opaque continuation token from a previous response's - next_cursor field. None → first page. - Encoded as a plain integer string by VodManager - (the $offset value for the next UnstructuredGrid call). - page_size: Number of items to request per page (default from - VOD_DEFAULT_PAGE_SIZE). Passed straight through to - VodManager and then to the $size query parameter. - """ + def get_vod_category(self, content_id: str = "", **kwargs: Any) -> Any: + """Return children of a VOD node (empty string → root).""" if not self._vod_manager: - raise RuntimeError("VodManager not available - configuration discovery may have failed") + raise RuntimeError( + "VodManager not available — configuration discovery may have failed" + ) return self._vod_manager.get_children(content_id=content_id, **kwargs) - - def get_recordings(self, include_deleted: bool = False, **kwargs): + def get_recordings(self, include_deleted: bool = False, **kwargs: Any) -> Any: """Return a list of Recording objects from the nPVR backend.""" if not self._recordings_manager: raise RuntimeError( @@ -1411,172 +686,32 @@ class Magenta2Provider(StreamingProvider): recordings = self._recordings_manager.get_recordings( include_deleted=include_deleted, **kwargs ) - # Cache content_id → manifest_script so get_manifest() / get_drm() - # can inject smil_base_url automatically without callers needing to - # know about it (e.g. DRMOperations passes only content_id). for rec in recordings: if rec.content_id and rec.manifest_script: self._recording_url_cache[rec.content_id] = rec.manifest_script return recordings - def delete_recording(self, recording_id: str, **kwargs) -> None: - """Permanently delete a recording on the nPVR backend.""" + def delete_recording(self, recording_id: str, **kwargs: Any) -> None: if not self._recordings_manager: raise RuntimeError( "RecordingsManager not available — configuration discovery may have failed" ) self._recordings_manager.delete_recording(recording_id) - def get_recording_manifest(self, recording_id: str, **kwargs) -> Optional[str]: - """ - Return the playback URL for a specific recording by ID. - - For recordings already in memory (returned by get_recordings), the - manifest_script field already contains the playback URL. This method - performs a fresh API lookup — use it only when the Recording object is - not available. - """ + def get_recording_manifest(self, recording_id: str, **kwargs: Any) -> Optional[str]: + """Return the playback URL for a recording by ID (fresh API lookup).""" if not self._recordings_manager: return None return self._recordings_manager.get_recording_manifest(recording_id) - def _ensure_live_cache(self) -> None: - """ - Populate _live_manifest_cache and _live_pid_cache on demand. - Called automatically by get_manifest() / get_drm() so that callers - do not have to call get_channels() explicitly first. - """ - if not self._live_manifest_cache: - logger.debug("Live channel cache is empty — auto-populating via get_channels()") - self.get_channels() - - def get_manifest( - self, content_id: str, content_type: str = CONTENT_TYPE_LIVE, **kwargs - ) -> Optional[str]: - """Get MPD manifest URL. - - For live channels whose manifest was already fetched by get_channels(), - return the cached MPD URL directly without a SMIL round-trip. - VOD and recordings fall through to SmilManager as before. - """ - if content_type == CONTENT_TYPE_LIVE: - self._ensure_live_cache() # ← auto-bootstrap - if content_id in self._live_manifest_cache: - logger.debug(f"get_manifest: cache hit for live channel {content_id}") - return self._live_manifest_cache[content_id] - - if not self._smil_manager: - raise RuntimeError("SmilManager not available") - if "smil_base_url" not in kwargs and content_id in self._recording_url_cache: - kwargs["smil_base_url"] = self._recording_url_cache[content_id] - logger.debug(f"Injected smil_base_url from recording cache for {content_id}") - return self._smil_manager.get_manifest(content_id, content_type, **kwargs) - - def get_drm( - self, content_id: str, content_type: str = CONTENT_TYPE_LIVE, **kwargs - ) -> List[DRMConfig]: - """Get DRM configuration. - - For live channels whose releasePid was cached by get_channels(), build - the Widevine licence URL directly using lib_theplatform — no SMIL fetch. - VOD and recordings fall through to SmilManager as before. - """ - if content_type == CONTENT_TYPE_LIVE: - self._ensure_live_cache() # ← auto-bootstrap - if content_id in self._live_pid_cache: - release_pid = content_id - logger.debug( - f"get_drm: building licence directly for live channel " - f"(releasePid: {release_pid})" - ) - try: - persona_token = self._ensure_authenticated() - raw_jwt = extract_persona_jwt(persona_token) - if not raw_jwt: - logger.error("get_drm: failed to extract persona JWT") - return [] - - widevine_endpoint = ( - self.endpoint_manager.get_endpoint("widevine_license") - if self.endpoint_manager - else None - ) - if not widevine_endpoint: - logger.error("get_drm: no widevine_license endpoint available") - return [] - - account_uri = ( - self.provider_config.manifest.mpx.get_account_uri() - if self.provider_config and self.provider_config.manifest - else None - ) or MAGENTA2_FALLBACK_ACCOUNT_URI - - licence_url = build_licence_url( - widevine_endpoint=widevine_endpoint, - release_pid=release_pid, - persona_jwt=raw_jwt, - account_uri=account_uri, - ) - return [build_widevine_drm_config( - licence_url=licence_url, - user_agent=self.platform_config["user_agent"], - )] - except Exception as exc: - logger.error(f"get_drm: direct licence build failed for {content_id}: {exc}") - return [] - - if not self._smil_manager: - raise RuntimeError("SmilManager not available") - if "smil_base_url" not in kwargs and content_id in self._recording_url_cache: - kwargs["smil_base_url"] = self._recording_url_cache[content_id] - logger.debug(f"Injected smil_base_url from recording cache for {content_id}") - return self._smil_manager.get_drm(content_id, content_type, **kwargs) - - def get_catchup_manifest( - self, channel_id: str, start_time: int, end_time: int, **kwargs - ) -> Optional[str]: - """Get catchup manifest URL via SmilManager.""" - if not self._smil_manager: - raise RuntimeError("SmilManager not available") - return self._smil_manager.get_catchup_manifest( - channel_id, start_time, end_time, **kwargs - ) - - @staticmethod - def _extract_persona_jwt_from_token(persona_token: str) -> Optional[str]: - """ - Extract the raw persona JWT token from Base64-encoded persona token - - The persona token format is: Base64(account_uri + ":" + persona_jwt) - This method decodes it and extracts just the persona_jwt part. - """ - try: - decoded = base64.b64decode(persona_token).decode("utf-8") - last_colon_index = decoded.rfind(":") - - if last_colon_index == -1: - logger.error("No colon found in decoded persona token") - return None - - persona_jwt = decoded[last_colon_index + 1:] - - if not persona_jwt.startswith("eyJ"): - logger.error(f"Extracted token doesn't look like a JWT") - return None - - return persona_jwt - except Exception as e: - logger.error(f"Error extracting persona JWT token: {e}") - return None - def get_epg( self, channel_id: str, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, - **kwargs, - ) -> List[Dict]: - """Get EPG data for a channel""" + **kwargs: Any, + ) -> List[Dict[str, Any]]: + """Fetch EPG data for a channel.""" try: if start_time is None: start_time = datetime.now() @@ -1584,11 +719,9 @@ class Magenta2Provider(StreamingProvider): end_time = datetime.now() + timedelta(hours=DEFAULT_EPG_WINDOW_HOURS) headers = self._get_api_headers(require_auth=False) - - url = ( + url: str = ( self.endpoint_manager.get_endpoint("epg") - if self.endpoint_manager - else "https://api.magentatv.de/proxy/device/epg" + or "https://api.magentatv.de/proxy/device/epg" ) params = { @@ -1596,7 +729,6 @@ class Magenta2Provider(StreamingProvider): "start": start_time.isoformat(), "end": end_time.isoformat(), } - response = self.http_manager.get( url, operation="api", @@ -1605,159 +737,29 @@ class Magenta2Provider(StreamingProvider): timeout=DEFAULT_REQUEST_TIMEOUT, ) response.raise_for_status() - return response.json() - + return response.json() # type: ignore[no-any-return] except Exception as e: logger.error(f"Error getting EPG for channel {channel_id}: {e}") return [] - def _calculate_auth_state(self, context) -> "AuthState": - """ - Custom auth state calculation for Magenta2. - Checks persona token with its special structure. - """ - # Get persona token data - persona_token = context.get_token(self.provider_name, "persona", self.country) + # ------------------------------------------------------------------ # + # Auth state / readiness introspection # + # ------------------------------------------------------------------ # - if not persona_token: - logger.debug("No persona token found") - return AuthState.NOT_AUTHENTICATED + def _calculate_auth_state(self, context: Any) -> AuthState: + """Delegates to AuthBridge.""" + return self._auth.calculate_auth_state(context) - # Validate persona token structure - if not isinstance(persona_token, dict) or "persona_token" not in persona_token: - logger.warning("Invalid persona token structure") - return AuthState.NOT_AUTHENTICATED + def _calculate_readiness(self, context: Any) -> Tuple[bool, str]: + """Delegates to AuthBridge.""" + return self._auth.calculate_readiness(context) - # Check if persona token is expired - if "expires_at" in persona_token: - current_time = time.time() - expires_at = persona_token["expires_at"] - - # Add 5-minute buffer - if current_time >= (expires_at - 300): - logger.debug( - f"Persona token expired (expires_at: {expires_at}, now: {current_time})" - ) - return AuthState.EXPIRED - - # Token exists and is valid - logger.debug("Persona token is valid") - return AuthState.AUTHENTICATED - - def _calculate_readiness(self, context) -> Tuple[bool, str]: - """ - Custom readiness calculation for Magenta2. - Ready if we have a valid persona token. - """ - # Get persona token data - persona_token = context.get_token(self.provider_name, "persona", self.country) - - if not persona_token: - return False, "No persona token available" - - # Validate structure - if not isinstance(persona_token, dict) or "persona_token" not in persona_token: - return False, "Invalid persona token structure" - - # Check expiration - if "expires_at" in persona_token: - current_time = time.time() - expires_at = persona_token["expires_at"] - - # Add 5-minute buffer - if current_time >= (expires_at - 300): - # Token expired but might be refreshable - # Check if we have yo_digital token with refresh capability - yo_token = context.get_token(self.provider_name, "yo_digital", self.country) - if yo_token and "refresh_token" in yo_token: - # Check if yo_digital refresh token is still valid - if ( - "refresh_token_expires_in" in yo_token - and "refresh_token_issued_at" in yo_token - ): - refresh_expires_at = ( - yo_token["refresh_token_issued_at"] - + yo_token["refresh_token_expires_in"] - ) - if current_time < (refresh_expires_at - 300): - return ( - True, - "Persona token expired but can be refreshed via yo_digital", - ) - - return ( - False, - f"Persona token expired (expired at {time.ctime(expires_at)})", - ) - - # Valid persona token - return True, "Has valid persona token" - - def get_auth_details(self, context) -> Dict[str, Any]: - """ - Provide Magenta2-specific auth details. - Shows status of all token scopes. - """ - details = {} - - # Check all token scopes - for scope in self.token_scopes: - token = context.get_token(self.provider_name, scope, self.country) - - if not token: - details[scope] = {"available": False} - continue - - scope_info = {"available": True} - - # Handle persona token specially - if scope == "persona": - if "expires_at" in token: - current_time = time.time() - expires_at = token["expires_at"] - scope_info["expires_at"] = expires_at - scope_info["is_expired"] = current_time >= expires_at - scope_info["time_remaining"] = int(max(0, expires_at - current_time)) - - if "composed_at" in token: - scope_info["composed_at"] = token["composed_at"] - - # Handle yo_digital token - elif scope == "yo_digital": - if "access_token_expires_in" in token and "access_token_issued_at" in token: - current_time = time.time() - expires_at = token["access_token_issued_at"] + token["access_token_expires_in"] - scope_info["access_token_expires_at"] = expires_at - scope_info["access_token_is_expired"] = current_time >= expires_at - - if "refresh_token_expires_in" in token and "refresh_token_issued_at" in token: - current_time = time.time() - refresh_expires_at = ( - token["refresh_token_issued_at"] + token["refresh_token_expires_in"] - ) - scope_info["refresh_token_expires_at"] = refresh_expires_at - scope_info["refresh_token_is_expired"] = current_time >= refresh_expires_at - scope_info["has_refresh_token"] = True - - # Standard token handling (tvhubs, taa) - else: - if "expires_in" in token and "issued_at" in token: - current_time = time.time() - expires_at = token["issued_at"] + token["expires_in"] - scope_info["expires_at"] = expires_at - scope_info["is_expired"] = current_time >= expires_at - - details[scope] = scope_info - - return details + def get_auth_details(self, context: Any) -> Dict[str, Any]: + """Delegates to AuthBridge.""" + return self._auth.get_auth_details(self.token_scopes, context) def debug_authentication(self) -> Dict[str, Any]: - """ - Debug authentication state and token flow - - Returns comprehensive information about the current authentication state, - token availability, and TokenFlowManager status. - """ + """Return comprehensive auth-state and token-flow diagnostic info.""" result: Dict[str, Any] = { "provider": { "provider_name": self.provider_name, @@ -1766,7 +768,6 @@ class Magenta2Provider(StreamingProvider): } } - # Try to get current persona token status try: persona_token = self.get_persona_token(force_refresh=False) persona_info: Dict[str, Any] = { @@ -1774,30 +775,23 @@ class Magenta2Provider(StreamingProvider): "length": len(persona_token), "preview": persona_token[:50] + "...", } - - # Try to extract and verify the JWT inside try: - persona_jwt = self._extract_persona_jwt_from_token(persona_token) + persona_jwt = PlaybackManager.extract_persona_jwt_from_token(persona_token) persona_info["jwt_available"] = bool(persona_jwt) if persona_jwt: persona_info["jwt_length"] = len(persona_jwt) persona_info["jwt_preview"] = persona_jwt[:50] + "..." except Exception as e: persona_info["jwt_extraction_error"] = str(e) - result["persona_token"] = persona_info - except Exception as e: result["persona_token"] = {"available": False, "error": str(e)} - # TokenFlowManager status - if ( - hasattr(self.authenticator, "token_flow_manager") - and self.authenticator.token_flow_manager - ): + tfm = getattr(self.authenticator, "token_flow_manager", None) + if tfm is not None: result["token_flow_manager"] = { "available": True, - "token_status": self.authenticator.token_flow_manager.get_token_status(), + "token_status": tfm.get_token_status(), } else: result["token_flow_manager"] = { @@ -1805,23 +799,19 @@ class Magenta2Provider(StreamingProvider): "error": "TokenFlowManager not initialized", } - # Authenticator capabilities if hasattr(self.authenticator, "get_authentication_capabilities"): result["authentication_capabilities"] = ( self.authenticator.get_authentication_capabilities() ) - # Endpoint manager info - if self.endpoint_manager: - result["endpoints"] = { - "has_taa_auth": self.endpoint_manager.has_endpoint("taa_auth"), - "has_entitlement": self.endpoint_manager.has_endpoint("entitlement"), - "has_widevine_license": self.endpoint_manager.has_endpoint("widevine_license"), - "has_mpx_selector": self.endpoint_manager.has_endpoint("mpx_selector"), - "total_endpoints": len(self.endpoint_manager.get_all_endpoints()), - } + result["endpoints"] = { + "has_taa_auth": self.endpoint_manager.has_endpoint("taa_auth"), + "has_entitlement": self.endpoint_manager.has_endpoint("entitlement"), + "has_widevine_license": self.endpoint_manager.has_endpoint("widevine_license"), + "has_mpx_selector": self.endpoint_manager.has_endpoint("mpx_selector"), + "total_endpoints": len(self.endpoint_manager.get_all_endpoints()), + } - # SAM3 client status if hasattr(self.authenticator, "get_sam3_client_status"): result["sam3_client"] = self.authenticator.get_sam3_client_status()