From 11ae583a9df4a35121134dc6d976d9d31b4afca4 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Thu, 19 Mar 2026 09:35:25 +0100 Subject: [PATCH] Add discovery --- .../providers/magenta2/provider.py | 51 +++++++++++++++++-- .../providers/magenta2/vod_manager.py | 36 +++++-------- 2 files changed, 59 insertions(+), 28 deletions(-) diff --git a/lib/streaming_providers/providers/magenta2/provider.py b/lib/streaming_providers/providers/magenta2/provider.py index 7b14240..aaaa894 100644 --- a/lib/streaming_providers/providers/magenta2/provider.py +++ b/lib/streaming_providers/providers/magenta2/provider.py @@ -988,20 +988,51 @@ class Magenta2Provider(StreamingProvider): return successful_channels + def _get_tvhubs_bearer(self) -> Optional[str]: + """ + Return the 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. + Falls back to persona_jwt if the tvhubs token is unavailable. + """ + try: + tfm = self.authenticator.token_flow_manager + token_data = tfm.session_manager.load_scoped_token( + self.provider_name, "tvhubs", self.country + ) + if token_data and token_data.get("access_token"): + return token_data["access_token"] + except Exception as exc: + logger.debug(f"Could not load 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() - persona_jwt = self._extract_persona_jwt_from_token(persona_token) - auth_value = ( - f"Bearer {persona_jwt}" if persona_jwt else f"Basic {persona_token}" - ) + # 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 @@ -1274,8 +1305,18 @@ class Magenta2Provider(StreamingProvider): items = self._vod_manager.get_children(gn_id) if items: item = items[0] - # Prefer the guid extracted from manifest_script href manifest_script = getattr(item, "manifest_script", None) + + # If manifest_script is a vodproductinformation URL the content + # is not playable (subscriptionMissing / not entitled). + # Stop here rather than passing an unplayable URL to the SMIL chain. + if manifest_script and "vodproductinformation" in manifest_script: + logger.warning( + f"{gn_id}: content not entitled (manifest_script is " + f"vodproductinformation URL — subscriptionMissing)" + ) + return None + if manifest_script: # href format: https://link.theplatform.eu/s/mdeprod/media/zRPPGLNGgPa1 guid = manifest_script.rstrip("/").rsplit("/media/", 1)[-1].split("?")[0] diff --git a/lib/streaming_providers/providers/magenta2/vod_manager.py b/lib/streaming_providers/providers/magenta2/vod_manager.py index 2627753..621802a 100644 --- a/lib/streaming_providers/providers/magenta2/vod_manager.py +++ b/lib/streaming_providers/providers/magenta2/vod_manager.py @@ -285,6 +285,11 @@ class VodManager: def _base_params(self) -> Dict[str, str]: """Build query parameters common to every tvhubs VOD API request. + Note: $subscriberType is intentionally omitted — the real Android TV + device does NOT send it on tvhubs calls (StructuredGrid, VodDetails, + PersonalBar, UnstructuredGrid etc.). It is only sent on wcps calls + (vodproductinformation, VodPlayer) via _playback_params(). + Session correlation follows the platform convention: - Android TV: ``$cid`` = ``::`` (no sid / t) - Web / other: ``sid`` + ``t`` (no $cid) @@ -292,7 +297,6 @@ class VodManager: params: Dict[str, str] = { "$deviceModel": self._device_model, "$profile": self._profile_name, - "$subscriberType": self._subscriber_type, "$theme": self._theme_id, "$redirect": "false", } @@ -325,31 +329,18 @@ class VodManager: def _get(self, url: str, params: Dict) -> Optional[Dict]: """ - Perform an authenticated GET request, injecting all device-identity - headers on every call: + Perform a GET request and return the parsed JSON body, or None on error. - Bearer + x-mpx-authorization — from auth_headers_callback - x-stbserialnumber — stable device serial - dt-session-id — stable session ID - dt-call-id — fresh UUID per request - - The server uses all of these to scope responses (partner lists, - entitlements, personalisation) to the correct device session. - Omitting the serial/session headers causes the server to return - generic or reduced responses. + Auth + device-identity headers are injected by auth_headers_callback + (Bearer, x-mpx-authorization, x-stbserialnumber, dt-session-id, + dt-call-id) — all tvhubs and wcps VOD endpoints require them. """ - import uuid as _iuuid - headers: Dict[str, str] = {} + headers = None if self._auth_headers_callback: try: - headers = dict(self._auth_headers_callback()) + headers = self._auth_headers_callback() except Exception as exc: logger.warning(f"{self._provider}: auth_headers_callback failed: {exc}") - if self._serial_number: - headers["x-stbserialnumber"] = self._serial_number - if self._session_id: - headers["dt-session-id"] = self._session_id - headers["dt-call-id"] = str(_iuuid.uuid4()) try: response = self._http.get(url, params=params, headers=headers) if response and response.status_code == 200: @@ -363,7 +354,7 @@ class VodManager: return None def _get_auth(self, url: str, params: Dict) -> Optional[Dict]: - """Alias for _get.""" + """Alias for _get — auth is always injected when callback is set.""" return self._get(url, params) def _get_no_auth(self, url: str, params: Dict) -> Optional[Dict]: @@ -1495,8 +1486,7 @@ class VodManager: "$reloadAfterChange": "false", } if self._is_androidtv(): - import uuid as _uuid - iup_params["$cid"] = f"{self._session_id}::{str(_uuid.uuid4())}" + iup_params["$cid"] = f"{self._session_id}::{str(_uuid_mod.uuid4())}" else: iup_params["sid"] = self._session_id iup_params["t"] = str(int(time.time() * 1000))