diff --git a/lib/streaming_providers/providers/discovery/channel_manager.py b/lib/streaming_providers/providers/discovery/channel_manager.py new file mode 100644 index 0000000..eeee868 --- /dev/null +++ b/lib/streaming_providers/providers/discovery/channel_manager.py @@ -0,0 +1,277 @@ +# streaming_providers/providers/discovery/channel_manager.py +""" +Discovery+ Channel Manager + +Handles CMS channel extraction, distribution channel parsing, +and CMS navigation route discovery from the /home response graph. +""" +from typing import Dict, List, Optional + +from ...base.models import StreamingChannel +from ...base.utils.logger import logger + +from .constants import ( + CMS_INCLUDE_PARAMS, + CMS_PAGE_SIZE, + CMS_ROUTE_SPORT_SCHEDULE, +) +from .models import DiscoveryChannel + + +class DiscoveryChannelManager: + """ + Manages channel extraction and CMS route discovery for Discovery+. + + Responsibilities: + - Fetching and parsing distribution channels from the CMS /home route + - Maintaining the ``_channels_cache`` (channel_id → DiscoveryChannel) + - Discovering and caching navigation routes from the /home response graph + into ``_cms_routes`` (route_id → label) + + Both caches are owned by the provider and passed in by reference so that + the event manager and playback manager can share them without coupling + directly to this class. + """ + + def __init__( + self, + provider, # DiscoveryProvider — avoid circular import with a late reference + channels_cache: Dict[str, DiscoveryChannel], + cms_routes: Dict[str, str], + ): + self._provider = provider + self._channels_cache = channels_cache + self._cms_routes = cms_routes + + # ========================================================================= + # Public interface + # ========================================================================= + + def fetch_channels( + self, + fetch_manifests: bool = False, + populate_streaming_data: bool = True, + ) -> List[StreamingChannel]: + """ + Fetch available channels from the Discovery+ CMS /home route. + + Populates ``_channels_cache`` with DiscoveryChannel objects and + refreshes ``_cms_routes`` from the navigation graph embedded in the + same response (avoiding a second round-trip). + + Args: + fetch_manifests: Whether to immediately populate streaming data. + populate_streaming_data: Whether to populate streaming data. + + Returns: + List of StreamingChannel objects. + """ + try: + headers = self._provider.get_auth_headers() + url = self._provider.authenticator.cms_home_endpoint + params = { + "include": CMS_INCLUDE_PARAMS, + "decorators": "viewingHistory,isFavorite,contentAction,badges", + "page[items.size]": CMS_PAGE_SIZE, + } + + response = self._provider.http_manager.get( + url, + operation="cms", + headers=headers, + params=params, + ) + response.raise_for_status() + data = response.json() + + # Refresh CMS route cache from this /home response. + refreshed = self._discover_cms_routes(data) + if refreshed: + self._cms_routes.clear() + self._cms_routes.update(refreshed) + logger.debug( + f"CMS routes refreshed: {list(self._cms_routes.keys())}" + ) + + # Extract channels and populate the shared cache + discovery_channels = self._extract_distribution_channels(data) + self._channels_cache.clear() + self._channels_cache.update( + {ch.channel_id: ch for ch in discovery_channels} + ) + + # Convert to StreamingChannel objects + streaming_channels = [ + ch.to_streaming_channel(self._provider.provider_name) + for ch in discovery_channels + ] + + logger.info( + f"Found {len(streaming_channels)} distribution channels" + ) + + if fetch_manifests and populate_streaming_data and streaming_channels: + streaming_channels = ( + self._provider.playback_manager.populate_streaming_data( + streaming_channels + ) + ) + + return streaming_channels + + except Exception as e: + logger.error(f"Error fetching channels: {e}") + return [] + + def init_cms_routes(self) -> None: + """ + Fetch ``/home`` and populate ``_cms_routes`` eagerly. + + Called once at the end of ``DiscoveryProvider.__init__`` so route + discovery is complete before any call to ``get_events()`` or + ``get_channels()``. Failures are caught and logged — the provider + remains usable and ``get_events()`` will fall back to + ``CMS_ROUTE_SPORT_SCHEDULE``. + """ + try: + headers = self._provider.get_auth_headers() + url = self._provider.authenticator.cms_home_endpoint + params = { + "include": CMS_INCLUDE_PARAMS, + "decorators": "viewingHistory,isFavorite,contentAction,badges", + "page[items.size]": CMS_PAGE_SIZE, + } + response = self._provider.http_manager.get( + url, + operation="cms", + headers=headers, + params=params, + ) + response.raise_for_status() + data = response.json() + routes = self._discover_cms_routes(data) + self._cms_routes.update(routes) + if routes: + logger.debug( + f"CMS routes discovered at init: {list(self._cms_routes.keys())}" + ) + else: + logger.warning( + "CMS route discovery returned no routes — " + f"get_events() will fall back to '{CMS_ROUTE_SPORT_SCHEDULE}'" + ) + except Exception as e: + logger.warning(f"Could not discover CMS routes at init: {e}") + + # ========================================================================= + # CMS route discovery + # ========================================================================= + + @staticmethod + def _discover_cms_routes(data: dict) -> Dict[str, str]: + """ + Parse a /home CMS response and return a map of all navigation routes. + + The /home response contains a ``data.relationships.navigationLinks`` + array whose entries are resolved via the top-level ``included`` list. + Each resolved node has ``attributes.url`` (e.g. ``"/sports"``) and + ``attributes.label`` (localised display name). + + Returns: + dict mapping route_id → label, e.g. + ``{"sports": "Sport", "channels": "Kanäle", "home": "Home"}`` + Returns an empty dict if the navigation graph is absent. + """ + included_by_id: Dict[str, dict] = { + f"{item['type']}:{item['id']}": item + for item in data.get("included", []) + if item.get("id") and item.get("type") + } + nav_refs = ( + data.get("data", {}) + .get("relationships", {}) + .get("navigationLinks", {}) + .get("data", []) + ) + if not nav_refs: + logger.debug( + "_discover_cms_routes: no navigationLinks found in /home response" + ) + return {} + + routes: Dict[str, str] = {} + for ref in nav_refs: + key = f"{ref.get('type', 'link')}:{ref['id']}" + node = included_by_id.get(key, {}) + attrs = node.get("attributes", {}) + url_path = attrs.get("url", "").lstrip("/") + route_id = url_path or ref["id"] + label = attrs.get("label", route_id) + routes[route_id] = label + + return routes + + # ========================================================================= + # Channel extraction helpers + # ========================================================================= + + def _extract_distribution_channels( + self, data: Dict + ) -> List[DiscoveryChannel]: + """ + Extract distributionChannel objects from a CMS response. + + Args: + data: Parsed JSON response from the CMS home route. + + Returns: + List of DiscoveryChannel objects. + """ + channels = [] + + included_by_id: Dict[str, dict] = {} + for item in data.get("included", []): + item_id = item.get("id") + item_type = item.get("type") + if item_id and item_type: + included_by_id[f"{item_type}:{item_id}"] = item + + for item in data.get("included", []): + if item.get("type") == "distributionChannel": + channel = self._create_discovery_channel_from_distribution( + item, included_by_id + ) + if channel: + channels.append(channel) + + return channels + + @staticmethod + def _create_discovery_channel_from_distribution( + distribution_data: Dict, + included_by_id: Dict, + ) -> Optional[DiscoveryChannel]: + """ + Create a DiscoveryChannel from a distributionChannel CMS object. + + Args: + distribution_data: The distributionChannel item from the CMS. + included_by_id: Lookup dict of all included items (``type:id`` → item). + + Returns: + DiscoveryChannel instance, or None if the data is invalid. + """ + try: + discovery_channel = DiscoveryChannel.from_distribution_data( + distribution_data, included_by_id + ) + logger.debug( + f"Created channel: {discovery_channel.name} " + f"(edit_id: {discovery_channel.edit_id})" + ) + return discovery_channel + except Exception as e: + logger.error( + f"Error creating channel from distribution data: {e}" + ) + return None \ No newline at end of file diff --git a/lib/streaming_providers/providers/discovery/event_manager.py b/lib/streaming_providers/providers/discovery/event_manager.py new file mode 100644 index 0000000..a3c7bc1 --- /dev/null +++ b/lib/streaming_providers/providers/discovery/event_manager.py @@ -0,0 +1,757 @@ +# streaming_providers/providers/discovery/event_manager.py +""" +Discovery+ Event Manager + +Handles sport schedule and live event fetching from the Discovery+ CMS. +Supports two parse paths: + 1. /sports — airing items directly in ``included`` (fast path) + 2. /sport-schedule — items nested inside a collection (legacy fallback) +""" +from datetime import datetime, timezone +from typing import ClassVar, Dict, List, Optional + +from ...base.models import Event, EventStatus +from ...base.utils.logger import logger + +from .constants import ( + AIRING_BADGE_LIVE, + AIRING_BADGE_UP_NEXT, + AIRING_BADGE_UPCOMING_LEGACY, + AIRING_ITEM_TYPE, + CMS_INCLUDE_PARAMS, + CMS_ROUTE_SPORT_SCHEDULE, + CMS_ROUTE_SPORTS, +) + + +class DiscoveryEventManager: + """ + Manages event (sport schedule) fetching for Discovery+. + + Responsibilities: + - Dynamic CMS route selection from the cached ``_cms_routes`` map + - Airing-based event parsing (``/sports`` route — fast path) + - Collection-based event traversal (``/sport-schedule`` — legacy fallback) + - Taxonomy relationship resolution (genre, competition, venue, …) + + The shared ``_channels_cache`` and ``_cms_routes`` dicts are owned by the + provider and passed in so the manager can look up channel names and pick + the right route without coupling to the provider itself. + """ + + # Route preference order — first matching discovered route wins. + # ``sport-schedule`` (paginated full schedule, ~300 events) is preferred + # over ``sports`` (featured/current airings only, ~14 events). + _EVENT_ROUTE_PREFERENCE: ClassVar[tuple] = ( + CMS_ROUTE_SPORT_SCHEDULE, # "sport-schedule" — full paginated schedule + CMS_ROUTE_SPORTS, # "sports" — featured airings only (fallback) + ) + # Generic keyword scan used when neither preferred route is discovered + _EVENT_ROUTE_KEYWORDS: ClassVar[tuple] = ("schedule", "sport", "live", "event") + + def __init__(self, provider, channels_cache: Dict, cms_routes: Dict): + self._provider = provider + self._channels_cache = channels_cache + self._cms_routes = cms_routes + + # ========================================================================= + # Public interface + # ========================================================================= + + def fetch_events( + self, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + **kwargs, + ) -> List[Event]: + """ + Fetch events from the Discovery+ schedule. + + **Route selection** (no ``route_id`` kwarg supplied): + Calls ``_select_event_route()`` which prefers a route discovered + dynamically from the ``/home`` navigation graph (populated by + ``get_channels()``). Falls back to ``CMS_ROUTE_SPORT_SCHEDULE`` + if no suitable route is found. + + **Parse paths** (determined by the selected route's response shape): + 1. ``/sports`` — ``airing`` items appear directly in ``included`` + (fast path, ``_parse_airing_events()``). + 2. ``/sport-schedule`` — items are nested inside a collection + (legacy fallback, ``_fetch_events_via_collection()``). + The fast path is attempted first; if no ``airing`` items are found + the method retries with ``CMS_ROUTE_SPORT_SCHEDULE``. + + Args: + start_time: Optional start time filter. + end_time: Optional end time filter. + **kwargs: Additional parameters including: + - route_id: Override the CMS route (bypasses dynamic selection). + - page_size: Number of items per page (collection path only). + + Returns: + List of Event objects. + """ + events = [] + + try: + headers = self._provider.get_auth_headers() + + route_id = kwargs.get("route_id") or self._select_event_route() + + # Build the CMS route URL + base_url = self._provider.authenticator.cms_home_endpoint + if base_url.endswith("/home"): + url = base_url.replace("/home", f"/{route_id}") + else: + url = ( + f"https://default.any-{self._provider.authenticator.home_market}" + f".{self._provider.authenticator.env}.api.discoveryplus.com" + f"/cms/routes/{route_id}" + ) + + params = { + "include": CMS_INCLUDE_PARAMS, + "decorators": "viewingHistory,isFavorite,contentAction,badges", + "page[items.size]": kwargs.get("page_size", 10), + } + if start_time: + params["filter[from]"] = start_time.isoformat() + if end_time: + params["filter[to]"] = end_time.isoformat() + + logger.debug(f"Fetching events route: {url}") + response = self._provider.http_manager.get( + url, + operation="cms", + headers=headers, + params=params, + ) + response.raise_for_status() + data = response.json() + + # ------------------------------------------------------------------ + # Fast path: /sports route returns airing items directly in included + # ------------------------------------------------------------------ + airing_items = [ + item for item in data.get("included", []) + if item.get("type") == AIRING_ITEM_TYPE + ] + + if airing_items: + logger.debug( + f"Found {len(airing_items)} airing items in /sports route — " + "using direct airing parse path" + ) + events = self._parse_airing_events( + airing_items, start_time, end_time + ) + logger.info(f"Found {len(events)} events (airing path)") + return events + + # ------------------------------------------------------------------ + # Slow path: collection-based traversal (legacy /sport-schedule shape) + # ------------------------------------------------------------------ + if route_id == CMS_ROUTE_SPORTS: + logger.debug( + "No airing items on /sports route — falling back to " + f"/{CMS_ROUTE_SPORT_SCHEDULE} collection path" + ) + return self.fetch_events( + start_time=start_time, + end_time=end_time, + route_id=CMS_ROUTE_SPORT_SCHEDULE, + **{k: v for k, v in kwargs.items() if k != "route_id"}, + ) + + events = self._fetch_events_via_collection( + data=data, + headers=headers, + start_time=start_time, + end_time=end_time, + **kwargs, + ) + logger.info(f"Found {len(events)} events (collection path)") + + except Exception as e: + logger.error(f"Error fetching events: {e}") + + return events + + # ========================================================================= + # Route selection + # ========================================================================= + + def _select_event_route(self) -> str: + """ + Return the best event-producing CMS route from ``_cms_routes``. + + Preference order: + 1. ``CMS_ROUTE_SPORT_SCHEDULE`` — full paginated schedule (~300 events). + 2. ``CMS_ROUTE_SPORTS`` — featured/current airings only (~14). + 3. First discovered route matching a keyword in + ``_EVENT_ROUTE_KEYWORDS`` (generic fallback for unknown layouts). + 4. ``CMS_ROUTE_SPORT_SCHEDULE`` hard-coded constant when + ``_cms_routes`` is empty (e.g. ``get_channels()`` hasn't been + called yet). + """ + for preferred in self._EVENT_ROUTE_PREFERENCE: + if preferred in self._cms_routes: + logger.debug( + f"_select_event_route: selected preferred route '{preferred}'" + ) + return preferred + + for keyword in self._EVENT_ROUTE_KEYWORDS: + match = next( + (r for r in self._cms_routes if keyword in r.lower()), None + ) + if match: + logger.debug( + f"_select_event_route: selected '{match}' via keyword '{keyword}'" + ) + return match + + logger.debug( + f"_select_event_route: no event route found in " + f"{list(self._cms_routes.keys())!r}, " + f"falling back to '{CMS_ROUTE_SPORT_SCHEDULE}'" + ) + return CMS_ROUTE_SPORT_SCHEDULE + + # ========================================================================= + # Airing-based event parsing (/sports route — fast path) + # ========================================================================= + + def _parse_airing_events( + self, + airing_items: List[Dict], + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + ) -> List[Event]: + """ + Parse a list of ``airing`` objects from the /sports CMS route into + ``Event`` instances. + + The /sports route places ``airing`` items directly in the top-level + ``included`` array. Each airing carries: + + - ``attributes.scheduleStart / scheduleEnd`` — ISO-8601 UTC timestamps + - ``attributes.name`` — episode/event title + - ``attributes.showName`` — parent show name + - ``attributes.description / secondaryTitle`` — optional metadata + - ``attributes.episodeNumber / seasonNumber`` — episode metadata + - ``relationships.badges`` — live / up-next status + - ``relationships.distributionChannel`` — links to channel cache + + Badge → EventStatus mapping: + AIRING_BADGE_LIVE ("live") → EventStatus.LIVE + AIRING_BADGE_UP_NEXT ("release-state-up-next") → EventStatus.UPCOMING + (fallback clock-based logic handles ENDED / LIVE for unlabelled items) + + Args: + airing_items: List of airing dicts from ``data["included"]``. + start_time: Optional lower bound filter (inclusive). + end_time: Optional upper bound filter (inclusive). + + Returns: + List of Event objects, optionally filtered by time window. + """ + events: List[Event] = [] + now_utc = datetime.now(timezone.utc) + + for airing in airing_items: + try: + attributes = airing.get("attributes", {}) + relationships = airing.get("relationships", {}) + + # ---- schedule times ----------------------------------------- + start_dt, end_dt = self._parse_schedule_times(attributes) + + # ---- optional caller-supplied time window filter ------------- + if start_time and end_dt and end_dt < start_time: + continue + if end_time and start_dt and start_dt > end_time: + continue + + # ---- event status from badges -------------------------------- + status = self._resolve_badge_status( + relationships, now_utc, start_dt, end_dt + ) + + # ---- channel lookup ----------------------------------------- + channel_name: Optional[str] = None + dist_channel_ref = ( + relationships.get("distributionChannel", {}).get("data", {}) + ) + dist_channel_id = ( + dist_channel_ref.get("id") if dist_channel_ref else None + ) + if dist_channel_id: + cached = self._channels_cache.get(dist_channel_id) + if cached: + channel_name = cached.name + else: + logger.debug( + f"distributionChannel {dist_channel_id} not in cache " + f"for airing '{attributes.get('name')}'" + ) + + # ---- build Event -------------------------------------------- + event = Event( + name=attributes.get("name", "Unknown Event"), + content_id=airing.get("id", ""), + provider=self._provider.provider_name, + logo_url=None, + mode="live" if status == EventStatus.LIVE else "vod", + session_manifest=False, + manifest_script=None, + cdm=None, + content_type="AIRING", + description=attributes.get("description", ""), + genre=None, + language="de", + country=self._provider.country.upper(), + start_time=start_dt, + end_time=end_dt, + status=status, + subtitle=attributes.get("secondaryTitle"), + original_name=attributes.get("showName"), + competition=None, + venue=None, + gender=None, + discipline=None, + age_category=None, + master_event=None, + channel=channel_name, + ) + events.append(event) + + except Exception as e: + logger.error( + f"Error processing airing item {airing.get('id')}: {e}" + ) + continue + + return events + + # ========================================================================= + # Legacy collection-based event fetching (/sport-schedule route) + # ========================================================================= + + def _fetch_events_via_collection( + self, + data: Dict, + headers: Dict, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + **kwargs, + ) -> List[Event]: + """ + Traverse the legacy collection-based CMS shape to extract events. + + This implements the original multi-hop traversal: + route data → target page → page items → collection → + paginated collection items → video attributes + + Args: + data: Parsed JSON from the initial CMS route request. + headers: Auth headers for subsequent requests. + start_time: Optional lower bound filter. + end_time: Optional upper bound filter. + **kwargs: Forwarded kwargs (page_size, etc.). + + Returns: + List of Event objects. + """ + events: List[Event] = [] + + included_by_id: Dict[str, Dict] = { + f"{item.get('type')}:{item.get('id')}": item + for item in data.get("included", []) + if item.get("id") and item.get("type") + } + + # Find the target page from the route + route_data = data.get("data", {}) + target_ref = ( + route_data.get("relationships", {}) + .get("target", {}) + .get("data") + ) + if not target_ref: + logger.error( + "No target page found in route response (collection path)" + ) + return events + + page_key = f"{target_ref.get('type')}:{target_ref.get('id')}" + page = included_by_id.get(page_key) + if not page: + logger.error( + "Target page not found in included items (collection path)" + ) + return events + + # Find the collection from page items + page_items = ( + page.get("relationships", {}).get("items", {}).get("data", []) + ) + collection_id = None + for page_item_ref in page_items: + page_item_key = ( + f"{page_item_ref.get('type')}:{page_item_ref.get('id')}" + ) + page_item = included_by_id.get(page_item_key) + if page_item: + collection_ref = ( + page_item.get("relationships", {}) + .get("collection", {}) + .get("data") + ) + if collection_ref: + collection_id = collection_ref.get("id") + break + + if not collection_id: + logger.error("No collection found in page (collection path)") + return events + + # Paginate through the collection + collection_url = ( + f"{self._provider.authenticator.cms_collections_endpoint}" + f"/{collection_id}" + ) + current_page = 1 + + while True: + collection_params = { + "include": "default", + "decorators": "viewingHistory,isFavorite,contentAction,badges", + "page[items.size]": kwargs.get("page_size", 30), + "page[items.number]": current_page, + } + + logger.debug( + f"Fetching schedule collection page {current_page}: " + f"{collection_url}" + ) + collection_response = self._provider.http_manager.get( + collection_url, + operation="cms", + headers=headers, + params=collection_params, + ) + collection_response.raise_for_status() + collection_data = collection_response.json() + + page_included: Dict[str, Dict] = { + f"{item.get('type')}:{item.get('id')}": item + for item in collection_data.get("included", []) + if item.get("id") and item.get("type") + } + + items = ( + collection_data.get("data", {}) + .get("relationships", {}) + .get("items", {}) + .get("data", []) + ) + + meta = collection_data.get("meta", {}) + total_pages = meta.get("itemsTotalPages", 1) + logger.debug( + f"Fetched collection page {current_page}/{total_pages} " + f"({len(items)} items)" + ) + + for item_ref in items: + try: + item_key = ( + f"{item_ref.get('type')}:{item_ref.get('id')}" + ) + collection_item = page_included.get(item_key) + if not collection_item: + continue + + video_ref = ( + collection_item.get("relationships", {}) + .get("video", {}) + .get("data") + ) + if not video_ref: + continue + + video_key = ( + f"{video_ref.get('type')}:{video_ref.get('id')}" + ) + video_data = page_included.get(video_key) + if not video_data: + continue + + attributes = video_data.get("attributes", {}) + relationships = video_data.get("relationships", {}) + + if attributes.get("materialType") != "EVENT": + continue + + # ---- status from badges ---------------------------------- + now_utc = datetime.now(timezone.utc) + status = EventStatus.SCHEDULED + badge_refs = relationships.get("badges", {}).get("data", []) + badge_ids = {b.get("id") for b in badge_refs} + + if AIRING_BADGE_LIVE in badge_ids: + status = EventStatus.LIVE + elif ( + AIRING_BADGE_UP_NEXT in badge_ids + or AIRING_BADGE_UPCOMING_LEGACY in badge_ids + ): + status = EventStatus.SCHEDULED + + for badge_ref in badge_refs: + overlay_key = ( + f"{badge_ref.get('type')}:{badge_ref.get('id')}" + ) + overlay = page_included.get(overlay_key, {}) + if overlay.get("id") == "live:default": + status = EventStatus.LIVE + break + + # ---- schedule times ------------------------------------- + start_dt, end_dt = self._parse_schedule_times(attributes) + + # Clock-based status refinement + if status != EventStatus.LIVE and start_dt and end_dt: + _start = ( + start_dt if start_dt.tzinfo + else start_dt.replace(tzinfo=timezone.utc) + ) + _end = ( + end_dt if end_dt.tzinfo + else end_dt.replace(tzinfo=timezone.utc) + ) + if now_utc > _end: + status = EventStatus.ENDED + elif _start <= now_utc <= _end: + status = EventStatus.LIVE + + # ---- logo ----------------------------------------------- + logo_url = None + image_refs = relationships.get("images", {}).get("data", []) + if image_refs: + image_key = ( + f"{image_refs[0].get('type')}:{image_refs[0].get('id')}" + ) + image = page_included.get(image_key, {}) + img_attrs = image.get("attributes", {}) + logo_url = img_attrs.get("src") or img_attrs.get("url") + + # ---- edit ID for streaming -------------------------------- + edit_ref = ( + relationships.get("edit", {}).get("data", {}) + ) + edit_id = edit_ref.get("id") if edit_ref else None + + # ---- taxonomy lookups ----------------------------------- + genre = self._resolve_taxonomy( + relationships, "txSports", page_included + ) + competition = self._resolve_taxonomy( + relationships, "txCompetition", page_included + ) + venue = self._resolve_taxonomy( + relationships, "txEvent", page_included + ) + gender_val = self._resolve_taxonomy( + relationships, "txGender", page_included + ) + gender = ( + gender_val + if gender_val and gender_val != "." + else None + ) + discipline_val = self._resolve_taxonomy( + relationships, "txDiscipline", page_included + ) + discipline = ( + discipline_val + if discipline_val and discipline_val != "." + else None + ) + age_val = self._resolve_taxonomy( + relationships, "txAge", page_included + ) + age_category = ( + age_val if age_val and age_val != "." else None + ) + master_event = self._resolve_taxonomy( + relationships, "txMaster-sporting-event", page_included + ) + + # ---- primary channel ------------------------------------ + channel_name = None + channel_ref = ( + relationships.get("primaryChannel", {}).get("data") + ) + if channel_ref: + channel_key = ( + f"{channel_ref.get('type')}:{channel_ref.get('id')}" + ) + channel_node = page_included.get(channel_key, {}) + channel_name = ( + channel_node.get("attributes", {}).get("name") + ) + + # ---- language from audio tracks ------------------------- + audio_tracks = attributes.get("audioTracks", []) + language = "de" + if audio_tracks: + if "Deutsch" in audio_tracks: + language = "de" + elif "Englisch" in audio_tracks: + language = "en" + + event = Event( + name=attributes.get("name", "Unknown Event"), + content_id=video_data.get("id", ""), + provider=self._provider.provider_name, + logo_url=logo_url, + mode=( + "live" + if attributes.get("videoType") == "LIVE" + else "vod" + ), + session_manifest=True, + manifest_script=( + f"editid={edit_id}" if edit_id else None + ), + cdm=f"editid={edit_id}" if edit_id else None, + content_type="EVENT", + description=attributes.get("description", ""), + genre=genre, + language=language, + country=self._provider.country.upper(), + start_time=start_dt, + end_time=end_dt, + status=status, + subtitle=attributes.get("secondaryTitle"), + original_name=attributes.get("originalName"), + competition=competition, + venue=venue, + gender=gender, + discipline=discipline, + age_category=age_category, + master_event=master_event, + channel=channel_name, + ) + events.append(event) + + except Exception as e: + logger.error(f"Error processing collection item: {e}") + continue + + if current_page >= total_pages: + break + current_page += 1 + + return events + + # ========================================================================= + # Shared parsing helpers + # ========================================================================= + + @staticmethod + def _parse_schedule_times( + attributes: Dict, + ): + """ + Extract and parse scheduleStart / scheduleEnd from an attributes dict. + + Returns: + Tuple of (start_dt, end_dt), either of which may be None. + """ + start_dt: Optional[datetime] = None + end_dt: Optional[datetime] = None + + for attr_key, target in ( + ("scheduleStart", "start_dt"), + ("scheduleEnd", "end_dt"), + ): + raw = attributes.get(attr_key) + if raw: + try: + parsed = datetime.fromisoformat( + raw.replace("Z", "+00:00") + ) + if target == "start_dt": + start_dt = parsed + else: + end_dt = parsed + except (ValueError, TypeError): + pass + + return start_dt, end_dt + + @staticmethod + def _resolve_badge_status( + relationships: Dict, + now_utc: datetime, + start_dt: Optional[datetime], + end_dt: Optional[datetime], + ) -> EventStatus: + """ + Derive EventStatus from badge relationships, falling back to clock logic. + + Badge priority: explicit "live" wins over "up-next"; clock-based logic + is applied when neither badge is present. + """ + badge_refs = relationships.get("badges", {}).get("data", []) + badge_ids = {b.get("id") for b in badge_refs} + + if AIRING_BADGE_LIVE in badge_ids: + return EventStatus.LIVE + if ( + AIRING_BADGE_UP_NEXT in badge_ids + or AIRING_BADGE_UPCOMING_LEGACY in badge_ids + ): + return EventStatus.SCHEDULED + + if start_dt and end_dt: + _start = ( + start_dt if start_dt.tzinfo + else start_dt.replace(tzinfo=timezone.utc) + ) + _end = ( + end_dt if end_dt.tzinfo + else end_dt.replace(tzinfo=timezone.utc) + ) + if now_utc > _end: + return EventStatus.ENDED + if _start <= now_utc <= _end: + return EventStatus.LIVE + + return EventStatus.SCHEDULED + + @staticmethod + def _resolve_taxonomy( + relationships: Dict, + key: str, + included: Dict, + ) -> Optional[str]: + """ + Resolve a single taxonomy relationship to its ``attributes.name`` value. + + Args: + relationships: The ``relationships`` dict from a video/event item. + key: Relationship key, e.g. ``"txSports"``, ``"txCompetition"``. + included: Per-page included lookup (``type:id`` → item). + + Returns: + The ``attributes.name`` string, or ``None`` if not found. + """ + refs = relationships.get(key, {}).get("data", []) + if not refs: + return None + ref = refs[0] + node = included.get(f"{ref.get('type')}:{ref.get('id')}", {}) + return node.get("attributes", {}).get("name") \ No newline at end of file diff --git a/lib/streaming_providers/providers/discovery/playback_manager.py b/lib/streaming_providers/providers/discovery/playback_manager.py new file mode 100644 index 0000000..0bfbfc1 --- /dev/null +++ b/lib/streaming_providers/providers/discovery/playback_manager.py @@ -0,0 +1,453 @@ +# streaming_providers/providers/discovery/playback_manager.py +""" +Discovery+ Playback Manager + +Handles all streaming/playback concerns: +- Fetching and caching playback info (tokens valid for 24 hours) +- Extracting manifest URLs and DRM details from playback responses +- Building DRMConfig objects +- Populating StreamingChannel objects with streaming data +""" +import base64 +import json +import secrets +import time +import uuid +from datetime import datetime +from typing import Any, Dict, List, Optional + +from ...base.models import DRMConfig, DRMSystem, LicenseConfig, StreamingChannel +from ...base.utils.logger import logger + +from .constants import PlatformOS, get_default_capabilities, get_default_device_info, get_drm_request_headers +from .exceptions import ManifestFetchError, PlaybackRestrictedException +from .models import DiscoveryChannel + + +class DiscoveryPlaybackManager: + """ + Manages playback info fetching and streaming data population for Discovery+. + + Responsibilities: + - Building the playback request payload (device capabilities, DRM, ads) + - Fetching playback info from the Discovery+ playback endpoint + - Caching playback tokens keyed on ``edit_id`` until DRM expiration + - Extracting manifest URLs and DRM scheme details from playback responses + - Building DRMConfig objects for Widevine / PlayReady + - Populating a list of StreamingChannel objects (``populate_streaming_data``) + + The shared ``_channels_cache`` dict is owned by the provider and passed in + so the manager can resolve ``edit_id`` values without extra API calls. + """ + + def __init__( + self, + provider, # DiscoveryProvider — avoid circular import + channels_cache: Dict[str, DiscoveryChannel], + playback_cache: Dict[str, tuple], + ): + self._provider = provider + self._channels_cache = channels_cache + # {edit_id: (expiry_timestamp, playback_data)} + self._playback_cache = playback_cache + + # ========================================================================= + # Public interface + # ========================================================================= + + def populate_streaming_data( + self, + channels: List[StreamingChannel], + max_retries: int = 3, + ) -> List[StreamingChannel]: + """ + Populate streaming data for a list of StreamingChannel objects. + + Looks up each channel's ``edit_id`` in the shared ``_channels_cache``, + fetches (or returns cached) playback info, and attaches the manifest + URL, streaming format, and DRM config to the channel object. + + Args: + channels: StreamingChannel objects to populate. + max_retries: Maximum retry attempts per channel. + + Returns: + List of successfully populated channels. + """ + 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: + disco_channel = self._channels_cache.get(channel.channel_id) + if not disco_channel: + logger.warning( + f"Channel {channel.name} not in cache, skipping" + ) + break + + edit_id = disco_channel.edit_id + if not edit_id: + logger.warning( + f"No edit_id for channel {channel.name}, skipping" + ) + break + + logger.debug( + f"Getting playback info for: {channel.name} " + f"(edit_id: {edit_id}, attempt {retries + 1})" + ) + + playback_data = self.get_cached_playback_info( + edit_id=edit_id + ) + streaming_data = self.extract_streaming_data(playback_data) + + if streaming_data["manifest_url"]: + channel.manifest = streaming_data["manifest_url"] + channel.streaming_format = streaming_data[ + "streaming_format" + ] + + if streaming_data["license_url"]: + drm_config = self.build_drm_config(streaming_data) + if drm_config: + channel.drm_config = drm_config + channel.license_url = streaming_data["license_url"] + channel.cdm_type = streaming_data["drm_system"] + + if streaming_data["drm_auth"]: + disco_channel.raw_data["drm_auth"] = ( + streaming_data["drm_auth"] + ) + + logger.info( + f"Streaming data populated for: {channel.name}" + ) + successful_channels.append(channel) + success = True + else: + raise ManifestFetchError("No manifest URL in response") + + except PlaybackRestrictedException 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 " + f"{channel.name}: {e}" + ) + time.sleep(1) + else: + logger.error( + f"Failed to get streaming data for " + f"{channel.name}: {e}" + ) + + logger.info( + f"Streaming data population complete: " + f"{len(successful_channels)}/{len(channels)}" + ) + return successful_channels + + def get_playback_info(self, edit_id: str, **kwargs) -> Dict: + """ + Fetch playback information for the given edit_id. + + Args: + edit_id: Content edit ID. + **kwargs: Additional parameters (unused, reserved for future use). + + Returns: + Playback info dictionary from the Discovery+ playback endpoint. + + Raises: + ValueError: If edit_id is not provided. + PlaybackRestrictedException: If the server returns PLAYBACK_RESTRICTED. + ManifestFetchError: On any other HTTP / parse error. + """ + if not edit_id: + raise ValueError("edit_id is required for playback") + + headers = self._provider.get_auth_headers() + payload = self._build_playback_payload(edit_id) + + try: + response = self._provider.http_manager.post( + self._provider.authenticator.playback_endpoint, + operation="playback", + headers=headers, + json=payload, + timeout=30, + ) + + if response.status_code >= 400: + error_text = response.text + logger.error(f"Playback error response: {error_text}") + if "PLAYBACK_RESTRICTED" in error_text: + raise PlaybackRestrictedException( + reason=f"Playback restricted for {edit_id}", + error_code="PLAYBACK_RESTRICTED", + ) + response.raise_for_status() + + response.raise_for_status() + return response.json() + + except PlaybackRestrictedException: + raise + except Exception as e: + raise ManifestFetchError( + f"Error getting playback info for edit_id {edit_id}: {e}" + ) + + # ========================================================================= + # Playback cache + # ========================================================================= + + def get_cached_playback_info(self, edit_id: str) -> Dict: + """ + Return playback info for edit_id, using a time-based cache. + + Tokens issued by Discovery+ are valid for 24 hours. The cache entry + is kept until the ``drm_expiration`` timestamp returned in the response + (minus a 60-second safety margin). If no expiration is present a + fallback TTL of 23 hours is used. + + Args: + edit_id: Content edit ID. + + Returns: + Playback info dictionary (from cache or freshly fetched). + """ + now = time.time() + cached = self._playback_cache.get(edit_id) + + if cached: + expiry, data = cached + if now < expiry: + logger.debug( + f"Playback cache hit for edit_id {edit_id} " + f"(expires in {int(expiry - now)}s)" + ) + return data + else: + logger.debug( + f"Playback cache expired for edit_id {edit_id}, re-fetching" + ) + + data = self.get_playback_info(edit_id=edit_id) + + expiry = None + try: + drm_expiration_str = data.get("drm", {}).get("expirationDate") + if drm_expiration_str: + dt = datetime.fromisoformat( + drm_expiration_str.replace("Z", "+00:00") + ) + expiry = dt.timestamp() - 60 + except Exception as e: + logger.debug(f"Could not parse drm_expiration: {e}") + + if expiry is None or expiry <= now: + expiry = now + 23 * 3600 + logger.debug( + f"Using fallback 23h TTL for playback cache " + f"(edit_id: {edit_id})" + ) + + self._playback_cache[edit_id] = (expiry, data) + logger.debug( + f"Playback cache stored for edit_id {edit_id} " + f"(expires in {int(expiry - now)}s)" + ) + return data + + # ========================================================================= + # Payload / streaming data helpers + # ========================================================================= + + def _build_playback_payload(self, edit_id: str) -> Dict[str, Any]: + """ + Build the playback request payload for the Discovery+ playback API. + + Args: + edit_id: Content edit ID. + + Returns: + JSON-serialisable payload dict. + """ + nonce = secrets.token_bytes(256) + google_pal_nonce = base64.b64encode(nonce).decode("ascii") + + return { + "appBundle": "dplus", + "advertisingInfo": { + "adBlockerDetection": False, + "debug": {}, + "device": {}, + "googlePALNonce": google_pal_nonce, + "server": { + "deviceId": "", + "iabTCFString": "", + "isLimitedAdTracking": 0, + "nielsenAppId": "", + }, + "ssaiProvider": {"version": "2.2.0"}, + }, + "consumptionType": "streaming", + "deviceInfo": get_default_device_info( + self._provider.platform_os, + device_id=self._provider.authenticator.device_id, + ), + "editId": edit_id, + "capabilities": get_default_capabilities( + self._provider.platform_os + ), + "gdpr": False, + "firstPlay": False, + "playbackSessionId": str(uuid.uuid4()), + "applicationSessionId": str(uuid.uuid4()), + "userPreferences": { + "videoQuality": "best", + "uiLanguage": f"{self._provider.country}-DE".upper(), + }, + "features": ["mlp"], + } + + @staticmethod + def extract_streaming_data(playback_data: Dict) -> Dict[str, Any]: + """ + Extract streaming URLs and DRM info from a playback response. + + Detects whichever DRM scheme is present — either ``widevine`` + (Linux/Chrome) or ``playready`` (Windows/Edge) — so callers can build + the correct DRMConfig without knowing which platform produced the + response. + + Args: + playback_data: Playback API response dict. + + Returns: + Dictionary with keys: + manifest_url – DASH manifest URL + license_url – DRM license server URL (or None) + drm_system – 'widevine' | 'playready' | None + drm_auth – JWT auth token from license URL (or None) + streaming_format – 'dash' (always for Discovery+) + drm_expiration – ISO-8601 expiration string (or None) + fallback_manifest – Fallback manifest URL (or None) + """ + result: Dict[str, Any] = { + "manifest_url": None, + "license_url": None, + "drm_system": None, + "drm_auth": None, + "streaming_format": "dash", + "drm_expiration": None, + "fallback_manifest": None, + } + + try: + manifest = playback_data.get("manifest", {}) + if manifest: + result["manifest_url"] = manifest.get("url") + result["streaming_format"] = manifest.get("format", "dash") + + fallback = playback_data.get("fallback", {}) + if fallback: + fallback_manifest = fallback.get("manifest", {}) + if fallback_manifest: + result["fallback_manifest"] = fallback_manifest.get("url") + + drm = playback_data.get("drm", {}) + if drm: + result["drm_expiration"] = drm.get("expirationDate") + schemes = drm.get("schemes", {}) + + for scheme_name in ("widevine", "playready"): + scheme = schemes.get(scheme_name, {}) + if scheme and scheme.get("licenseUrl"): + result["license_url"] = scheme["licenseUrl"] + result["drm_system"] = scheme_name + + if "auth=" in result["license_url"]: + import urllib.parse + parsed = urllib.parse.urlparse( + result["license_url"] + ) + query = urllib.parse.parse_qs(parsed.query) + if "auth" in query: + result["drm_auth"] = query["auth"][0] + break + + cdn = playback_data.get("cdn", {}) + if cdn: + result["cdn_provider"] = cdn.get("provider") + + except Exception as e: + logger.error(f"Error extracting streaming data: {e}") + + return result + + def build_drm_config( + self, streaming_data: Dict[str, Any] + ) -> Optional[DRMConfig]: + """ + Build a DRMConfig from extracted streaming data. + + Selects the correct DRMSystem based on the scheme returned by the + server (``streaming_data["drm_system"]``). Falls back to the active + ``platform_os`` when the scheme is absent. + + Supported schemes: + 'widevine' → DRMSystem.WIDEVINE (Linux/Chrome path) + 'playready' → DRMSystem.PLAYREADY (Windows/Edge path) + + Args: + streaming_data: Dict returned by ``extract_streaming_data()``. + + Returns: + DRMConfig instance, or None if no license URL is available. + """ + license_url = streaming_data.get("license_url") + if not license_url: + return None + + drm_system_str = streaming_data.get("drm_system") + if drm_system_str == "playready": + drm_system = DRMSystem.PLAYREADY + elif drm_system_str == "widevine": + drm_system = DRMSystem.WIDEVINE + else: + drm_system = ( + DRMSystem.PLAYREADY + if self._provider.platform_os == PlatformOS.WINDOWS + else DRMSystem.WIDEVINE + ) + logger.debug( + f"drm_system not in streaming_data, inferred from platform_os " + f"({self._provider.platform_os.value}): {drm_system.name}" + ) + + license_headers = get_drm_request_headers(self._provider.platform_os) + + return DRMConfig( + system=drm_system, + priority=1, + license=LicenseConfig( + server_url=license_url, + req_headers=json.dumps(license_headers), + req_data="{CHA-RAW}", + use_http_get_request=False, + ), + ) \ No newline at end of file diff --git a/lib/streaming_providers/providers/discovery/provider.py b/lib/streaming_providers/providers/discovery/provider.py index a7c2720..fc6b722 100644 --- a/lib/streaming_providers/providers/discovery/provider.py +++ b/lib/streaming_providers/providers/discovery/provider.py @@ -2,19 +2,19 @@ """ Discovery+ Streaming Provider -Provides access to Discovery+ live channels and VOD content with proper -authentication, DRM support, and channel management. +Thin orchestrator that wires together the three domain managers: + - DiscoveryChannelManager (channel extraction + CMS route discovery) + - DiscoveryEventManager (sport schedule / live event fetching) + - DiscoveryPlaybackManager (playback tokens, manifest extraction, DRM) + +All heavy logic lives in those managers. This class owns the shared caches, +handles authentication, and exposes the public StreamingProvider interface. """ -import json -import time -import uuid -import secrets -import base64 -from datetime import datetime, timezone -from typing import ClassVar, Dict, List, Optional, Any +from datetime import datetime +from typing import ClassVar, Dict, List, Optional from ...base.models.proxy_models import ProxyConfig -from ...base.models import DRMConfig, DRMSystem, LicenseConfig, StreamingChannel, Event, EventStatus +from ...base.models import DRMConfig, StreamingChannel, Event from ...base.provider import AuthType, StreamingProvider from ...base.utils.logger import logger @@ -23,6 +23,7 @@ from .auth import ( DiscoveryAnonymousCredentials, DiscoveryUserCredentials, ) +from .channel_manager import DiscoveryChannelManager from .constants import ( DEFAULT_COUNTRY, DEFAULT_PLATFORM_OS, @@ -31,25 +32,12 @@ from .constants import ( PlatformOS, SUPPORTED_AUTH_TYPES, SUPPORTED_COUNTRIES, - CMS_INCLUDE_PARAMS, - CMS_PAGE_SIZE, - CMS_ROUTE_SPORTS, - CMS_ROUTE_SPORT_SCHEDULE, - AIRING_BADGE_LIVE, - AIRING_BADGE_UP_NEXT, - AIRING_BADGE_UPCOMING_LEGACY, - AIRING_ITEM_TYPE, - get_default_capabilities, - get_default_device_info, - get_drm_request_headers, get_user_agent, ) -from .exceptions import ( - PlaybackRestrictedException, - ChannelNotFoundError, - ManifestFetchError, -) +from .event_manager import DiscoveryEventManager +from .exceptions import ChannelNotFoundError, ManifestFetchError from .models import DiscoveryChannel +from .playback_manager import DiscoveryPlaybackManager class DiscoveryProvider(StreamingProvider): @@ -65,27 +53,15 @@ class DiscoveryProvider(StreamingProvider): - Provider cache pattern for efficient channel management - Dynamic CMS route discovery from ``/home`` navigation graph - Event fetching uses **dynamic route selection** (``_select_event_route``): - ``get_channels()`` parses the ``/home`` navigation graph and caches all - available routes in ``self._cms_routes``. ``get_events()`` then picks the - first route whose ID contains ``sport``, ``live``, or ``event`` — no - hardcoded route names needed. Two parse paths are supported: - - 1. /sports (CMS_ROUTE_SPORTS, default) - The API returns ``airing`` items directly in the top-level ``included`` - array. No nested collection traversal is needed; ``get_events()`` - detects this shape and calls ``_parse_airing_events()`` directly. - - 2. /sport-schedule (CMS_ROUTE_SPORT_SCHEDULE, legacy fallback) - Items are nested inside a collection and require the full multi-hop - traversal: route → target page → page items → collection → paginated - items → video. Used automatically when the /sports route returns no - airing items (e.g. future API layout changes). + Delegates to three managers: + channel_manager — channel extraction, CMS route discovery + event_manager — sport schedule / live event fetching + playback_manager — playback tokens, manifest & DRM extraction """ - # ============================================================================ + # ========================================================================= # STATIC METADATA - # ============================================================================ + # ========================================================================= PROVIDER_LABEL: ClassVar[str] = "Discovery+" SUPPORTED_AUTH_TYPES: ClassVar[List[str]] = SUPPORTED_AUTH_TYPES PROVIDER_LOGO: ClassVar[str] = DISCOVERY_LOGO @@ -117,7 +93,7 @@ class DiscoveryProvider(StreamingProvider): proxy_url: Optional proxy URL string platform_os: OS platform to spoof (PlatformOS.LINUX or PlatformOS.WINDOWS). Defaults to DEFAULT_PLATFORM_OS from constants. - LINUX → Widevine + Chrome on Linux (original behaviour) + LINUX → Widevine + Chrome on Linux WINDOWS → PlayReady + Edge on Windows Raises: @@ -131,27 +107,25 @@ class DiscoveryProvider(StreamingProvider): super().__init__(country=country) - # OS platform used for all device/header/DRM spoofing - self.platform_os: PlatformOS = platform_os if platform_os is not None else DEFAULT_PLATFORM_OS - + self.platform_os: PlatformOS = ( + platform_os if platform_os is not None else DEFAULT_PLATFORM_OS + ) self.auth_type = auth_type self._settings_manager = settings_manager - # Provider cache for DiscoveryChannel objects + # ------------------------------------------------------------------ + # Shared caches (owned here; passed by reference into each manager) + # ------------------------------------------------------------------ + # DiscoveryChannel objects keyed by channel_id self._channels_cache: Dict[str, DiscoveryChannel] = {} - - # CMS navigation routes discovered from /home response. - # Populated by get_channels() and used by get_events() to select the - # best event-producing route without hardcoding names. - # Format: { route_id: label } e.g. {"sports": "Sport", "home": "Home"} + # Navigation routes discovered from /home: { route_id: label } self._cms_routes: Dict[str, str] = {} - - # Playback info cache: {edit_id: (expiry_timestamp, playback_data)} - # Entries are valid until expiry_timestamp (derived from drm_expiration, - # or 23 hours from fetch time as a safe fallback for the 24h token window). + # Playback info cache: { edit_id: (expiry_timestamp, playback_data) } self._playback_cache: Dict[str, tuple] = {} - # Setup HTTP manager using abstraction + # ------------------------------------------------------------------ + # HTTP manager + # ------------------------------------------------------------------ self.http_manager = self._setup_http_manager( provider_name="discovery", proxy_config=proxy_config, @@ -162,12 +136,13 @@ class DiscoveryProvider(StreamingProvider): max_retries=3, ) - # Create appropriate credentials. + # ------------------------------------------------------------------ + # Credentials # Priority: # 1. Explicit username/password passed at construction time - # 2. Stored user credentials from CredentialManager (same /config path - # the authenticator uses — does not depend on injected settings_manager) + # 2. Stored user credentials from CredentialManager # 3. Anonymous fallback + # ------------------------------------------------------------------ if auth_type == "user_credentials": if not username or not password: raise ValueError( @@ -188,7 +163,9 @@ class DiscoveryProvider(StreamingProvider): f"{type(stored).__name__ if stored else 'None'}" ) except Exception as e: - logger.debug(f"Could not load stored credentials via CredentialManager: {e}") + logger.debug( + f"Could not load stored credentials via CredentialManager: {e}" + ) if stored and stored.validate() and hasattr(stored, "username"): logger.info( @@ -207,14 +184,9 @@ class DiscoveryProvider(StreamingProvider): ) self.credentials = DiscoveryAnonymousCredentials() - # Create authenticator with dynamic endpoint discovery. - # settings_manager is intentionally not passed so the authenticator - # self-constructs its own SettingsManager via the backward-compat path, - # identical to how JoynAuthenticator works. This ensures it always - # resolves the correct /config path via the environment variable. - # Resolve stable device ID from settings manager before constructing - # the authenticator — the authenticator intentionally receives - # settings_manager=None so it cannot do this lookup itself. + # ------------------------------------------------------------------ + # Authenticator + # ------------------------------------------------------------------ _device_id = None if self._settings_manager: try: @@ -223,7 +195,9 @@ class DiscoveryProvider(StreamingProvider): ) logger.debug(f"Resolved device_id from settings: {_device_id}") except Exception as e: - logger.debug(f"Could not resolve device_id from settings: {e}") + logger.debug( + f"Could not resolve device_id from settings: {e}" + ) self.authenticator = DiscoveryAuthenticator( country=country, @@ -236,7 +210,6 @@ class DiscoveryProvider(StreamingProvider): device_id=_device_id, ) - # Authenticate try: self.bearer_token = self.authenticator.get_bearer_token() self.token_info = self.authenticator.get_token_info() @@ -248,19 +221,39 @@ class DiscoveryProvider(StreamingProvider): self.bearer_token = None self.token_info = None - # Discover CMS navigation routes from /home. - # Done eagerly so _cms_routes is populated before any call to - # get_events() — regardless of whether get_channels() is called first. - self._init_cms_routes() + # ------------------------------------------------------------------ + # Domain managers — all share the same cache references + # ------------------------------------------------------------------ + self.channel_manager = DiscoveryChannelManager( + provider=self, + channels_cache=self._channels_cache, + cms_routes=self._cms_routes, + ) + self.event_manager = DiscoveryEventManager( + provider=self, + channels_cache=self._channels_cache, + cms_routes=self._cms_routes, + ) + self.playback_manager = DiscoveryPlaybackManager( + provider=self, + channels_cache=self._channels_cache, + playback_cache=self._playback_cache, + ) + + # Eagerly discover CMS routes so get_events() works even if + # get_channels() has not been called first. + self.channel_manager.init_cms_routes() + + # ========================================================================= + # Provider identity + # ========================================================================= @property def provider_name(self) -> str: - """Provider identifier""" return "discovery" @property def provider_label(self) -> str: - """Return country-specific label""" country_map = { "de": "Discovery+ Germany", "at": "Discovery+ Austria", @@ -282,26 +275,25 @@ class DiscoveryProvider(StreamingProvider): @property def provider_logo(self) -> str: - """Provider logo URL""" return self.PROVIDER_LOGO @property def supported_auth_types(self) -> List[str]: - """Supported authentication types""" return self.SUPPORTED_AUTH_TYPES @property def uses_dynamic_manifests(self) -> bool: - """Discovery+ uses session-based manifests""" return True @property def implements_epg(self) -> bool: - """Discovery+ does not implement native EPG""" return False - def _get_auth_headers(self) -> Dict[str, str]: - """Get headers with authentication""" + # ========================================================================= + # Authentication helpers + # ========================================================================= + + def get_auth_headers(self) -> Dict[str, str]: return self._build_provider_headers( base_headers={ "User-Agent": get_user_agent(self.platform_os), @@ -312,7 +304,6 @@ class DiscoveryProvider(StreamingProvider): ) def authenticate(self, **kwargs) -> str: - """Authenticate and return bearer token""" self.bearer_token = self.authenticator.get_bearer_token( force_refresh=kwargs.get("force_refresh", False) ) @@ -320,58 +311,36 @@ class DiscoveryProvider(StreamingProvider): return self.bearer_token def refresh_authentication(self) -> str: - """Force refresh authentication""" return self.authenticate(force_refresh=True) def is_anonymous(self) -> bool: - """Check if using anonymous authentication""" if self.token_info: return self.token_info.get("anonymous", True) return True + # ========================================================================= + # Country helpers + # ========================================================================= + @classmethod def supports_multiple_countries(cls) -> bool: - """ - Discovery+ uses a wildcard country list, but still registers one - instance per country — so the registry must treat it as multi-country - to produce names like ``discovery_de``. - """ return cls.SUPPORTED_COUNTRIES == ["*"] @classmethod def validate_country(cls, country: str) -> bool: - """ - Override base validation. - - With the wildcard sentinel any valid ISO-3166-1 alpha-2 code is - accepted. The user's actual serving country is determined at runtime - via get_user_country(). - """ if cls.SUPPORTED_COUNTRIES == ["*"]: return bool(country and country.isalpha() and len(country) == 2) - # Fallback to standard list-based check if constant is ever changed back return country.lower() in [c.lower() for c in cls.SUPPORTED_COUNTRIES] def get_user_country(self) -> Optional[str]: """ - Detect the user's current country by calling the /users/me endpoint. - - The API returns ``currentLocationTerritory`` (e.g. ``"DE"``) inside - ``data.attributes``. This method normalises the value to lowercase - so it is consistent with the rest of the codebase (e.g. ``"de"``). + Detect the user's current country via the /users/me endpoint. Returns: - Two-letter lowercase country code (e.g. ``"de"``, ``"uk"``) or - ``None`` if the request fails or the field is absent. - - Example:: - - provider = DiscoveryProvider(country="de") - country = provider.get_user_country() - # → "de" (or whatever territory the server reports) + Two-letter lowercase country code (e.g. ``"de"``) or None. """ try: - headers = self._get_auth_headers() + headers = self.get_auth_headers() response = self.http_manager.get( DISCOVERY_USERS_ME_URL, operation="users_me", @@ -390,86 +359,29 @@ class DiscoveryProvider(StreamingProvider): f"Detected user country from /users/me: {country_code}" ) return country_code - logger.warning("/users/me response missing currentLocationTerritory") + logger.warning( + "/users/me response missing currentLocationTerritory" + ) return None except Exception as e: logger.error(f"Failed to fetch user country from /users/me: {e}") return None + # ========================================================================= + # Public StreamingProvider interface — delegates to managers + # ========================================================================= + def get_channels( self, fetch_manifests: bool = False, populate_streaming_data: bool = True, **kwargs, ) -> List[StreamingChannel]: - """ - Fetch available channels from Discovery+ CMS. - - Args: - fetch_manifests: Whether to immediately populate streaming data - populate_streaming_data: Whether to populate streaming data - **kwargs: Additional parameters - - Returns: - List of StreamingChannel objects - """ - try: - headers = self._get_auth_headers() - - # Fetch home route with all needed includes - url = self.authenticator.cms_home_endpoint - params = { - "include": CMS_INCLUDE_PARAMS, - "decorators": "viewingHistory,isFavorite,contentAction,badges", - "page[items.size]": CMS_PAGE_SIZE, - } - - response = self.http_manager.get( - url, - operation="cms", - headers=headers, - params=params, - ) - response.raise_for_status() - data = response.json() - - # Refresh CMS route cache from this /home response. - # _cms_routes is already populated at init, but get_channels() may - # be called later in a session when routes could have changed. - refreshed = self._discover_cms_routes(data) - if refreshed: - self._cms_routes = refreshed - logger.debug(f"CMS routes refreshed: {list(self._cms_routes.keys())}") - - # Extract channels and cache DiscoveryChannel objects - discovery_channels = self._extract_distribution_channels(data) - - # Cache the original DiscoveryChannel objects - self._channels_cache = { - ch.channel_id: ch for ch in discovery_channels - } - - # Convert to StreamingChannel objects - streaming_channels = [ - ch.to_streaming_channel(self.provider_name) - for ch in discovery_channels - ] - - logger.info( - f"Found {len(streaming_channels)} distribution channels" - ) - - # Populate streaming data if requested - if fetch_manifests and populate_streaming_data and streaming_channels: - streaming_channels = self.populate_streaming_data( - streaming_channels - ) - - return streaming_channels - - except Exception as e: - logger.error(f"Error fetching channels: {e}") - return [] + """Fetch available channels from the Discovery+ CMS.""" + return self.channel_manager.fetch_channels( + fetch_manifests=fetch_manifests, + populate_streaming_data=populate_streaming_data, + ) def get_events( self, @@ -477,1117 +389,11 @@ class DiscoveryProvider(StreamingProvider): end_time: Optional[datetime] = None, **kwargs, ) -> List[Event]: - """ - Fetch events from Discovery+ schedule. - - **Route selection** (no ``route_id`` kwarg supplied): - Calls ``_select_event_route()`` which prefers a route discovered - dynamically from the ``/home`` navigation graph (populated by - ``get_channels()``). The first route whose ID contains one of - ``sport``, ``live``, or ``event`` wins. If no match is found — - e.g. ``get_channels()`` hasn't been called yet — falls back to - ``CMS_ROUTE_SPORT_SCHEDULE`` (the full paginated schedule). - - **Parse paths** (determined by the selected route's response shape): - 1. ``/sports`` — ``airing`` items appear directly in ``included`` - (fast path, ``_parse_airing_events()``). - 2. ``/sport-schedule`` — items are nested inside a collection - (legacy fallback, ``_fetch_events_via_collection()``). - The fast path is attempted first; if no ``airing`` items are found - the method retries with ``CMS_ROUTE_SPORT_SCHEDULE``. - - Args: - start_time: Optional start time filter - end_time: Optional end time filter - **kwargs: Additional parameters including: - - route_id: Override the CMS route (bypasses dynamic selection) - - page_size: Number of items per page (collection path only) - - Returns: - List of Event objects - """ - events = [] - - try: - headers = self._get_auth_headers() - - # Dynamic route selection: prefer a discovered route that looks - # event-bearing. Falls back to CMS_ROUTE_SPORTS if get_channels() - # hasn't been called yet (self._cms_routes is empty). - if "route_id" not in kwargs: - route_id = self._select_event_route() - else: - route_id = kwargs["route_id"] - - # Build the CMS route URL - base_url = self.authenticator.cms_home_endpoint - if base_url.endswith('/home'): - url = base_url.replace('/home', f'/{route_id}') - else: - url = ( - f"https://default.any-{self.authenticator.home_market}" - f".{self.authenticator.env}.api.discoveryplus.com" - f"/cms/routes/{route_id}" - ) - - params = { - "include": CMS_INCLUDE_PARAMS, - "decorators": "viewingHistory,isFavorite,contentAction,badges", - "page[items.size]": kwargs.get("page_size", 10), - } - - # Add time filters if provided - if start_time: - params["filter[from]"] = start_time.isoformat() - if end_time: - params["filter[to]"] = end_time.isoformat() - - logger.debug(f"Fetching events route: {url}") - response = self.http_manager.get( - url, - operation="cms", - headers=headers, - params=params, - ) - response.raise_for_status() - data = response.json() - - # ------------------------------------------------------------------ - # Fast path: /sports route returns airing items directly in included - # ------------------------------------------------------------------ - airing_items = [ - item for item in data.get("included", []) - if item.get("type") == AIRING_ITEM_TYPE - ] - - if airing_items: - logger.debug( - f"Found {len(airing_items)} airing items in /sports route — " - "using direct airing parse path" - ) - events = self._parse_airing_events(airing_items, start_time, end_time) - logger.info(f"Found {len(events)} events (airing path)") - return events - - # ------------------------------------------------------------------ - # Slow path: collection-based traversal (legacy /sport-schedule shape) - # ------------------------------------------------------------------ - # If the caller explicitly asked for the sports route but got no - # airings, retry once with the legacy route before giving up. - if route_id == CMS_ROUTE_SPORTS: - logger.debug( - "No airing items on /sports route — falling back to " - f"/{CMS_ROUTE_SPORT_SCHEDULE} collection path" - ) - return self.get_events( - start_time=start_time, - end_time=end_time, - route_id=CMS_ROUTE_SPORT_SCHEDULE, - **{k: v for k, v in kwargs.items() if k != "route_id"}, - ) - - # ---- legacy collection traversal (sport-schedule) ---------------- - events = self._fetch_events_via_collection( - data=data, - headers=headers, - start_time=start_time, - end_time=end_time, - **kwargs, - ) - logger.info(f"Found {len(events)} events (collection path)") - - except Exception as e: - logger.error(f"Error fetching events: {e}") - - return events - - # ========================================================================= - # CMS route discovery - # ========================================================================= - - def _init_cms_routes(self) -> None: - """ - Fetch ``/home`` and populate ``self._cms_routes`` eagerly. - - Called once at the end of ``__init__`` so route discovery is complete - before any call to ``get_events()`` or ``get_channels()``. Failures - are caught and logged — the provider remains usable and ``get_events()`` - will fall back to ``CMS_ROUTE_SPORT_SCHEDULE``. - """ - try: - headers = self._get_auth_headers() - url = self.authenticator.cms_home_endpoint - params = { - "include": CMS_INCLUDE_PARAMS, - "decorators": "viewingHistory,isFavorite,contentAction,badges", - "page[items.size]": CMS_PAGE_SIZE, - } - response = self.http_manager.get( - url, - operation="cms", - headers=headers, - params=params, - ) - response.raise_for_status() - data = response.json() - self._cms_routes = self._discover_cms_routes(data) - if self._cms_routes: - logger.debug( - f"CMS routes discovered at init: {list(self._cms_routes.keys())}" - ) - else: - logger.warning( - "CMS route discovery returned no routes — " - f"get_events() will fall back to '{CMS_ROUTE_SPORT_SCHEDULE}'" - ) - except Exception as e: - logger.warning(f"Could not discover CMS routes at init: {e}") - - @staticmethod - def _discover_cms_routes(data: dict) -> Dict[str, str]: - """ - Parse a /home CMS response and return a map of all navigation routes. - - The /home response contains a ``data.relationships.navigationLinks`` - array whose entries are resolved via the top-level ``included`` list. - Each resolved node has ``attributes.url`` (e.g. ``"/sports"``) and - ``attributes.label`` (localised display name). - - Returns: - dict mapping route_id → label, e.g. - ``{"sports": "Sport", "channels": "Kanäle", "home": "Home"}`` - Returns an empty dict if the navigation graph is absent. - """ - included_by_id: Dict[str, dict] = { - f"{item['type']}:{item['id']}": item - for item in data.get("included", []) - if item.get("id") and item.get("type") - } - nav_refs = ( - data.get("data", {}) - .get("relationships", {}) - .get("navigationLinks", {}) - .get("data", []) - ) - if not nav_refs: - logger.debug("_discover_cms_routes: no navigationLinks found in /home response") - return {} - - routes: Dict[str, str] = {} - for ref in nav_refs: - key = f"{ref.get('type', 'link')}:{ref['id']}" - node = included_by_id.get(key, {}) - attrs = node.get("attributes", {}) - # Use the URL path segment as the route ID; fall back to the ref id - url_path = attrs.get("url", "").lstrip("/") - route_id = url_path or ref["id"] - label = attrs.get("label", route_id) - routes[route_id] = label - - return routes - - # Route preference order for event fetching. - # ``sport-schedule`` (paginated full schedule, ~300 events) is preferred - # over ``sports`` (featured/current airings only, ~14 events). - # Order matters: the first matching route wins. - _EVENT_ROUTE_PREFERENCE: ClassVar[tuple] = ( - CMS_ROUTE_SPORT_SCHEDULE, # "sport-schedule" — full paginated schedule - CMS_ROUTE_SPORTS, # "sports" — featured airings only (fallback) - ) - # Generic keyword scan used when neither preferred route is discovered - _EVENT_ROUTE_KEYWORDS: ClassVar[tuple] = ("schedule", "sport", "live", "event") - - def _select_event_route(self) -> str: - """ - Return the best event-producing CMS route from ``self._cms_routes``. - - Preference order: - 1. ``CMS_ROUTE_SPORT_SCHEDULE`` (``sport-schedule``) — full paginated - schedule, yields ~300 events via collection traversal. - 2. ``CMS_ROUTE_SPORTS`` (``sports``) — featured/current airings only, - ~14 events. Used only when sport-schedule is absent. - 3. First discovered route matching a keyword in - ``_EVENT_ROUTE_KEYWORDS`` (generic fallback for unknown layouts). - 4. ``CMS_ROUTE_SPORT_SCHEDULE`` hard-coded constant when ``_cms_routes`` - is empty (e.g. ``get_channels()`` hasn't been called yet). - """ - # 1 & 2: explicit preference list checked against discovered routes - for preferred in self._EVENT_ROUTE_PREFERENCE: - if preferred in self._cms_routes: - logger.debug(f"_select_event_route: selected preferred route '{preferred}'") - return preferred - - # 3: generic keyword scan for unexpected route layouts - for keyword in self._EVENT_ROUTE_KEYWORDS: - match = next( - (r for r in self._cms_routes if keyword in r.lower()), - None, - ) - if match: - logger.debug( - f"_select_event_route: selected '{match}' via keyword '{keyword}'" - ) - return match - - # 4: hard fallback — use the full schedule route, not the featured-only /sports route - logger.debug( - f"_select_event_route: no event route found in {list(self._cms_routes.keys())!r}, " - f"falling back to '{CMS_ROUTE_SPORT_SCHEDULE}'" - ) - return CMS_ROUTE_SPORT_SCHEDULE - - # ========================================================================= - # Airing-based event parsing (/sports route) - # ========================================================================= - - def _parse_airing_events( - self, - airing_items: List[Dict], - start_time: Optional[datetime] = None, - end_time: Optional[datetime] = None, - ) -> List[Event]: - """ - Parse a list of ``airing`` objects from the /sports CMS route into - ``Event`` instances. - - The /sports route places ``airing`` items directly in the top-level - ``included`` array. Each airing carries: - - - ``attributes.scheduleStart / scheduleEnd`` — ISO-8601 UTC timestamps - - ``attributes.name`` — episode/event title - - ``attributes.showName`` — parent show name - - ``attributes.description / secondaryTitle`` — optional metadata - - ``attributes.episodeNumber / seasonNumber`` — episode metadata - - ``relationships.badges`` — live / up-next status - - ``relationships.distributionChannel`` — links to channel cache - - Badge → EventStatus mapping: - AIRING_BADGE_LIVE ("live") → EventStatus.LIVE - AIRING_BADGE_UP_NEXT ("release-state-up-next") → EventStatus.UPCOMING - (fallback clock-based logic handles ENDED / LIVE for unlabelled items) - - The ``distributionChannel`` relationship ID is looked up in - ``self._channels_cache`` so the event carries the correct channel name - without an extra API round-trip. - - Args: - airing_items: List of airing dicts from ``data["included"]``. - start_time: Optional lower bound filter (inclusive). - end_time: Optional upper bound filter (inclusive). - - Returns: - List of Event objects, optionally filtered by time window. - """ - events: List[Event] = [] - now_utc = datetime.now(timezone.utc) - - for airing in airing_items: - try: - attributes = airing.get("attributes", {}) - relationships = airing.get("relationships", {}) - - # ---- schedule times ----------------------------------------- - schedule_start = attributes.get("scheduleStart") - schedule_end = attributes.get("scheduleEnd") - - start_dt: Optional[datetime] = None - end_dt: Optional[datetime] = None - - if schedule_start: - try: - start_dt = datetime.fromisoformat( - schedule_start.replace("Z", "+00:00") - ) - except (ValueError, TypeError): - pass - - if schedule_end: - try: - end_dt = datetime.fromisoformat( - schedule_end.replace("Z", "+00:00") - ) - except (ValueError, TypeError): - pass - - # ---- optional caller-supplied time window filter ------------- - if start_time and end_dt and end_dt < start_time: - continue - if end_time and start_dt and start_dt > end_time: - continue - - # ---- event status from badges -------------------------------- - # Badge priority: explicit "live" wins over "up-next"; clock- - # based logic is applied when neither badge is present. - status = EventStatus.SCHEDULED - badge_refs = relationships.get("badges", {}).get("data", []) - badge_ids = {b.get("id") for b in badge_refs} - - if AIRING_BADGE_LIVE in badge_ids: - status = EventStatus.LIVE - elif AIRING_BADGE_UP_NEXT in badge_ids or AIRING_BADGE_UPCOMING_LEGACY in badge_ids: - status = EventStatus.SCHEDULED - elif start_dt and end_dt: - # Ensure tz-aware before comparison - _start = start_dt if start_dt.tzinfo else start_dt.replace(tzinfo=timezone.utc) - _end = end_dt if end_dt.tzinfo else end_dt.replace(tzinfo=timezone.utc) - - if now_utc > _end: - status = EventStatus.ENDED - elif _start <= now_utc <= _end: - status = EventStatus.LIVE - - # ---- channel lookup ----------------------------------------- - # distributionChannel.data.id matches the keys in _channels_cache - channel_name: Optional[str] = None - dist_channel_ref = ( - relationships.get("distributionChannel", {}).get("data", {}) - ) - dist_channel_id = dist_channel_ref.get("id") if dist_channel_ref else None - - if dist_channel_id: - cached_channel = self._channels_cache.get(dist_channel_id) - if cached_channel: - channel_name = cached_channel.name - else: - logger.debug( - f"distributionChannel {dist_channel_id} not in cache " - f"for airing '{attributes.get('name')}'" - ) - - # ---- build Event -------------------------------------------- - event = Event( - name=attributes.get("name", "Unknown Event"), - content_id=airing.get("id", ""), - provider=self.provider_name, - logo_url=None, # airing items carry image IDs only; - # full URLs require a separate fetch - mode="live" if status == EventStatus.LIVE else "vod", - session_manifest=False, - manifest_script=None, - cdm=None, - content_type="AIRING", - description=attributes.get("description", ""), - genre=None, - language="de", # airings don't expose audioTracks - country=self.country.upper(), - start_time=start_dt, - end_time=end_dt, - status=status, - subtitle=attributes.get("secondaryTitle"), - original_name=attributes.get("showName"), - competition=None, - venue=None, - gender=None, - discipline=None, - age_category=None, - master_event=None, - channel=channel_name, - ) - - events.append(event) - - except Exception as e: - logger.error( - f"Error processing airing item {airing.get('id')}: {e}" - ) - continue - - return events - - # ========================================================================= - # Legacy collection-based event fetching (/sport-schedule route) - # ========================================================================= - - def _fetch_events_via_collection( - self, - data: Dict, - headers: Dict, - start_time: Optional[datetime] = None, - end_time: Optional[datetime] = None, + """Fetch sport schedule events from Discovery+.""" + return self.event_manager.fetch_events( + start_time=start_time, + end_time=end_time, **kwargs, - ) -> List[Event]: - """ - Traverse the legacy collection-based CMS shape to extract events. - - This implements the original multi-hop traversal: - route data → target page → page items → collection → - paginated collection items → video attributes - - Args: - data: Parsed JSON from the initial CMS route request. - headers: Auth headers for subsequent requests. - start_time: Optional lower bound filter. - end_time: Optional upper bound filter. - **kwargs: Forwarded kwargs (page_size, etc.). - - Returns: - List of Event objects. - """ - events: List[Event] = [] - - # Build lookup of included items by type:id - included_by_id: Dict[str, Dict] = {} - for item in data.get("included", []): - item_id = item.get("id") - item_type = item.get("type") - if item_id and item_type: - included_by_id[f"{item_type}:{item_id}"] = item - - # Find the target page from the route - route_data = data.get("data", {}) - target_ref = route_data.get("relationships", {}).get("target", {}).get("data") - - if not target_ref: - logger.error("No target page found in route response (collection path)") - return events - - page_key = f"{target_ref.get('type')}:{target_ref.get('id')}" - page = included_by_id.get(page_key) - - if not page: - logger.error("Target page not found in included items (collection path)") - return events - - # Find the collection from page items - page_items = page.get("relationships", {}).get("items", {}).get("data", []) - collection_id = None - - for page_item_ref in page_items: - page_item_key = f"{page_item_ref.get('type')}:{page_item_ref.get('id')}" - page_item = included_by_id.get(page_item_key) - if page_item: - collection_ref = ( - page_item.get("relationships", {}) - .get("collection", {}) - .get("data") - ) - if collection_ref: - collection_id = collection_ref.get("id") - break - - if not collection_id: - logger.error("No collection found in page (collection path)") - return events - - # Paginate through the collection - collection_url = ( - f"{self.authenticator.cms_collections_endpoint}/{collection_id}" - ) - current_page = 1 - - while True: - collection_params = { - "include": "default", - "decorators": "viewingHistory,isFavorite,contentAction,badges", - "page[items.size]": kwargs.get("page_size", 30), - "page[items.number]": current_page, - } - - logger.debug( - f"Fetching schedule collection page {current_page}: {collection_url}" - ) - collection_response = self.http_manager.get( - collection_url, - operation="cms", - headers=headers, - params=collection_params, - ) - collection_response.raise_for_status() - collection_data = collection_response.json() - - # Build per-page included lookup - page_included: Dict[str, Dict] = {} - for item in collection_data.get("included", []): - item_id = item.get("id") - item_type = item.get("type") - if item_id and item_type: - page_included[f"{item_type}:{item_id}"] = item - - items = ( - collection_data.get("data", {}) - .get("relationships", {}) - .get("items", {}) - .get("data", []) - ) - - meta = collection_data.get("meta", {}) - total_pages = meta.get("itemsTotalPages", 1) - logger.debug( - f"Fetched collection page {current_page}/{total_pages} " - f"({len(items)} items)" - ) - - for item_ref in items: - try: - item_key = f"{item_ref.get('type')}:{item_ref.get('id')}" - collection_item = page_included.get(item_key) - - if not collection_item: - continue - - # Get video reference from collection item - video_ref = ( - collection_item.get("relationships", {}) - .get("video", {}) - .get("data") - ) - if not video_ref: - continue - - video_key = f"{video_ref.get('type')}:{video_ref.get('id')}" - video_data = page_included.get(video_key) - - if not video_data: - continue - - # Only process EVENT materialType - attributes = video_data.get("attributes", {}) - relationships = video_data.get("relationships", {}) - - material_type = attributes.get("materialType") - if material_type != "EVENT": - continue - - # ---- status from badges ---------------------------------- - status = EventStatus.SCHEDULED - badge_refs = relationships.get("badges", {}).get("data", []) - badge_ids = {b.get("id") for b in badge_refs} - - if AIRING_BADGE_LIVE in badge_ids: - status = EventStatus.LIVE - elif ( - AIRING_BADGE_UP_NEXT in badge_ids - or AIRING_BADGE_UPCOMING_LEGACY in badge_ids - ): - status = EventStatus.SCHEDULED - - # Also check overlays for live status - for badge_ref in badge_refs: - overlay_key = f"{badge_ref.get('type')}:{badge_ref.get('id')}" - overlay = page_included.get(overlay_key, {}) - if overlay.get("id") == "live:default": - status = EventStatus.LIVE - break - - # ---- schedule times ------------------------------------- - schedule_start = attributes.get("scheduleStart") - schedule_end = attributes.get("scheduleEnd") - - start_dt: Optional[datetime] = None - end_dt: Optional[datetime] = None - - if schedule_start: - try: - start_dt = datetime.fromisoformat( - schedule_start.replace("Z", "+00:00") - ) - except (ValueError, TypeError): - pass - - if schedule_end: - try: - end_dt = datetime.fromisoformat( - schedule_end.replace("Z", "+00:00") - ) - except (ValueError, TypeError): - pass - - # Clock-based status refinement - if status != EventStatus.LIVE and start_dt and end_dt: - now_utc = datetime.now(timezone.utc) - _start = start_dt if start_dt.tzinfo else start_dt.replace(tzinfo=timezone.utc) - _end = end_dt if end_dt.tzinfo else end_dt.replace(tzinfo=timezone.utc) - - if now_utc > _end: - status = EventStatus.ENDED - elif _start <= now_utc <= _end: - status = EventStatus.LIVE - - # ---- logo ----------------------------------------------- - logo_url = None - image_refs = relationships.get("images", {}).get("data", []) - if image_refs: - image_key = f"{image_refs[0].get('type')}:{image_refs[0].get('id')}" - image = page_included.get(image_key, {}) - img_attrs = image.get("attributes", {}) - logo_url = img_attrs.get("src") or img_attrs.get("url") - - # ---- edit ID for streaming -------------------------------- - edit_ref = relationships.get("edit", {}).get("data", {}) - edit_id = edit_ref.get("id") if edit_ref else None - - # ---- taxonomy lookups ----------------------------------- - genre = self._resolve_taxonomy( - relationships, "txSports", page_included - ) - competition = self._resolve_taxonomy( - relationships, "txCompetition", page_included - ) - venue = self._resolve_taxonomy( - relationships, "txEvent", page_included - ) - gender_val = self._resolve_taxonomy( - relationships, "txGender", page_included - ) - gender = gender_val if gender_val and gender_val != "." else None - - discipline_val = self._resolve_taxonomy( - relationships, "txDiscipline", page_included - ) - discipline = discipline_val if discipline_val and discipline_val != "." else None - - age_val = self._resolve_taxonomy( - relationships, "txAge", page_included - ) - age_category = age_val if age_val and age_val != "." else None - - master_event = self._resolve_taxonomy( - relationships, "txMaster-sporting-event", page_included - ) - - # ---- primary channel ------------------------------------ - channel_name = None - channel_ref = relationships.get("primaryChannel", {}).get("data") - if channel_ref: - channel_key = f"{channel_ref.get('type')}:{channel_ref.get('id')}" - channel_node = page_included.get(channel_key, {}) - channel_name = channel_node.get("attributes", {}).get("name") - - # ---- language from audio tracks ------------------------- - audio_tracks = attributes.get("audioTracks", []) - language = "de" - if audio_tracks: - if "Deutsch" in audio_tracks: - language = "de" - elif "Englisch" in audio_tracks: - language = "en" - - event = Event( - name=attributes.get("name", "Unknown Event"), - content_id=video_data.get("id", ""), - provider=self.provider_name, - logo_url=logo_url, - mode="live" if attributes.get("videoType") == "LIVE" else "vod", - session_manifest=True, - manifest_script=f"editid={edit_id}" if edit_id else None, - cdm=f"editid={edit_id}" if edit_id else None, - content_type="EVENT", - description=attributes.get("description", ""), - genre=genre, - language=language, - country=self.country.upper(), - start_time=start_dt, - end_time=end_dt, - status=status, - subtitle=attributes.get("secondaryTitle"), - original_name=attributes.get("originalName"), - competition=competition, - venue=venue, - gender=gender, - discipline=discipline, - age_category=age_category, - master_event=master_event, - channel=channel_name, - ) - - events.append(event) - - except Exception as e: - logger.error(f"Error processing collection item: {e}") - continue - - if current_page >= total_pages: - break - current_page += 1 - - return events - - @staticmethod - def _resolve_taxonomy( - relationships: Dict, - key: str, - included: Dict, - ) -> Optional[str]: - """ - Resolve a single taxonomy relationship to its name attribute. - - Args: - relationships: ``relationships`` dict from a video/event item. - key: Relationship key, e.g. ``"txSports"``, ``"txCompetition"``. - included: Per-page included lookup (``type:id`` → item). - - Returns: - The ``attributes.name`` string, or ``None`` if not found. - """ - refs = relationships.get(key, {}).get("data", []) - if not refs: - return None - ref = refs[0] - node = included.get(f"{ref.get('type')}:{ref.get('id')}", {}) - return node.get("attributes", {}).get("name") - - def _extract_distribution_channels( - self, data: Dict - ) -> List[DiscoveryChannel]: - """ - Extract distributionChannel objects from CMS response. - - Args: - data: Parsed JSON response from CMS home route - - Returns: - List of DiscoveryChannel objects (not StreamingChannel!) - """ - channels = [] - - # Build lookup of included items by ID and type - included_by_id = {} - for item in data.get("included", []): - item_id = item.get("id") - item_type = item.get("type") - if item_id and item_type: - key = f"{item_type}:{item_id}" - included_by_id[key] = item - - # Find all distributionChannel objects in included - for item in data.get("included", []): - if item.get("type") == "distributionChannel": - channel = self._create_discovery_channel_from_distribution( - item, included_by_id - ) - if channel: - channels.append(channel) - - return channels - - @staticmethod - def _create_discovery_channel_from_distribution( - distribution_data: Dict, - included_by_id: Dict - ) -> Optional[DiscoveryChannel]: - """ - Create a DiscoveryChannel from distributionChannel object. - - Args: - distribution_data: The distributionChannel object from CMS - included_by_id: Lookup dictionary of all included items - - Returns: - DiscoveryChannel object or None if invalid - """ - try: - # Use the factory method from DiscoveryChannel - discovery_channel = DiscoveryChannel.from_distribution_data( - distribution_data, included_by_id - ) - - logger.debug( - f"Created channel: {discovery_channel.name} " - f"(edit_id: {discovery_channel.edit_id})" - ) - - return discovery_channel - - except Exception as e: - logger.error( - f"Error creating channel from distribution data: {e}" - ) - return None - - def _build_playback_payload(self, edit_id: str) -> Dict[str, Any]: - """ - Build playback payload for the API. - - Args: - edit_id: Content edit ID - - Returns: - Playback request payload - """ - # Generate Google PAL nonce - nonce = secrets.token_bytes(256) - google_pal_nonce = base64.b64encode(nonce).decode('ascii') - - return { - "appBundle": "dplus", - "advertisingInfo": { - "adBlockerDetection": False, - "debug": {}, - "device": {}, - "googlePALNonce": google_pal_nonce, - "server": { - "deviceId": "", - "iabTCFString": "", - "isLimitedAdTracking": 0, - "nielsenAppId": "", - }, - "ssaiProvider": {"version": "2.2.0"}, - }, - "consumptionType": "streaming", - "deviceInfo": get_default_device_info( - self.platform_os, - device_id=self.authenticator.device_id, - ), - "editId": edit_id, - "capabilities": get_default_capabilities(self.platform_os), - "gdpr": False, - "firstPlay": False, - "playbackSessionId": str(uuid.uuid4()), - "applicationSessionId": str(uuid.uuid4()), - "userPreferences": { - "videoQuality": "best", - "uiLanguage": f"{self.country}-DE".upper() - }, - "features": ["mlp"], - } - - def get_playback_info(self, edit_id: str, **kwargs) -> Dict: - """ - Get playback information using edit_id. - - Args: - edit_id: Edit ID for content - **kwargs: Additional parameters - - Returns: - Playback info dictionary - - Raises: - ValueError: If edit_id not provided - PlaybackRestrictedException: If playback restricted - """ - if not edit_id: - raise ValueError("edit_id is required for playback") - - headers = self._get_auth_headers() - payload = self._build_playback_payload(edit_id) - - try: - response = self.http_manager.post( - self.authenticator.playback_endpoint, - operation="playback", - headers=headers, - json=payload, - timeout=30, - ) - - if response.status_code >= 400: - error_text = response.text - logger.error(f"Playback error response: {error_text}") - if "PLAYBACK_RESTRICTED" in error_text: - raise PlaybackRestrictedException( - reason=f"Playback restricted for {edit_id}", - error_code="PLAYBACK_RESTRICTED" - ) - response.raise_for_status() - - response.raise_for_status() - return response.json() - - except PlaybackRestrictedException: - raise - except Exception as e: - raise ManifestFetchError( - f"Error getting playback info for edit_id {edit_id}: {e}" - ) - - def _get_cached_playback_info(self, edit_id: str) -> Dict: - """ - Return playback info for edit_id, using a cache keyed on expiry time. - - Tokens issued by Discovery+ are valid for 24 hours. The cache entry - is kept until the drm_expiration timestamp returned in the response - (minus a 60-second safety margin). If no expiration is present in the - response, a fallback TTL of 23 hours is used. - - Args: - edit_id: Content edit ID - - Returns: - Playback info dictionary (from cache or freshly fetched) - """ - now = time.time() - cached = self._playback_cache.get(edit_id) - - if cached: - expiry, data = cached - if now < expiry: - logger.debug( - f"Playback cache hit for edit_id {edit_id} " - f"(expires in {int(expiry - now)}s)" - ) - return data - else: - logger.debug(f"Playback cache expired for edit_id {edit_id}, re-fetching") - - data = self.get_playback_info(edit_id=edit_id) - - # Determine expiry from drm_expiration field if available - expiry = None - try: - drm_expiration_str = ( - data.get("drm", {}).get("expirationDate") - ) - if drm_expiration_str: - dt = datetime.fromisoformat( - drm_expiration_str.replace("Z", "+00:00") - ) - # Apply a 60-second safety margin - expiry = dt.timestamp() - 60 - except Exception as e: - logger.debug(f"Could not parse drm_expiration: {e}") - - if expiry is None or expiry <= now: - # Fallback: 23 hours from now - expiry = now + 23 * 3600 - logger.debug( - f"Using fallback 23h TTL for playback cache (edit_id: {edit_id})" - ) - - self._playback_cache[edit_id] = (expiry, data) - logger.debug( - f"Playback cache stored for edit_id {edit_id} " - f"(expires in {int(expiry - now)}s)" - ) - return data - - @staticmethod - def extract_streaming_data(playback_data: Dict) -> Dict[str, Any]: - """ - Extract streaming URLs and DRM info from playback response. - - Detects whichever DRM scheme is present in the response — either - 'widevine' (Linux/Chrome) or 'playready' (Windows/Edge) — and - records it as ``drm_system`` so callers can build the correct DRMConfig - without needing to know the active PlatformOS. - - Args: - playback_data: Playback API response - - Returns: - Dictionary with: - manifest_url – DASH manifest URL - license_url – DRM license server URL (or None) - drm_system – 'widevine' | 'playready' | None - drm_auth – JWT auth token extracted from license URL (or None) - streaming_format – 'dash' (always for Discovery+) - drm_expiration – ISO-8601 expiration string (or None) - fallback_manifest – Fallback manifest URL (or None) - """ - result = { - "manifest_url": None, - "license_url": None, - "drm_system": None, - "drm_auth": None, - "streaming_format": "dash", - "drm_expiration": None, - "fallback_manifest": None, - } - - try: - # Extract main manifest - manifest = playback_data.get("manifest", {}) - if manifest: - result["manifest_url"] = manifest.get("url") - result["streaming_format"] = manifest.get("format", "dash") - - # Extract fallback manifest if available - fallback = playback_data.get("fallback", {}) - if fallback: - fallback_manifest = fallback.get("manifest", {}) - if fallback_manifest: - result["fallback_manifest"] = fallback_manifest.get("url") - - # Extract DRM info — check both schemes in priority order. - # The server returns exactly the scheme(s) matching the capabilities - # we advertised in the playback request (widevine for Linux, - # playready for Windows). We probe both so this method stays - # stateless and works regardless of which platform produced the response. - drm = playback_data.get("drm", {}) - if drm: - result["drm_expiration"] = drm.get("expirationDate") - schemes = drm.get("schemes", {}) - - # Preference order: widevine → playready (mirrors Linux default) - scheme_priority = ["widevine", "playready"] - for scheme_name in scheme_priority: - scheme = schemes.get(scheme_name, {}) - if scheme and scheme.get("licenseUrl"): - result["license_url"] = scheme["licenseUrl"] - result["drm_system"] = scheme_name - - # Extract JWT auth token from the license URL if present - if "auth=" in result["license_url"]: - import urllib.parse - parsed = urllib.parse.urlparse(result["license_url"]) - query = urllib.parse.parse_qs(parsed.query) - if "auth" in query: - result["drm_auth"] = query["auth"][0] - break # Stop at first found scheme - - # Extract CDN info - cdn = playback_data.get("cdn", {}) - if cdn: - result["cdn_provider"] = cdn.get("provider") - - except Exception as e: - logger.error(f"Error extracting streaming data: {e}") - - return result - - def _build_drm_config(self, streaming_data: Dict[str, Any]) -> Optional[DRMConfig]: - """ - Build a DRMConfig from extracted streaming data. - - Selects the correct DRMSystem based on the scheme returned by the - server (recorded in ``streaming_data["drm_system"]``). Falls back to - the active ``self.platform_os`` when the scheme is absent so the caller - never has to make this decision themselves. - - Supported schemes: - 'widevine' → DRMSystem.WIDEVINE (Linux/Chrome path) - 'playready' → DRMSystem.PLAYREADY (Windows/Edge path) - - Args: - streaming_data: Dictionary returned by extract_streaming_data() - - Returns: - DRMConfig instance or None if no license URL is available - """ - license_url = streaming_data.get("license_url") - if not license_url: - return None - - # Resolve DRM system from what the server actually returned, - # falling back to platform_os expectation if not present. - drm_system_str = streaming_data.get("drm_system") - if drm_system_str == "playready": - drm_system = DRMSystem.PLAYREADY - elif drm_system_str == "widevine": - drm_system = DRMSystem.WIDEVINE - else: - # Fallback: derive from active platform - drm_system = ( - DRMSystem.PLAYREADY - if self.platform_os == PlatformOS.WINDOWS - else DRMSystem.WIDEVINE - ) - logger.debug( - f"drm_system not in streaming_data, inferred from platform_os " - f"({self.platform_os.value}): {drm_system.name}" - ) - - license_headers = get_drm_request_headers(self.platform_os) - - return DRMConfig( - system=drm_system, - priority=1, - license=LicenseConfig( - server_url=license_url, - req_headers=json.dumps(license_headers), - req_data="{CHA-RAW}", - use_http_get_request=False, - ), ) def populate_streaming_data( @@ -1595,132 +401,35 @@ class DiscoveryProvider(StreamingProvider): channels: List[StreamingChannel], max_retries: int = 3, ) -> List[StreamingChannel]: - """ - Populate streaming data for channels using cached DiscoveryChannel data. - - Args: - channels: List of StreamingChannel objects to populate - max_retries: Maximum retry attempts per channel - - Returns: - List of successfully populated channels - """ - 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: - # Get DiscoveryChannel from cache - disco_channel = self._channels_cache.get(channel.channel_id) - - if not disco_channel: - logger.warning( - f"Channel {channel.name} not in cache, skipping" - ) - break - - # Get edit_id from cached DiscoveryChannel - edit_id = disco_channel.edit_id - - if not edit_id: - logger.warning( - f"No edit_id for channel {channel.name}, skipping" - ) - break - - logger.debug( - f"Getting playback info for: {channel.name} " - f"(edit_id: {edit_id}, attempt {retries + 1})" - ) - - playback_data = self._get_cached_playback_info(edit_id=edit_id) - streaming_data = self.extract_streaming_data(playback_data) - - if streaming_data["manifest_url"]: - channel.manifest = streaming_data["manifest_url"] - channel.streaming_format = streaming_data["streaming_format"] - - if streaming_data["license_url"]: - drm_config = self._build_drm_config(streaming_data) - if drm_config: - channel.drm_config = drm_config - channel.license_url = streaming_data["license_url"] - channel.cdm_type = streaming_data["drm_system"] - - # Store DRM token in DiscoveryChannel cache - if streaming_data["drm_auth"]: - disco_channel.raw_data["drm_auth"] = ( - streaming_data["drm_auth"] - ) - - logger.info( - f"Streaming data populated for: {channel.name}" - ) - successful_channels.append(channel) - success = True - else: - raise ManifestFetchError("No manifest URL in response") - - except PlaybackRestrictedException 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 " - f"{channel.name}: {e}" - ) - time.sleep(1) - else: - logger.error( - f"Failed to get streaming data for " - f"{channel.name}: {e}" - ) - - logger.info( - f"Streaming data population complete: " - f"{len(successful_channels)}/{len(channels)}" + """Populate streaming data for a list of channels.""" + return self.playback_manager.populate_streaming_data( + channels, max_retries=max_retries ) - return successful_channels def get_manifest(self, content_id: str, **kwargs) -> Optional[str]: """ - Get manifest URL for a specific channel by ID. - - Args: - content_id: Channel identifier - **kwargs: Additional parameters - - Returns: - Manifest URL or None + Get manifest URL for a channel by ID. Raises: - ChannelNotFoundError: If channel not in cache + ChannelNotFoundError: If channel not in cache. """ try: - # Get DiscoveryChannel from cache disco_channel = self._channels_cache.get(content_id) - if not disco_channel: raise ChannelNotFoundError(content_id) edit_id = disco_channel.edit_id - if not edit_id: raise ManifestFetchError( f"No edit_id for channel {content_id}" ) - playback_data = self._get_cached_playback_info(edit_id=edit_id) - streaming_data = self.extract_streaming_data(playback_data) + playback_data = self.playback_manager.get_cached_playback_info( + edit_id=edit_id + ) + streaming_data = self.playback_manager.extract_streaming_data( + playback_data + ) return streaming_data.get("manifest_url") except (ChannelNotFoundError, ManifestFetchError): @@ -1732,20 +441,9 @@ class DiscoveryProvider(StreamingProvider): return None def get_drm(self, content_id: str, **kwargs) -> List[DRMConfig]: - """ - Get all DRM configurations for a channel by ID. - - Args: - content_id: Channel identifier - **kwargs: Additional parameters - - Returns: - List of DRM configurations - """ + """Get all DRM configurations for a channel by ID.""" try: - # Get DiscoveryChannel from cache disco_channel = self._channels_cache.get(content_id) - if not disco_channel: logger.warning( f"Channel {content_id} not in cache for DRM" @@ -1753,17 +451,22 @@ class DiscoveryProvider(StreamingProvider): return [] edit_id = disco_channel.edit_id - if not edit_id: return [] - playback_data = self._get_cached_playback_info(edit_id=edit_id) - streaming_data = self.extract_streaming_data(playback_data) + playback_data = self.playback_manager.get_cached_playback_info( + edit_id=edit_id + ) + streaming_data = self.playback_manager.extract_streaming_data( + playback_data + ) if not streaming_data["license_url"]: return [] - drm_config = self._build_drm_config(streaming_data) + drm_config = self.playback_manager.build_drm_config( + streaming_data + ) return [drm_config] if drm_config else [] except Exception as e: @@ -1779,20 +482,8 @@ class DiscoveryProvider(StreamingProvider): end_time: Optional[datetime] = None, **kwargs, ) -> List[Dict]: - """ - Get EPG data for a channel (not yet implemented). - - Args: - channel_id: Channel identifier - start_time: Start time for EPG data - end_time: End time for EPG data - **kwargs: Additional parameters - - Returns: - Empty list (not implemented) - """ + """Get EPG data for a channel (not yet implemented).""" logger.info( - f"EPG data requested for channel {channel_id} - " - "not yet implemented" + f"EPG data requested for channel {channel_id} — not yet implemented" ) return [] \ No newline at end of file